124 lines
3.5 KiB
Python
124 lines
3.5 KiB
Python
import importlib.util
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from app.services.time_api import parse_utc_time
|
|
|
|
|
|
def _load_scheme_repository():
|
|
module_path = (
|
|
Path(__file__).resolve().parents[2]
|
|
/ "app"
|
|
/ "infra"
|
|
/ "db"
|
|
/ "timescaledb"
|
|
/ "repositories"
|
|
/ "scheme.py"
|
|
)
|
|
spec = importlib.util.spec_from_file_location(
|
|
"customer_scheme_repository_for_timestep_test", module_path
|
|
)
|
|
module = importlib.util.module_from_spec(spec)
|
|
assert spec and spec.loader
|
|
spec.loader.exec_module(module)
|
|
return module.SchemeRepository
|
|
|
|
|
|
SchemeRepository = _load_scheme_repository()
|
|
|
|
|
|
def _node_result(periods: int) -> list[dict]:
|
|
return [
|
|
{
|
|
"node": "J1",
|
|
"result": [
|
|
{"demand": index, "head": index, "pressure": index, "quality": index}
|
|
for index in range(periods)
|
|
],
|
|
}
|
|
]
|
|
|
|
|
|
def _link_result(periods: int) -> list[dict]:
|
|
return [
|
|
{
|
|
"link": "P1",
|
|
"result": [
|
|
{
|
|
"flow": index,
|
|
"friction": index,
|
|
"headloss": index,
|
|
"quality": index,
|
|
"reaction": index,
|
|
"setting": index,
|
|
"status": index,
|
|
"velocity": index,
|
|
}
|
|
for index in range(periods)
|
|
],
|
|
}
|
|
]
|
|
|
|
|
|
def test_store_scheme_simulation_uses_15_minute_report_step(monkeypatch):
|
|
inserted: dict[str, list[dict]] = {}
|
|
|
|
monkeypatch.setattr(
|
|
SchemeRepository,
|
|
"insert_nodes_batch_sync",
|
|
staticmethod(lambda conn, data: inserted.setdefault("nodes", data)),
|
|
)
|
|
monkeypatch.setattr(
|
|
SchemeRepository,
|
|
"insert_links_batch_sync",
|
|
staticmethod(lambda conn, data: inserted.setdefault("links", data)),
|
|
)
|
|
|
|
SchemeRepository.store_scheme_simulation_result_sync(
|
|
conn=object(),
|
|
scheme_type="burst_analysis",
|
|
scheme_name="five_hour_case",
|
|
node_result_list=_node_result(21),
|
|
link_result_list=_link_result(21),
|
|
result_start_time="2026-07-16T00:00:00Z",
|
|
num_periods=21,
|
|
result_timestep_seconds=900,
|
|
)
|
|
|
|
start_time = parse_utc_time("2026-07-16T00:00:00Z")
|
|
assert len(inserted["nodes"]) == 21
|
|
assert inserted["nodes"][0]["time"] == start_time
|
|
assert inserted["nodes"][-1]["time"] == start_time + timedelta(hours=5)
|
|
assert inserted["links"][-1]["time"] == start_time + timedelta(hours=5)
|
|
|
|
|
|
def test_store_scheme_simulation_uses_hourly_report_step(monkeypatch):
|
|
inserted: dict[str, list[dict]] = {}
|
|
|
|
monkeypatch.setattr(
|
|
SchemeRepository,
|
|
"insert_nodes_batch_sync",
|
|
staticmethod(lambda conn, data: inserted.setdefault("nodes", data)),
|
|
)
|
|
monkeypatch.setattr(
|
|
SchemeRepository,
|
|
"insert_links_batch_sync",
|
|
staticmethod(lambda conn, data: inserted.setdefault("links", data)),
|
|
)
|
|
|
|
SchemeRepository.store_scheme_simulation_result_sync(
|
|
conn=object(),
|
|
scheme_type="burst_analysis",
|
|
scheme_name="hourly_case",
|
|
node_result_list=_node_result(6),
|
|
link_result_list=_link_result(6),
|
|
result_start_time="2026-07-16T00:00:00Z",
|
|
num_periods=6,
|
|
result_timestep_seconds=3600,
|
|
)
|
|
|
|
start_time = parse_utc_time("2026-07-16T00:00:00Z")
|
|
assert [item["time"] for item in inserted["nodes"]] == [
|
|
start_time + timedelta(hours=index) for index in range(6)
|
|
]
|