147 lines
4.5 KiB
Python
147 lines
4.5 KiB
Python
import asyncio
|
|
from datetime import UTC, datetime, timedelta
|
|
from unittest.mock import AsyncMock
|
|
|
|
import pytest
|
|
|
|
from app.domain.schemas.timeseries_history import ElementHistoryQuery
|
|
from app.services import timeseries_history
|
|
|
|
|
|
START = datetime(2026, 9, 1, tzinfo=UTC)
|
|
END = START + timedelta(hours=1)
|
|
|
|
|
|
def test_realtime_history_batches_devices_and_converts_units(monkeypatch):
|
|
monkeypatch.setattr(
|
|
timeseries_history.ScadaInfoRepository,
|
|
"get_scadas_for_elements",
|
|
AsyncMock(
|
|
return_value=[
|
|
{
|
|
"device_id": "F-1",
|
|
"device_type": "pipe_flow",
|
|
"link_id": "P-1",
|
|
"node_id": None,
|
|
"measurement_unit": "m3/h",
|
|
},
|
|
{
|
|
"device_id": "P-1A",
|
|
"device_type": "pressure",
|
|
"link_id": None,
|
|
"node_id": "J-1",
|
|
"measurement_unit": "m",
|
|
},
|
|
{
|
|
"device_id": "P-1B",
|
|
"device_type": "pressure",
|
|
"link_id": None,
|
|
"node_id": "J-1",
|
|
"measurement_unit": "m",
|
|
},
|
|
]
|
|
),
|
|
)
|
|
scada_query = AsyncMock(
|
|
return_value=[
|
|
{
|
|
"time": START,
|
|
"device_id": "F-1",
|
|
"monitored_value": 36.0,
|
|
"cleaned_value": 35.0,
|
|
},
|
|
{
|
|
"time": START,
|
|
"device_id": "P-1A",
|
|
"monitored_value": 20.0,
|
|
"cleaned_value": 21.0,
|
|
},
|
|
{
|
|
"time": START,
|
|
"device_id": "P-1B",
|
|
"monitored_value": 22.0,
|
|
"cleaned_value": 23.0,
|
|
},
|
|
]
|
|
)
|
|
monkeypatch.setattr(
|
|
timeseries_history.ScadaRepository,
|
|
"get_scada_by_ids_time_range",
|
|
scada_query,
|
|
)
|
|
monkeypatch.setattr(
|
|
timeseries_history.NetworkSettingsRepository,
|
|
"get_result_units",
|
|
AsyncMock(return_value={"flow": "MLD", "pressure": "METERS"}),
|
|
)
|
|
monkeypatch.setattr(
|
|
timeseries_history.RealtimeRepository,
|
|
"get_link_fields_by_ids_time_range",
|
|
AsyncMock(return_value={"P-1": [{"time": START, "value": 2.0}]}),
|
|
)
|
|
monkeypatch.setattr(
|
|
timeseries_history.RealtimeRepository,
|
|
"get_node_fields_by_ids_time_range",
|
|
AsyncMock(return_value={"J-1": [{"time": START, "value": 24.0}]}),
|
|
)
|
|
|
|
result = asyncio.run(
|
|
timeseries_history.TimeseriesHistoryService.query(
|
|
object(),
|
|
object(),
|
|
ElementHistoryQuery(
|
|
start_time=START,
|
|
end_time=END,
|
|
mode="realtime_comparison",
|
|
elements=[
|
|
{"element_id": "P-1", "element_type": "pipe"},
|
|
{"element_id": "J-1", "element_type": "junction"},
|
|
],
|
|
),
|
|
)
|
|
)
|
|
|
|
assert scada_query.await_count == 1
|
|
assert set(scada_query.await_args.args[1]) == {"F-1", "P-1A", "P-1B"}
|
|
assert {item.device_id for item in result.series if item.device_id} == {
|
|
"F-1",
|
|
"P-1A",
|
|
"P-1B",
|
|
}
|
|
simulation = {
|
|
(item.element_id, item.metric.value): item
|
|
for item in result.series
|
|
if item.source.value == "realtime_simulation"
|
|
}
|
|
assert simulation[("P-1", "flow")].points[0].value == 2.0
|
|
assert simulation[("P-1", "flow")].source_unit == "MLD"
|
|
assert simulation[("J-1", "pressure")].points[0].value == 24.0
|
|
assert all(item.display_unit in {"m³/h", "m"} for item in result.series)
|
|
|
|
|
|
def test_requested_device_must_belong_to_element(monkeypatch):
|
|
monkeypatch.setattr(
|
|
timeseries_history.ScadaInfoRepository,
|
|
"get_scadas_for_elements",
|
|
AsyncMock(return_value=[]),
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="do not belong"):
|
|
asyncio.run(
|
|
timeseries_history.TimeseriesHistoryService.query(
|
|
object(),
|
|
object(),
|
|
ElementHistoryQuery(
|
|
start_time=START,
|
|
end_time=END,
|
|
elements=[
|
|
{
|
|
"element_id": "J-1",
|
|
"element_type": "junction",
|
|
"device_ids": ["missing"],
|
|
}
|
|
],
|
|
),
|
|
)
|
|
)
|