fix(scada): use project-scoped metadata

This commit is contained in:
2026-07-17 16:28:40 +08:00
parent a204980944
commit b4ecfbb87a
11 changed files with 694 additions and 290 deletions
+23 -7
View File
@@ -12,12 +12,19 @@ def _load_burst_location_module():
Path(__file__).resolve().parents[2] / "app" / "services" / "burst_location.py"
)
missing = object()
previous_modules = {}
def install_module(name: str, module: types.ModuleType) -> None:
previous_modules.setdefault(name, sys.modules.get(name, missing))
sys.modules[name] = module
def ensure_package(name: str) -> types.ModuleType:
module = sys.modules.get(name)
if module is None:
module = types.ModuleType(name)
module.__path__ = []
sys.modules[name] = module
install_module(name, module)
return module
for package_name in [
@@ -46,11 +53,11 @@ def _load_burst_location_module():
)
)
time_api_module.utc_now = lambda: datetime.now(timezone.utc)
sys.modules["app.services.time_api"] = time_api_module
install_module("app.services.time_api", time_api_module)
algorithms_module = types.ModuleType("app.algorithms.burst_location")
algorithms_module.run_burst_location = lambda **kwargs: {}
sys.modules["app.algorithms.burst_location"] = algorithms_module
install_module("app.algorithms.burst_location", algorithms_module)
internal_queries_module = types.ModuleType(
"app.infra.db.timescaledb.internal_queries"
@@ -70,7 +77,9 @@ def _load_burst_location_module():
return {}
internal_queries_module.InternalQueries = DummyInternalQueries
sys.modules["app.infra.db.timescaledb.internal_queries"] = internal_queries_module
install_module(
"app.infra.db.timescaledb.internal_queries", internal_queries_module
)
scheme_management_module = types.ModuleType("app.services.scheme_management")
scheme_management_module.query_burst_location_scheme_detail = lambda *args, **kwargs: {}
@@ -78,18 +87,25 @@ def _load_burst_location_module():
scheme_management_module.query_scheme_list = lambda *args, **kwargs: []
scheme_management_module.scheme_name_exists = lambda *args, **kwargs: False
scheme_management_module.store_scheme_info = lambda *args, **kwargs: None
sys.modules["app.services.scheme_management"] = scheme_management_module
install_module("app.services.scheme_management", scheme_management_module)
tjnetwork_module = types.ModuleType("app.services.tjnetwork")
tjnetwork_module.dump_inp = lambda *args, **kwargs: None
tjnetwork_module.get_all_scada_info = lambda *args, **kwargs: []
sys.modules["app.services.tjnetwork"] = tjnetwork_module
install_module("app.services.tjnetwork", tjnetwork_module)
module_name = "tests_burst_location_under_test"
spec = importlib.util.spec_from_file_location(module_name, module_path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
try:
spec.loader.exec_module(module)
finally:
for name, previous in reversed(previous_modules.items()):
if previous is missing:
sys.modules.pop(name, None)
else:
sys.modules[name] = previous
return module