122 lines
3.4 KiB
Python
122 lines
3.4 KiB
Python
import asyncio
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock
|
|
|
|
from app.api.v1.endpoints import project_data
|
|
from app.infra.db.timescaledb import composite_queries
|
|
|
|
|
|
PROJECT_SCADA = {
|
|
"id": "fengyang-pressure-1",
|
|
"type": "pressure",
|
|
"associated_element_id": "J1",
|
|
"api_query_id": "query-1",
|
|
"transmission_mode": "realtime",
|
|
"transmission_frequency": None,
|
|
"reliability": 1.0,
|
|
"x": 117.1,
|
|
"y": 32.9,
|
|
}
|
|
START_TIME = datetime(2026, 6, 1, tzinfo=timezone.utc)
|
|
END_TIME = datetime(2026, 6, 2, tzinfo=timezone.utc)
|
|
|
|
|
|
def _patch_project_scadas(monkeypatch):
|
|
monkeypatch.setattr(
|
|
composite_queries.ScadaInfoRepository,
|
|
"get_scadas",
|
|
AsyncMock(return_value=[PROJECT_SCADA.copy()]),
|
|
)
|
|
|
|
|
|
def test_realtime_scada_simulation_uses_current_project_metadata(monkeypatch):
|
|
_patch_project_scadas(monkeypatch)
|
|
query_mock = AsyncMock(return_value=[{"time": START_TIME, "value": 26.5}])
|
|
monkeypatch.setattr(
|
|
composite_queries.RealtimeRepository,
|
|
"get_node_field_by_time_range",
|
|
query_mock,
|
|
)
|
|
|
|
result = asyncio.run(
|
|
composite_queries.CompositeQueries.get_scada_associated_realtime_simulation_data(
|
|
object(),
|
|
object(),
|
|
[PROJECT_SCADA["id"]],
|
|
START_TIME,
|
|
END_TIME,
|
|
)
|
|
)
|
|
|
|
assert result[PROJECT_SCADA["id"]][0]["scada_id"] == PROJECT_SCADA["id"]
|
|
assert query_mock.await_count == 1
|
|
assert query_mock.await_args.args[1:] == (
|
|
START_TIME,
|
|
END_TIME,
|
|
"J1",
|
|
"pressure",
|
|
)
|
|
|
|
|
|
def test_scheme_scada_simulation_uses_current_project_metadata(monkeypatch):
|
|
_patch_project_scadas(monkeypatch)
|
|
query_mock = AsyncMock(return_value=[{"time": START_TIME, "value": 26.5}])
|
|
monkeypatch.setattr(
|
|
composite_queries.SchemeRepository,
|
|
"get_node_field_by_scheme_and_time_range",
|
|
query_mock,
|
|
)
|
|
|
|
result = asyncio.run(
|
|
composite_queries.CompositeQueries.get_scada_associated_scheme_simulation_data(
|
|
object(),
|
|
object(),
|
|
[PROJECT_SCADA["id"]],
|
|
START_TIME,
|
|
END_TIME,
|
|
"baseline",
|
|
"scheme-1",
|
|
)
|
|
)
|
|
|
|
assert result[PROJECT_SCADA["id"]][0]["scada_id"] == PROJECT_SCADA["id"]
|
|
assert query_mock.await_args.args[5:] == ("J1", "pressure")
|
|
|
|
|
|
def test_element_scada_query_uses_current_project_metadata(monkeypatch):
|
|
_patch_project_scadas(monkeypatch)
|
|
query_mock = AsyncMock(
|
|
return_value={
|
|
PROJECT_SCADA["id"]: [{"time": START_TIME, "value": 26.5}]
|
|
}
|
|
)
|
|
monkeypatch.setattr(
|
|
composite_queries.ScadaRepository,
|
|
"get_scada_field_by_id_time_range",
|
|
query_mock,
|
|
)
|
|
|
|
result = asyncio.run(
|
|
composite_queries.CompositeQueries.get_element_associated_scada_data(
|
|
object(),
|
|
object(),
|
|
"J1",
|
|
START_TIME,
|
|
END_TIME,
|
|
)
|
|
)
|
|
|
|
assert result == {"J1": [{"time": START_TIME, "value": 26.5}]}
|
|
|
|
|
|
def test_scada_info_endpoint_uses_current_project_connection(monkeypatch):
|
|
monkeypatch.setattr(
|
|
project_data.ScadaInfoRepository,
|
|
"get_scadas",
|
|
AsyncMock(return_value=[PROJECT_SCADA.copy()]),
|
|
)
|
|
|
|
result = asyncio.run(project_data.get_scada_info_with_connection(object()))
|
|
|
|
assert result == {"success": True, "data": [PROJECT_SCADA], "count": 1}
|