fix: game plugin cache + test mode NullPointer crash on shutdown

- game_manager: add _scanned cache, _ensure_scanned() avoids re-scanning on every API call
- test_mode: fix AttributeError when stop_test_mode called before start
- listener: safe stop_test_mode with null check
This commit is contained in:
2026-07-26 18:19:03 +08:00
parent 90ab5f9c0b
commit 3b7edc4f14
2 changed files with 18 additions and 11 deletions
+8 -4
View File
@@ -68,11 +68,11 @@ class GamePluginManager:
USER_DIR.mkdir(parents=True, exist_ok=True) USER_DIR.mkdir(parents=True, exist_ok=True)
self._plugins: dict[str, GamePlugin] = {} self._plugins: dict[str, GamePlugin] = {}
self._parser_cache: dict[str, Any] = {} self._parser_cache: dict[str, Any] = {}
self._scanned = False
self._discover() self._discover()
def _discover(self): def _discover(self):
self._plugins.clear() self._plugins.clear()
for d in [BUILTIN_DIR, USER_DIR]: for d in [BUILTIN_DIR, USER_DIR]:
if not d.exists(): if not d.exists():
continue continue
@@ -84,14 +84,18 @@ class GamePluginManager:
gp = GamePlugin.from_manifest(manifest, is_builtin) gp = GamePlugin.from_manifest(manifest, is_builtin)
if gp: if gp:
self._plugins[gp.id] = gp self._plugins[gp.id] = gp
logger.debug("Discovered game plugin: %s", gp.id) self._scanned = True
def _ensure_scanned(self):
if not self._scanned:
self._discover()
def list_all(self) -> list[dict[str, Any]]: def list_all(self) -> list[dict[str, Any]]:
self._discover() self._ensure_scanned()
return [p.to_dict() for p in self._plugins.values()] return [p.to_dict() for p in self._plugins.values()]
def get(self, plugin_id: str) -> GamePlugin | None: def get(self, plugin_id: str) -> GamePlugin | None:
self._discover() self._ensure_scanned()
return self._plugins.get(plugin_id) return self._plugins.get(plugin_id)
def get_parser(self, plugin_id: str) -> Any | None: def get_parser(self, plugin_id: str) -> Any | None:
+10 -7
View File
@@ -123,17 +123,20 @@ class TelemetryListener:
@property @property
def is_testing(self) -> bool: def is_testing(self) -> bool:
from server.test_mode import _test_generator from server.test_mode import get_generator
return _test_generator.running gen = get_generator()
return gen is not None and gen.running
def start_test_mode(self): def start_test_mode(self):
from server.test_mode import _test_generator from server.test_mode import get_generator
_test_generator.listener = self gen = get_generator(self)
_test_generator.start() gen.start()
def stop_test_mode(self): def stop_test_mode(self):
from server.test_mode import _test_generator from server.test_mode import get_generator
_test_generator.stop() gen = get_generator()
if gen is not None and gen.running:
gen.stop()
def _get_parser(self, key: str) -> BaseParser: def _get_parser(self, key: str) -> BaseParser:
if key not in self._parser_cache: if key not in self._parser_cache: