112 lines
4.0 KiB
Python
112 lines
4.0 KiB
Python
import json
|
|
from contextlib import contextmanager
|
|
|
|
from tests.conftest import install_stub, load_module_from_path
|
|
|
|
|
|
def _load_scenarios_module(monkeypatch):
|
|
install_stub(monkeypatch, "app.services", package=True)
|
|
install_stub(monkeypatch, "app.algorithms", package=True)
|
|
install_stub(monkeypatch, "app.algorithms.simulation", package=True)
|
|
install_stub(monkeypatch, "app.services.simulation", {})
|
|
install_stub(
|
|
monkeypatch,
|
|
"app.algorithms.simulation.runner",
|
|
{
|
|
"run_simulation_ex": lambda *args, **kwargs: json.dumps(
|
|
{"output": {"node_results": [], "link_results": []}}
|
|
),
|
|
"from_clock_to_seconds_2": lambda value: value,
|
|
},
|
|
)
|
|
install_stub(monkeypatch, "app.services.scheme_management", {"store_scheme_info": lambda *args, **kwargs: None})
|
|
install_stub(
|
|
monkeypatch,
|
|
"app.services.tjnetwork",
|
|
{
|
|
"ChangeSet": type("ChangeSet", (), {}),
|
|
"OPTION_DEMAND_MODEL_PDA": "OPTION_DEMAND_MODEL_PDA",
|
|
"OPTION_QUALITY_CHEMICAL": "OPTION_QUALITY_CHEMICAL",
|
|
"SOURCE_TYPE_SETPOINT": "SOURCE_TYPE_SETPOINT",
|
|
"add_pattern": lambda *args, **kwargs: None,
|
|
"add_source": lambda *args, **kwargs: None,
|
|
"copy_project": lambda *args, **kwargs: None,
|
|
"delete_project": lambda *args, **kwargs: None,
|
|
"get_demand": lambda *args, **kwargs: None,
|
|
"get_emitter": lambda *args, **kwargs: None,
|
|
"get_node_links": lambda *args, **kwargs: None,
|
|
"get_option": lambda *args, **kwargs: None,
|
|
"get_pattern": lambda *args, **kwargs: None,
|
|
"get_pipe": lambda *args, **kwargs: None,
|
|
"get_source": lambda *args, **kwargs: None,
|
|
"get_time": lambda *args, **kwargs: None,
|
|
"have_project": lambda *args, **kwargs: False,
|
|
"is_junction": lambda *args, **kwargs: False,
|
|
"set_demand": lambda *args, **kwargs: None,
|
|
"set_emitter": lambda *args, **kwargs: None,
|
|
"set_option": lambda *args, **kwargs: None,
|
|
"set_source": lambda *args, **kwargs: None,
|
|
"set_time": lambda *args, **kwargs: None,
|
|
},
|
|
)
|
|
return load_module_from_path(
|
|
"tests_age_analysis_scenarios_module",
|
|
"app/algorithms/simulation/scenarios.py",
|
|
)
|
|
|
|
|
|
def test_age_analysis_passes_duration_by_keyword(monkeypatch):
|
|
module = _load_scenarios_module(monkeypatch)
|
|
captured = {}
|
|
|
|
@contextmanager
|
|
def fake_temporary_project(project, purpose):
|
|
yield f"{purpose}_{project}_run"
|
|
|
|
monkeypatch.setattr(module, "temporary_project_database", fake_temporary_project)
|
|
|
|
def fake_run_simulation_ex(*args, **kwargs):
|
|
captured["args"] = args
|
|
captured["kwargs"] = kwargs
|
|
return json.dumps({"output": {"node_results": [], "link_results": []}})
|
|
|
|
monkeypatch.setattr(module, "run_simulation_ex", fake_run_simulation_ex)
|
|
|
|
module.age_analysis("demo", "2026-06-03T07:00:00+08:00", 300)
|
|
|
|
assert captured["args"] == (
|
|
"age_analysis_demo_run",
|
|
"realtime",
|
|
"2026-06-03T07:00:00+08:00",
|
|
)
|
|
assert captured["kwargs"] == {
|
|
"duration": 300,
|
|
"downloading_prohibition": True,
|
|
}
|
|
|
|
|
|
def test_isolated_analysis_cleans_database_after_early_return(monkeypatch):
|
|
module = _load_scenarios_module(monkeypatch)
|
|
lifecycle: list[tuple[str, str]] = []
|
|
|
|
@contextmanager
|
|
def fake_temporary_project(project, purpose):
|
|
lifecycle.append(("create", purpose))
|
|
try:
|
|
yield "isolated_run"
|
|
finally:
|
|
lifecycle.append(("delete", purpose))
|
|
|
|
monkeypatch.setattr(
|
|
module, "temporary_project_database", fake_temporary_project
|
|
)
|
|
|
|
@module._isolated_analysis("probe")
|
|
def return_early(name, *, _temporary_project=None):
|
|
assert name == "demo"
|
|
assert _temporary_project == "isolated_run"
|
|
return "done"
|
|
|
|
assert return_early("demo") == "done"
|
|
assert lifecycle == [("create", "probe"), ("delete", "probe")]
|