54 lines
1.6 KiB
Python
54 lines
1.6 KiB
Python
import asyncio
|
|
|
|
|
|
class _FakeAcquire:
|
|
def __init__(self, conn):
|
|
self._conn = conn
|
|
|
|
async def __aenter__(self):
|
|
return self._conn
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
return False
|
|
|
|
|
|
class _FakePool:
|
|
def __init__(self, conn):
|
|
self._conn = conn
|
|
|
|
def acquire(self):
|
|
return _FakeAcquire(self._conn)
|
|
|
|
|
|
def test_status_full_parses_heartbeat_and_inverter_timestamps(monkeypatch):
|
|
# Regression: /status/full used to pass string timestamps into _age_seconds()
|
|
# which expects datetime and accesses .tzinfo.
|
|
from app.routers import full_status
|
|
|
|
async def _fake_fetch_json(conn, sql, *args):
|
|
assert "fn_site_full_status" in sql
|
|
return {
|
|
"site": {"code": "X"},
|
|
"operating_mode": {"mode_code": "AUTO"},
|
|
"heartbeat": {"last_seen": "2026-04-20T08:56:36.186Z"},
|
|
"inverter_latest": {"measured_at": "2026-04-20T08:56:31.165Z"},
|
|
"ev_chargers": [],
|
|
"heat_pump_latest": None,
|
|
"battery_limits": {},
|
|
"active_plan": None,
|
|
"planning_intervals": [],
|
|
"tomorrow_price_slot_count": 96,
|
|
}
|
|
|
|
monkeypatch.setattr(full_status, "fetch_json", _fake_fetch_json)
|
|
|
|
out = asyncio.run(
|
|
full_status.get_site_status_full(site_id=2, pool=_FakePool(conn=object()))
|
|
)
|
|
assert isinstance(out, dict)
|
|
assert out["heartbeat"]["last_seen"] is not None
|
|
assert out["heartbeat"]["age_seconds"] is not None
|
|
assert out["telemetry"]["inverter"]["measured_at"] is not None
|
|
assert out["telemetry"]["inverter"]["age_seconds"] is not None
|
|
|