OrgTJWater/TJWaterServerBinary
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

后端统一时区为 UTC

Browse Source
This commit is contained in:
Jiang
2026-04-14 14:46:51 +08:00
parent 51b481d174
commit bf2aaa5ff7
16 changed files with 263 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/unit/test_burst_location_service.py
+32 -14
View File
@@ -1,7 +1,7 @@
import importlib.util
import sys
import types
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
from pathlib import Path
import pytest
@@ -30,6 +30,24 @@ def _load_burst_location_module():
]:
ensure_package(package_name)
time_api_module = types.ModuleType("app.services.time_api")
time_api_module.parse_utc_time = (
lambda value, field_name="datetime": (
value.astimezone(timezone.utc)
if isinstance(value, datetime) and value.tzinfo is not None
else datetime.fromisoformat(value).astimezone(timezone.utc)
)
)
time_api_module.extract_date = (
lambda value, field_name="date": (
value.date()
if isinstance(value, datetime)
else datetime.fromisoformat(value).date()
)
)
time_api_module.utc_now = lambda: datetime.now(timezone.utc)
sys.modules["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
@@ -125,16 +143,16 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
def fake_scheme_query(**kwargs):
scheme_calls.append(kwargs)
start_hour = datetime.fromisoformat(kwargs["start_time"]).astimezone(
timezone(timedelta(hours=8))
).hour
if kwargs["element_type"] == "node" and kwargs["field"] == "pressure":
start_hour = datetime.fromisoformat(kwargs["start_time"]).hour
values = [12.0, 14.0, 16.0, 18.0] if start_hour == 8 else [8.0, 10.0, 12.0, 14.0]
return {"J1": _build_series(kwargs["start_time"], values)}
if kwargs["element_type"] == "link" and kwargs["field"] == "flow":
start_hour = datetime.fromisoformat(kwargs["start_time"]).hour
values = [5.0, 7.0, 9.0, 11.0] if start_hour == 8 else [2.0, 4.0, 6.0, 8.0]
return {"P1": _build_series(kwargs["start_time"], values)}
if kwargs["element_type"] == "node" and kwargs["field"] == "actual_demand":
start_hour = datetime.fromisoformat(kwargs["start_time"]).hour
values = [3.0, 5.0, 7.0, 9.0] if start_hour == 8 else [1.0, 3.0, 5.0, 7.0]
return {"J2": _build_series(kwargs["start_time"], values)}
raise AssertionError(f"Unexpected scheme query: {kwargs}")
@@ -167,8 +185,8 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
simulation_scheme_name="BurstSchemeA",
simulation_scheme_type="burst_analysis",
burst_leakage=10.0,
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0),
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=8))),
use_scada_flow=True,
)
@@ -192,14 +210,14 @@ def test_run_burst_location_uses_single_timerange_with_burst_source_split(monkey
assert any(call["element_type"] == "link" and call["field"] == "flow" for call in scheme_calls)
assert any(call["element_type"] == "node" and call["field"] == "actual_demand" for call in scheme_calls)
assert len(realtime_calls) == 3
assert all(datetime.fromisoformat(call["start_time"]).hour == 8 for call in realtime_calls)
assert all(datetime.fromisoformat(call["end_time"]).hour == 9 for call in realtime_calls)
assert all(datetime.fromisoformat(call["start_time"]).hour == 0 for call in realtime_calls)
assert all(datetime.fromisoformat(call["end_time"]).hour == 1 for call in realtime_calls)
assert any(call["element_type"] == "node" and call["field"] == "pressure" for call in realtime_calls)
assert any(call["element_type"] == "link" and call["field"] == "flow" for call in realtime_calls)
assert any(call["element_type"] == "node" and call["field"] == "actual_demand" for call in realtime_calls)
assert result["scada_window"] == {
"burst_start": "2025-01-01T08:00:00",
"burst_end": "2025-01-01T09:00:00",
"burst_start": "2025-01-01T00:00:00+00:00",
"burst_end": "2025-01-01T01:00:00+00:00",
}
@@ -225,8 +243,8 @@ def test_run_burst_location_requires_simulation_scheme_name(monkeypatch, tmp_pat
username="testuser",
data_source="simulation",
burst_leakage=1.0,
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0),
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=8))),
)
@@ -290,8 +308,8 @@ def test_run_burst_location_monitoring_uses_scada_for_burst_and_realtime_for_nor
username="testuser",
data_source="monitoring",
burst_leakage=1.0,
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0),
scada_burst_start=datetime(2025, 1, 1, 8, 0, 0, tzinfo=timezone(timedelta(hours=8))),
scada_burst_end=datetime(2025, 1, 1, 9, 0, 0, tzinfo=timezone(timedelta(hours=8))),
)
assert result["observed_source"] == "scada_burst_realtime_normal_timerange"
tests/unit/test_time_api.py
+45
View File
@@ -0,0 +1,45 @@
import importlib.util
from datetime import date, datetime, timedelta, timezone
from pathlib import Path
import pytest
def _load_time_api_module():
module_path = (
Path(__file__).resolve().parents[2] / "app" / "services" / "time_api.py"
)
spec = importlib.util.spec_from_file_location("tests_time_api_under_test", module_path)
module = importlib.util.module_from_spec(spec)
assert spec and spec.loader
spec.loader.exec_module(module)
return module
def test_parse_utc_time_rejects_naive_datetimes():
module = _load_time_api_module()
with pytest.raises(ValueError, match="timezone information"):
module.parse_utc_time("2025-01-01T08:00:00")
def test_parse_utc_time_normalizes_offset_datetime_to_utc():
module = _load_time_api_module()
result = module.parse_utc_time("2025-01-01T08:00:00+08:00")
assert result == datetime(2025, 1, 1, 0, 0, tzinfo=timezone.utc)
def test_extract_date_keeps_original_offset_calendar_day():
module = _load_time_api_module()
result = module.extract_date("2025-01-01T00:30:00+08:00")
assert result == date(2025, 1, 1)
def test_utc_now_returns_timezone_aware_utc_datetime():
module = _load_time_api_module()
result = module.utc_now()
assert result.tzinfo == timezone.utc
assert result.utcoffset() == timedelta(0)