"""Tests for M7-T05: Meter CRUD + swap declaration + retroactive recompute API. Coverage matrix --------------- GET /api/energy/meters - unauthenticated → 401 - authenticated, no meters → 200, items=[] - after declaring meters → items ordered by started_at asc POST /api/energy/meters - unauthenticated → 401 - missing CSRF → 403 - valid (first meter, no active) → 201, MeterResponse - valid swap (active meter exists) → 201, old meter closed, new meter active - retroactive started_at → 201, triggers recompute for affected window - started_at before active meter's started_at (overlap) → 422 PATCH /api/energy/meters/{id} - unauthenticated → 401 - missing CSRF → 403 - not found → 404 - rename label → 200, label updated - edit note → 200, note updated - correct started_at (retroactive) → 200, triggers recompute for affected window - started_at interval violation → 422 Retroactive recompute integration - PATCH started_at change triggers recompute over min(old, new)..now window - POST retroactive declaration triggers recompute from new started_at """ from __future__ import annotations from datetime import UTC, datetime, timedelta, timezone from decimal import Decimal from unittest.mock import patch import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine, event, select from sqlalchemy.orm import Session from app.models.energy import EnergyCostPeriod, Meter, MeterCostPeriod from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel # --------------------------------------------------------------------------- # Shared helpers # --------------------------------------------------------------------------- _CSRF = "test-csrf-token" def _login(client: TestClient) -> None: resp = client.post( "/api/auth/login", json={"username": "admin", "password": "test-password"}, ) assert resp.status_code == 200, f"Login failed: {resp.status_code} {resp.text}" def _declare_payload(**overrides) -> dict: base = { "label": "Test Meter", "started_at": datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC).isoformat(), "reason": "initial", "commodity": "electricity", } base.update(overrides) return base def _add_bound_channel( engine, *, meter_id: int, started_at: datetime, ended_at: datetime | None = None, unit: str = "kWh", ) -> str: """Persist one test-only channel binding and return its public UUID.""" with Session(engine) as session: source = MeterSource( name="Test DSMR", kind="dsmr_mqtt", enabled=True, config={}, status="online", created_at=started_at, updated_at=started_at, ) session.add(source) session.flush() channel = MeterSourceChannel( source_id=source.id, channel_key=f"test-total-{meter_id}", label="Test total", unit=unit, created_at=started_at, updated_at=started_at, ) session.add(channel) session.flush() session.add( MeterSourceBinding( meter_id=meter_id, channel_id=channel.id, started_at=started_at, ended_at=ended_at, created_at=started_at, updated_at=started_at, ) ) session.commit() return channel.uuid # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture(autouse=True) def mock_publish_discovery(): """Auto-mock publish_discovery for all tests in this module. The meters API now calls _trigger_discovery_republish (best-effort) after every successful write. publish_discovery is lazy-imported inside that helper, so we patch it at its canonical source path (app.services.ha_discovery.publish_discovery). Tests that need to assert the call receive this fixture explicitly; all others benefit from the isolation it provides (no live MQTT broker required). """ with patch( "app.services.ha_discovery.publish_discovery", return_value=None, ) as mock: yield mock @pytest.fixture() def meters_client(auth_database): """TestClient + SQLAlchemy engine for Meter API tests.""" from app.main import create_app app_url = auth_database["app_url"] engine = create_engine(app_url, connect_args={"check_same_thread": False}) fastapi_app = create_app() with TestClient(fastapi_app) as test_client: yield test_client, engine engine.dispose() # --------------------------------------------------------------------------- # GET /api/energy/meters # --------------------------------------------------------------------------- def test_list_meters_unauthenticated_returns_401(meters_client): client, _ = meters_client resp = client.get("/api/energy/meters") assert resp.status_code == 401 def test_list_meters_empty(meters_client): client, _ = meters_client _login(client) resp = client.get("/api/energy/meters") assert resp.status_code == 200 body = resp.json() assert body["items"] == [] assert body["total"] == 0 def test_list_meters_ordered_by_started_at(meters_client): """After declaring two meters, list returns them in ascending started_at order.""" client, _ = meters_client _login(client) t0 = datetime(2024, 6, 1, 0, 0, 0, tzinfo=UTC) t1 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): # Declare first meter (initial) resp1 = client.post( "/api/energy/meters", json=_declare_payload(label="Meter A", started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) assert resp1.status_code == 201 # Declare swap meter resp2 = client.post( "/api/energy/meters", json=_declare_payload(label="Meter B", started_at=t1.isoformat(), reason="meter_swap"), headers={"X-CSRF-Token": _CSRF}, ) assert resp2.status_code == 201 resp = client.get("/api/energy/meters") assert resp.status_code == 200 body = resp.json() assert body["total"] == 2 items = body["items"] # Ordered by started_at ascending: Meter A first, Meter B second assert items[0]["label"] == "Meter A" assert items[1]["label"] == "Meter B" # Meter B is active (ended_at is null) assert items[1]["ended_at"] is None # Meter A is closed assert items[0]["ended_at"] is not None # --------------------------------------------------------------------------- # POST /api/energy/meters # --------------------------------------------------------------------------- def test_declare_meter_unauthenticated_returns_401(meters_client): client, _ = meters_client resp = client.post( "/api/energy/meters", json=_declare_payload(), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 401 def test_declare_meter_missing_csrf_returns_403(meters_client): client, _ = meters_client _login(client) resp = client.post("/api/energy/meters", json=_declare_payload()) assert resp.status_code == 403 def test_declare_meter_first_no_active(meters_client): """Declaring the first meter succeeds without closing any previous meter.""" client, engine = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload(label="First Meter", started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 201 body = resp.json() assert body["label"] == "First Meter" assert body["commodity"] == "electricity" assert body["reason"] == "initial" assert body["ended_at"] is None # active # DB check: one meter, active with Session(engine) as s: meters = s.execute(select(Meter)).scalars().all() assert len(meters) == 1 assert meters[0].ended_at is None def test_declare_meter_swap_closes_previous(meters_client): """Declaring a swap closes the previous active meter at started_at.""" client, engine = meters_client _login(client) t0 = datetime(2024, 6, 1, 0, 0, 0, tzinfo=UTC) t1 = datetime(2025, 3, 15, 12, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): # First meter resp1 = client.post( "/api/energy/meters", json=_declare_payload(label="Old Meter", started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) assert resp1.status_code == 201 old_id = resp1.json()["id"] # Swap resp2 = client.post( "/api/energy/meters", json=_declare_payload(label="New Meter", started_at=t1.isoformat(), reason="meter_swap"), headers={"X-CSRF-Token": _CSRF}, ) assert resp2.status_code == 201 body2 = resp2.json() assert body2["label"] == "New Meter" assert body2["ended_at"] is None # new meter is active # DB check: old meter is closed at t1 with Session(engine) as s: old_meter = s.get(Meter, old_id) assert old_meter is not None assert old_meter.ended_at is not None # ended_at should equal t1 (modulo naive/aware round-trip) ended_naive = old_meter.ended_at if ended_naive.tzinfo is None: ended_naive = ended_naive.replace(tzinfo=UTC) assert ended_naive == t1 def test_declare_meter_swap_hands_off_previous_meter_channel_atomically(meters_client): client, engine = meters_client _login(client) t0 = datetime(2024, 6, 1, tzinfo=UTC) boundary = datetime(2025, 3, 15, 12, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): old_response = client.post( "/api/energy/meters", json=_declare_payload(label="Old meter", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) assert old_response.status_code == 201 old_id = old_response.json()["id"] channel_uuid = _add_bound_channel(engine, meter_id=old_id, started_at=t0) response = client.post( "/api/energy/meters", json=_declare_payload( label="New meter", started_at=boundary.isoformat(), reason="meter_swap", source_channel_uuid=channel_uuid, ), headers={"X-CSRF-Token": _CSRF}, ) assert response.status_code == 201 new_id = response.json()["id"] with Session(engine) as session: bindings = session.execute( select(MeterSourceBinding).order_by(MeterSourceBinding.id) ).scalars().all() old_binding_ended_at = bindings[0].ended_at if old_binding_ended_at is not None and old_binding_ended_at.tzinfo is None: old_binding_ended_at = old_binding_ended_at.replace(tzinfo=UTC) new_binding_started_at = bindings[1].started_at if new_binding_started_at.tzinfo is None: new_binding_started_at = new_binding_started_at.replace(tzinfo=UTC) assert [(bindings[0].meter_id, old_binding_ended_at), (bindings[1].meter_id, bindings[1].ended_at)] == [ (old_id, boundary), (new_id, None), ] assert new_binding_started_at == boundary def test_declare_meter_swap_rejects_other_meter_channel_and_rolls_back(meters_client): client, engine = meters_client _login(client) t0 = datetime(2024, 6, 1, tzinfo=UTC) boundary = datetime(2025, 3, 15, 12, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): old_response = client.post( "/api/energy/meters", json=_declare_payload(label="Old meter", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) old_id = old_response.json()["id"] other = Meter( label="Other meter", commodity="electricity", started_at=t0, ended_at=boundary + timedelta(days=1), reason="initial", created_at=t0, ) with Session(engine) as session: session.add(other) session.commit() other_id = other.id channel_uuid = _add_bound_channel(engine, meter_id=other_id, started_at=t0) response = client.post( "/api/energy/meters", json=_declare_payload( label="Rejected meter", started_at=boundary.isoformat(), reason="meter_swap", source_channel_uuid=channel_uuid, ), headers={"X-CSRF-Token": _CSRF}, ) assert response.status_code == 422 with Session(engine) as session: assert session.execute(select(Meter).where(Meter.label == "Rejected meter")).scalar_one_or_none() is None assert session.get(Meter, old_id).ended_at is None binding = session.execute(select(MeterSourceBinding)).scalar_one() assert binding.ended_at is None def test_declare_meter_non_swap_cannot_take_previous_meter_channel(meters_client): client, engine = meters_client _login(client) t0 = datetime(2024, 6, 1, tzinfo=UTC) boundary = datetime(2025, 3, 15, 12, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): old_response = client.post( "/api/energy/meters", json=_declare_payload(label="Old meter", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) old_id = old_response.json()["id"] channel_uuid = _add_bound_channel(engine, meter_id=old_id, started_at=t0) response = client.post( "/api/energy/meters", json=_declare_payload( label="Moved meter", started_at=boundary.isoformat(), reason="home_move", source_channel_uuid=channel_uuid, ), headers={"X-CSRF-Token": _CSRF}, ) assert response.status_code == 422 with Session(engine) as session: assert session.get(Meter, old_id).ended_at is None assert session.execute(select(MeterSourceBinding)).scalar_one().ended_at is None def test_declare_meter_recompute_failure_rolls_back_handoff(meters_client): client, engine = meters_client _login(client) t0 = datetime(2024, 6, 1, tzinfo=UTC) boundary = datetime(2025, 3, 15, 12, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): old_response = client.post( "/api/energy/meters", json=_declare_payload(label="Old meter", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) old_id = old_response.json()["id"] channel_uuid = _add_bound_channel(engine, meter_id=old_id, started_at=t0) with patch("app.api.routes.api.meters.recompute_range", side_effect=RuntimeError("recompute failed")): with pytest.raises(RuntimeError, match="recompute failed"): client.post( "/api/energy/meters", json=_declare_payload( label="Failed meter", started_at=boundary.isoformat(), reason="meter_swap", source_channel_uuid=channel_uuid, ), headers={"X-CSRF-Token": _CSRF}, ) with Session(engine) as session: assert session.execute(select(Meter).where(Meter.label == "Failed meter")).scalar_one_or_none() is None assert session.get(Meter, old_id).ended_at is None assert session.execute(select(MeterSourceBinding)).scalar_one().ended_at is None def test_declare_meter_final_commit_failure_rolls_back_handoff_and_recompute(meters_client): """A real recompute remains uncommitted until the route's final commit succeeds.""" client, engine = meters_client _login(client) boundary = datetime.now(UTC) - timedelta(minutes=45) boundary = boundary.replace(minute=boundary.minute - boundary.minute % 15, second=0, microsecond=0) old_start = boundary - timedelta(days=1) with Session(engine) as session: old_meter = Meter( label="Old meter", commodity="electricity", started_at=old_start, reason="initial", created_at=old_start, ) session.add(old_meter) session.commit() old_id = old_meter.id channel_uuid = _add_bound_channel(engine, meter_id=old_id, started_at=old_start) def fail_final_commit(_session: Session) -> None: raise RuntimeError("final commit failed") event.listen(Session, "before_commit", fail_final_commit) try: with pytest.raises(RuntimeError, match="final commit failed"): client.post( "/api/energy/meters", json=_declare_payload( label="New meter", started_at=boundary.isoformat(), reason="meter_swap", source_channel_uuid=channel_uuid, ), headers={"X-CSRF-Token": _CSRF}, ) finally: event.remove(Session, "before_commit", fail_final_commit) with Session(engine) as session: assert session.execute(select(Meter).where(Meter.label == "New meter")).scalar_one_or_none() is None assert session.get(Meter, old_id).ended_at is None binding = session.execute(select(MeterSourceBinding)).scalar_one() assert binding.ended_at is None assert session.execute(select(EnergyCostPeriod)).scalars().all() == [] def test_declare_meter_overlap_returns_422(meters_client): """Declaring a meter with started_at before active meter's started_at → 422.""" client, _ = meters_client _login(client) t0 = datetime(2025, 6, 1, 0, 0, 0, tzinfo=UTC) t_before = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): # Declare first meter resp = client.post( "/api/energy/meters", json=_declare_payload(started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 201 # Attempt to declare a meter before t0 → overlap error # (t_before is in the past so would trigger recompute, but service layer raises first) resp2 = client.post( "/api/energy/meters", json=_declare_payload( label="Backdated Meter", started_at=t_before.isoformat(), reason="meter_swap" ), headers={"X-CSRF-Token": _CSRF}, ) assert resp2.status_code == 422 assert "started_at" in resp2.json()["detail"].lower() def test_declare_meter_missing_fields_returns_422(meters_client): """Missing required fields (started_at, reason) → 422 from Pydantic validation.""" client, _ = meters_client _login(client) resp = client.post( "/api/energy/meters", json={"label": "No Reason Meter"}, # missing started_at and reason headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 422 def test_declare_meter_response_fields(meters_client): """POST response contains all expected MeterResponse fields.""" client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload( label="Full Fields Meter", started_at=t0.isoformat(), reason="home_move", note="Testing note", ), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 201 body = resp.json() for field in ("id", "label", "commodity", "started_at", "ended_at", "reason", "note", "created_at"): assert field in body, f"Missing field {field!r} in response" assert body["note"] == "Testing note" assert body["reason"] == "home_move" def test_declare_meter_retroactive_triggers_recompute(meters_client): """Declaring a meter with started_at in the past triggers recompute_range. Both POST calls are made with recompute_range mocked so the test does not spend time iterating over thousands of empty quarter-hour periods. """ client, _ = meters_client _login(client) t0 = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC) t_past = datetime(2025, 3, 1, 0, 0, 0, tzinfo=UTC) with patch( "app.api.routes.api.meters.recompute_range", return_value=5 ) as mock_recompute: # First meter (initial); also in the past, so recompute is called here too. client.post( "/api/energy/meters", json=_declare_payload(started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) # Reset call count before the swap we are actually testing. mock_recompute.reset_mock() # Retroactive swap: started_at in the past → should trigger recompute resp = client.post( "/api/energy/meters", json=_declare_payload( label="Retroactive Swap", started_at=t_past.isoformat(), reason="meter_swap" ), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 201 # recompute_range should have been called with start == t_past assert mock_recompute.called assert mock_recompute.call_args.kwargs["commit"] is False assert mock_recompute.call_args.kwargs["strict"] is True call_args = mock_recompute.call_args recompute_start = call_args[0][1] # positional arg index 1 (session is 0) # Normalise for comparison if recompute_start.tzinfo is None: recompute_start = recompute_start.replace(tzinfo=UTC) assert recompute_start <= t_past # --------------------------------------------------------------------------- # PATCH /api/energy/meters/{id} # --------------------------------------------------------------------------- def test_patch_meter_unauthenticated_returns_401(meters_client): client, _ = meters_client resp = client.patch( "/api/energy/meters/1", json={"label": "Renamed"}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 401 def test_patch_meter_missing_csrf_returns_403(meters_client): client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload(started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) meter_id = resp.json()["id"] resp = client.patch(f"/api/energy/meters/{meter_id}", json={"label": "Renamed"}) assert resp.status_code == 403 def test_patch_meter_not_found_returns_404(meters_client): client, _ = meters_client _login(client) resp = client.patch( "/api/energy/meters/99999", json={"label": "Does Not Exist"}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 404 def test_patch_meter_rename_label(meters_client): """PATCH label updates the meter's human-readable label.""" client, engine = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload(label="Original Label", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) meter_id = resp.json()["id"] resp = client.patch( f"/api/energy/meters/{meter_id}", json={"label": "Updated Label"}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 200 assert resp.json()["label"] == "Updated Label" # DB check with Session(engine) as s: m = s.get(Meter, meter_id) assert m.label == "Updated Label" def test_patch_meter_edit_note(meters_client): """PATCH note updates the meter's note field.""" client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload(started_at=t0.isoformat(), note=None), headers={"X-CSRF-Token": _CSRF}, ) meter_id = resp.json()["id"] resp = client.patch( f"/api/energy/meters/{meter_id}", json={"note": "Added a note"}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 200 assert resp.json()["note"] == "Added a note" def test_patch_meter_started_at_retroactive_triggers_recompute(meters_client): """PATCH started_at triggers recompute over min(old, new)..now window.""" client, _ = meters_client _login(client) # Set up two meters: initial + swap. All POST calls are mocked to avoid # running recompute over thousands of empty historical periods. t0 = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC) t1 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): client.post( "/api/energy/meters", json=_declare_payload(label="Meter A", started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) resp_b = client.post( "/api/energy/meters", json=_declare_payload(label="Meter B", started_at=t1.isoformat(), reason="meter_swap"), headers={"X-CSRF-Token": _CSRF}, ) meter_b_id = resp_b.json()["id"] # Correct Meter B's started_at to a slightly different past timestamp t1_corrected = datetime(2024, 12, 15, 0, 0, 0, tzinfo=UTC) # earlier than t1 with patch( "app.api.routes.api.meters.recompute_range", return_value=10 ) as mock_recompute: resp = client.patch( f"/api/energy/meters/{meter_b_id}", json={"started_at": t1_corrected.isoformat()}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 200 # recompute should be triggered assert mock_recompute.called assert mock_recompute.call_args.kwargs["commit"] is False assert mock_recompute.call_args.kwargs["strict"] is True call_args = mock_recompute.call_args recompute_start = call_args[0][1] if recompute_start.tzinfo is None: recompute_start = recompute_start.replace(tzinfo=UTC) # Window start should be min(t1_corrected, t1) = t1_corrected assert recompute_start <= t1_corrected def test_patch_meter_started_at_interval_violation_returns_422(meters_client): """PATCH started_at that would create an invalid interval → 422.""" client, _ = meters_client _login(client) # Set up: initial meter A, then swap to B. POST calls mocked to avoid slow recompute. t0 = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC) t1 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp_a = client.post( "/api/energy/meters", json=_declare_payload(label="Meter A", started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) meter_a_id = resp_a.json()["id"] client.post( "/api/energy/meters", json=_declare_payload(label="Meter B", started_at=t1.isoformat(), reason="meter_swap"), headers={"X-CSRF-Token": _CSRF}, ) # Try to set Meter A's started_at to after its ended_at (t1) → interval error t_too_late = datetime(2025, 6, 1, 0, 0, 0, tzinfo=UTC) resp = client.patch( f"/api/energy/meters/{meter_a_id}", json={"started_at": t_too_late.isoformat()}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 422 @pytest.mark.parametrize("shift", ["later", "earlier"]) def test_patch_meter_started_at_rejects_boundary_shift_that_strands_binding( meters_client, mock_publish_discovery, shift ): """Rejected boundary shifts leave adjacent meters/bindings untouched and emit no side effects.""" client, engine = meters_client _login(client) t0 = datetime(2024, 1, 1, tzinfo=UTC) boundary = datetime(2025, 1, 1, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): meter_a = client.post( "/api/energy/meters", json=_declare_payload(label="A", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ).json() meter_b = client.post( "/api/energy/meters", json=_declare_payload( label="B", started_at=boundary.isoformat(), reason="meter_swap" ), headers={"X-CSRF-Token": _CSRF}, ).json() if shift == "later": _add_bound_channel(engine, meter_id=meter_b["id"], started_at=boundary) proposed = boundary + timedelta(days=1) else: _add_bound_channel(engine, meter_id=meter_a["id"], started_at=t0) with Session(engine) as session: binding = session.scalar(select(MeterSourceBinding)) assert binding is not None binding.ended_at = boundary session.commit() proposed = boundary - timedelta(days=1) mock_publish_discovery.reset_mock() with patch("app.api.routes.api.meters.recompute_range", return_value=0) as recompute: response = client.patch( f"/api/energy/meters/{meter_b['id']}", json={"started_at": proposed.isoformat()}, headers={"X-CSRF-Token": _CSRF}, ) assert response.status_code == 422 recompute.assert_not_called() mock_publish_discovery.assert_not_called() with Session(engine) as observer: assert observer.get(Meter, meter_a["id"]).ended_at.replace(tzinfo=UTC) == boundary assert observer.get(Meter, meter_b["id"]).started_at.replace(tzinfo=UTC) == boundary binding = observer.scalar(select(MeterSourceBinding)) assert binding is not None if shift == "later": assert binding.meter_id == meter_b["id"] and binding.ended_at is None else: assert binding.meter_id == meter_a["id"] assert binding.ended_at.replace(tzinfo=UTC) == boundary def test_patch_meter_no_recompute_when_started_at_not_changed(meters_client): """PATCH that only changes label does NOT trigger recompute.""" client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0) as mock_recompute: resp = client.post( "/api/energy/meters", json=_declare_payload(started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) meter_id = resp.json()["id"] mock_recompute.reset_mock() resp = client.patch( f"/api/energy/meters/{meter_id}", json={"label": "Renamed Only"}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 200 # recompute should NOT be triggered (no started_at change) assert not mock_recompute.called @pytest.mark.parametrize("representation", ["aware_utc", "naive_local"]) def test_patch_meter_rejects_future_started_at_before_any_side_effect( meters_client, mock_publish_discovery, monkeypatch, representation ): """Future aware and local-naive starts leave all persisted state untouched.""" client, engine = meters_client _login(client) started = datetime.now(UTC) - timedelta(hours=2) with patch("app.api.routes.api.meters.recompute_range", return_value=0): declared = client.post( "/api/energy/meters", json=_declare_payload( label="Original", note="Original note", started_at=started.isoformat() ), headers={"X-CSRF-Token": _CSRF}, ) meter_id = declared.json()["id"] mock_publish_discovery.reset_mock() future = datetime.now(UTC) + timedelta(hours=2) if representation == "aware_utc": proposed = future.isoformat() else: monkeypatch.setattr( "app.services.timezone.local_tz", lambda: timezone(timedelta(hours=2)) ) proposed = (future + timedelta(hours=2)).replace(tzinfo=None).isoformat() with patch("app.api.routes.api.meters.recompute_range", return_value=0) as recompute: response = client.patch( f"/api/energy/meters/{meter_id}", json={"label": "Changed", "note": "Changed note", "started_at": proposed}, headers={"X-CSRF-Token": _CSRF}, ) assert response.status_code == 422 recompute.assert_not_called() mock_publish_discovery.assert_not_called() with Session(engine) as observer: meter = observer.get(Meter, meter_id) assert meter is not None assert meter.label == "Original" assert meter.note == "Original note" assert meter.started_at.replace(tzinfo=UTC) == started @pytest.mark.parametrize("commodity", ["heating", "hot_water"]) def test_patch_thermal_started_at_uses_thermal_recompute(meters_client, monkeypatch, commodity): """A successful thermal correction never routes through electricity recompute.""" from app.services import meter_cost client, _ = meters_client _login(client) started = datetime.now(UTC) - timedelta(hours=3) calls = [] monkeypatch.setattr( meter_cost, "recompute_range", lambda db, start, end, *, commit: calls.append((start, end, commit)) or 0, ) with patch("app.api.routes.api.meters.recompute_range", return_value=0) as electricity: declared = client.post( "/api/energy/meters", json=_declare_payload( commodity=commodity, label=f"{commodity} meter", started_at=started.isoformat(), ), headers={"X-CSRF-Token": _CSRF}, ) assert declared.status_code == 201 calls.clear() electricity.reset_mock() response = client.patch( f"/api/energy/meters/{declared.json()['id']}", json={"started_at": (started + timedelta(minutes=30)).isoformat()}, headers={"X-CSRF-Token": _CSRF}, ) assert response.status_code == 200 assert len(calls) == 1 assert calls[0][2] is False electricity.assert_not_called() def _create_adjacent_thermal_patch_state(engine, commodity: str): """Create an editable thermal boundary with two untouched bindings.""" old_start = datetime.now(UTC) - timedelta(hours=5) boundary = old_start + timedelta(hours=1) shifted = boundary + timedelta(minutes=30) current_binding_start = shifted + timedelta(minutes=30) with Session(engine) as session: previous = Meter( label="Previous thermal meter", commodity=commodity, started_at=old_start, ended_at=boundary, reason="initial", created_at=old_start, ) current = Meter( label="Current thermal meter", commodity=commodity, started_at=boundary, reason="meter_swap", created_at=boundary, ) session.add_all([previous, current]) session.commit() previous_id, current_id = previous.id, current.id unit = {"heating": "GJ", "hot_water": "m³"}[commodity] _add_bound_channel( engine, meter_id=previous_id, started_at=old_start, ended_at=boundary, unit=unit ) _add_bound_channel(engine, meter_id=current_id, started_at=current_binding_start, unit=unit) return previous_id, current_id, boundary, shifted, current_binding_start @pytest.mark.parametrize("commodity", ["heating", "hot_water"]) @pytest.mark.parametrize("failure", ["recompute", "flush", "commit"]) def test_patch_thermal_failure_rolls_back_lifecycle_and_cost_state( meters_client, mock_publish_discovery, monkeypatch, commodity, failure ): """Thermal PATCH failures roll back meters, bindings, cost writes, and HA.""" from app.services import meter_cost client, engine = meters_client _login(client) previous_id, current_id, boundary, shifted, binding_start = _create_adjacent_thermal_patch_state( engine, commodity ) def recompute_with_uncommitted_cost(db, start, end, *, commit): assert commit is False db.add( MeterCostPeriod( commodity=commodity, period_start=shifted, period_end=shifted + timedelta(minutes=15), quantity=Decimal("0"), cost=Decimal("0"), currency="EUR", cost_breakdown={}, pricing_snapshot={}, quality="invalid", degraded=True, degraded_reason="test rollback", created_at=shifted, updated_at=shifted, ) ) if failure == "recompute": raise RuntimeError("thermal recompute failed") return 0 monkeypatch.setattr(meter_cost, "recompute_range", recompute_with_uncommitted_cost) if failure == "flush": monkeypatch.setattr( Session, "flush", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("thermal flush failed")), ) if failure == "commit": def fail_commit(_session: Session) -> None: raise RuntimeError("thermal commit failed") event.listen(Session, "before_commit", fail_commit) try: expected = f"thermal {failure} failed" with pytest.raises(RuntimeError, match=expected): client.patch( f"/api/energy/meters/{current_id}", json={"started_at": shifted.isoformat()}, headers={"X-CSRF-Token": _CSRF}, ) finally: if failure == "commit": event.remove(Session, "before_commit", fail_commit) if failure == "flush": monkeypatch.undo() mock_publish_discovery.assert_not_called() with Session(engine) as observer: previous = observer.get(Meter, previous_id) current = observer.get(Meter, current_id) assert previous is not None and previous.ended_at.replace(tzinfo=UTC) == boundary assert current is not None and current.started_at.replace(tzinfo=UTC) == boundary bindings = observer.execute( select(MeterSourceBinding).order_by(MeterSourceBinding.meter_id) ).scalars().all() assert bindings[0].ended_at.replace(tzinfo=UTC) == boundary assert bindings[1].started_at.replace(tzinfo=UTC) == binding_start assert observer.execute(select(MeterCostPeriod)).scalars().all() == [] # --------------------------------------------------------------------------- # Timeline continuity (recompute mocked to avoid slow computation over empty quarters) # --------------------------------------------------------------------------- def test_swap_timeline_continuity(meters_client): """After two swaps, meter timeline is contiguous and self-consistent.""" client, engine = meters_client _login(client) t0 = datetime(2023, 1, 1, 0, 0, 0, tzinfo=UTC) t1 = datetime(2024, 1, 1, 0, 0, 0, tzinfo=UTC) t2 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) # Declare 3 meters in sequence. POST calls mocked to avoid slow recompute over # years of empty quarter-hour periods. with patch("app.api.routes.api.meters.recompute_range", return_value=0): for label, ts, reason in [ ("M1", t0, "initial"), ("M2", t1, "meter_swap"), ("M3", t2, "meter_swap"), ]: resp = client.post( "/api/energy/meters", json=_declare_payload(label=label, started_at=ts.isoformat(), reason=reason), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 201 # Verify timeline via list endpoint resp = client.get("/api/energy/meters") items = resp.json()["items"] assert len(items) == 3 # M1: [t0, t1); M2: [t1, t2); M3: [t2, None) m1 = next(i for i in items if i["label"] == "M1") m2 = next(i for i in items if i["label"] == "M2") m3 = next(i for i in items if i["label"] == "M3") assert m1["ended_at"] is not None assert m2["ended_at"] is not None assert m3["ended_at"] is None # active # ended_at of M1 == started_at of M2 (contiguous) m1_ended = datetime.fromisoformat(m1["ended_at"]).replace(tzinfo=None) m2_started = datetime.fromisoformat(m2["started_at"]).replace(tzinfo=None) assert m1_ended == m2_started m2_ended = datetime.fromisoformat(m2["ended_at"]).replace(tzinfo=None) m3_started = datetime.fromisoformat(m3["started_at"]).replace(tzinfo=None) assert m2_ended == m3_started # --------------------------------------------------------------------------- # Reason enum validation # --------------------------------------------------------------------------- def test_declare_meter_invalid_reason_returns_422(meters_client): """Unknown reason value → 422 from Pydantic enum validation.""" client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) resp = client.post( "/api/energy/meters", json=_declare_payload(started_at=t0.isoformat(), reason="unknown_reason_xyz"), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 422 # --------------------------------------------------------------------------- # Retroactive recompute & boundary update (recompute mocked) — window coverage check # --------------------------------------------------------------------------- def test_patch_started_at_earlier_updates_boundary(meters_client): """Moving started_at earlier should update the previous meter's ended_at.""" client, engine = meters_client _login(client) t0 = datetime(2024, 6, 1, 0, 0, 0, tzinfo=UTC) t1 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) t1_earlier = datetime(2024, 12, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp_a = client.post( "/api/energy/meters", json=_declare_payload(label="Meter A", started_at=t0.isoformat(), reason="initial"), headers={"X-CSRF-Token": _CSRF}, ) meter_a_id = resp_a.json()["id"] resp_b = client.post( "/api/energy/meters", json=_declare_payload(label="Meter B", started_at=t1.isoformat(), reason="meter_swap"), headers={"X-CSRF-Token": _CSRF}, ) meter_b_id = resp_b.json()["id"] # Correct Meter B's started_at to t1_earlier (moves boundary earlier) resp = client.patch( f"/api/energy/meters/{meter_b_id}", json={"started_at": t1_earlier.isoformat()}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 200 assert resp.json()["id"] == meter_b_id # DB check: Meter A's ended_at should now equal t1_earlier with Session(engine) as s: meter_a = s.get(Meter, meter_a_id) assert meter_a is not None ended = meter_a.ended_at if ended is not None and ended.tzinfo is None: ended = ended.replace(tzinfo=UTC) assert ended == t1_earlier # --------------------------------------------------------------------------- # FUE-T06: HA discovery re-publish triggered after meter writes # --------------------------------------------------------------------------- def test_declare_meter_triggers_publish_discovery(meters_client, mock_publish_discovery): """POST /api/energy/meters triggers publish_discovery after successful commit.""" client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload(label="Discovery Meter", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 201 # publish_discovery must have been called exactly once after the declare. mock_publish_discovery.assert_called_once() def test_patch_meter_triggers_publish_discovery(meters_client, mock_publish_discovery): """PATCH /api/energy/meters/{id} triggers publish_discovery after successful commit.""" client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with patch("app.api.routes.api.meters.recompute_range", return_value=0): resp = client.post( "/api/energy/meters", json=_declare_payload(label="Original Label", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) meter_id = resp.json()["id"] # Reset call count: the POST above also called publish_discovery. mock_publish_discovery.reset_mock() resp = client.patch( f"/api/energy/meters/{meter_id}", json={"label": "Renamed Label"}, headers={"X-CSRF-Token": _CSRF}, ) assert resp.status_code == 200 # publish_discovery must have been called exactly once after the PATCH. mock_publish_discovery.assert_called_once() def test_declare_meter_succeeds_when_publish_discovery_raises(meters_client): """publish_discovery raising an exception must NOT cause POST declare to return 500. The _trigger_discovery_republish helper is best-effort: it swallows all exceptions so that a broken MQTT / discovery layer never breaks the API. """ client, _ = meters_client _login(client) t0 = datetime(2025, 1, 1, 0, 0, 0, tzinfo=UTC) with ( patch("app.api.routes.api.meters.recompute_range", return_value=0), patch( "app.services.ha_discovery.publish_discovery", side_effect=RuntimeError("MQTT broker unreachable"), ), ): resp = client.post( "/api/energy/meters", json=_declare_payload(label="Best Effort Meter", started_at=t0.isoformat()), headers={"X-CSRF-Token": _CSRF}, ) # The meter must be created successfully despite the discovery failure. assert resp.status_code == 201 assert resp.json()["label"] == "Best Effort Meter" @pytest.mark.parametrize("commodity", ["heating", "hot_water"]) def test_thermal_declare_and_close_use_meter_cost_recompute(meters_client, monkeypatch, commodity): """Thermal lifecycle routes use the meter-cost helper's actual signature.""" from app.services import meter_cost client, _ = meters_client _login(client) calls = [] monkeypatch.setattr(meter_cost, "recompute_range", lambda db, start, end, *, commit: calls.append((start, end, commit)) or 0) started = datetime.now(UTC) - timedelta(hours=2) declared = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json={ "label": f"{commodity} meter", "started_at": started.isoformat(), "reason": "initial", "commodity": commodity, }) assert declared.status_code == 201 closed = client.post(f"/api/energy/meters/{declared.json()['id']}/close", headers={"X-CSRF-Token": _CSRF}, json={ "ended_at": (started + timedelta(hours=1)).isoformat(), }) assert closed.status_code == 200 assert len(calls) == 2 and all(call[2] is False for call in calls) def test_close_meter_closes_open_bindings_and_enforces_auth_csrf(meters_client): client, engine = meters_client started = datetime.now(UTC) - timedelta(hours=2) _login(client) with patch("app.api.routes.api.meters.recompute_range", return_value=0): declared = client.post("/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload( label="closable", started_at=started.isoformat(), reason="initial", )) assert declared.status_code == 201 _add_bound_channel(engine, meter_id=declared.json()["id"], started_at=started) boundary = started + timedelta(hours=1) assert client.post(f"/api/energy/meters/{declared.json()['id']}/close", json={"ended_at": boundary.isoformat()}).status_code == 403 with patch("app.api.routes.api.meters.recompute_range", return_value=0): closed = client.post(f"/api/energy/meters/{declared.json()['id']}/close", headers={"X-CSRF-Token": _CSRF}, json={ "ended_at": boundary.isoformat(), }) assert closed.status_code == 200 with Session(engine) as session: binding = session.scalar(select(MeterSourceBinding)) assert binding is not None and binding.ended_at.replace(tzinfo=UTC) == boundary @pytest.mark.parametrize("commodity,unit", [("electricity", "kWh"), ("heating", "GJ"), ("hot_water", "m³")]) @pytest.mark.parametrize("operation", ["close", "declare"]) def test_lifecycle_rejects_retained_closed_binding_beyond_proposed_end( meters_client, mock_publish_discovery, commodity, unit, operation, ): """Close and declare fail closed before recompute or HA for every commodity.""" client, engine = meters_client _login(client) start = datetime.now(UTC) - timedelta(hours=3) boundary = start + timedelta(hours=1) retained_end = start + timedelta(hours=2) declared = client.post( "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload(label="retained-history", started_at=start.isoformat(), reason="initial", commodity=commodity), ) assert declared.status_code == 201 meter_id = declared.json()["id"] _add_bound_channel(engine, meter_id=meter_id, started_at=start, ended_at=retained_end, unit=unit) mock_publish_discovery.reset_mock() with patch("app.api.routes.api.meters._recompute_commodity", side_effect=AssertionError("must not recompute")): if operation == "close": response = client.post( f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, json={"ended_at": boundary.isoformat()}, ) else: response = client.post( "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload( label="replacement", started_at=boundary.isoformat(), reason="meter_swap", commodity=commodity, ), ) assert response.status_code == 422 assert mock_publish_discovery.call_count == 0 with Session(engine) as observer: meter = observer.get(Meter, meter_id) binding = observer.scalar(select(MeterSourceBinding).where(MeterSourceBinding.meter_id == meter_id)) assert meter is not None and meter.ended_at is None assert binding is not None and binding.ended_at.replace(tzinfo=UTC) == retained_end assert observer.scalars(select(Meter).where(Meter.commodity == commodity)).all() == [meter] def test_close_flushes_lifecycle_boundary_before_strict_recompute(meters_client): """The strict recompute query observes the just-closed meter and binding.""" client, engine = meters_client _login(client) started = datetime.now(UTC) - timedelta(hours=2) boundary = started + timedelta(hours=1) with patch("app.api.routes.api.meters.recompute_range", return_value=0): declared = client.post( "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload(label="flush-visible", started_at=started.isoformat(), reason="initial"), ) meter_id = declared.json()["id"] _add_bound_channel(engine, meter_id=meter_id, started_at=started) def observe(session, *_args, **kwargs): assert kwargs == {"commit": False, "strict": True} observed_meter = session.get(Meter, meter_id) observed_binding = session.scalar(select(MeterSourceBinding)) assert observed_meter is not None and observed_meter.ended_at is not None assert observed_binding is not None and observed_binding.ended_at is not None return 0 with patch("app.api.routes.api.meters.recompute_range", side_effect=observe): response = client.post( f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, json={"ended_at": boundary.isoformat()}, ) assert response.status_code == 200 def test_close_strict_compute_failure_rolls_back_persisted_lifecycle_state(meters_client, monkeypatch): """A real per-period strict failure rolls back the close in a fresh Session.""" from app.services import energy_cost client, engine = meters_client _login(client) started = datetime.now(UTC) - timedelta(hours=2) boundary = started + timedelta(hours=1) with patch("app.api.routes.api.meters.recompute_range", return_value=0): declared = client.post( "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload(label="strict-rollback", started_at=started.isoformat(), reason="initial"), ) meter_id = declared.json()["id"] _add_bound_channel(engine, meter_id=meter_id, started_at=started) monkeypatch.setattr( energy_cost, "compute_period", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("strict boom")) ) with pytest.raises(RuntimeError, match="strict boom"): client.post( f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, json={"ended_at": boundary.isoformat()}, ) with Session(engine) as observer: assert observer.get(Meter, meter_id).ended_at is None assert observer.scalar(select(MeterSourceBinding)).ended_at is None def test_close_flush_failure_rolls_back_persisted_lifecycle_state(meters_client, monkeypatch): """The mandatory pre-recompute flush shares the route rollback boundary.""" client, engine = meters_client _login(client) started = datetime.now(UTC) - timedelta(hours=2) with patch("app.api.routes.api.meters.recompute_range", return_value=0): declared = client.post( "/api/energy/meters", headers={"X-CSRF-Token": _CSRF}, json=_declare_payload(label="flush-rollback", started_at=started.isoformat(), reason="initial"), ) meter_id = declared.json()["id"] _add_bound_channel(engine, meter_id=meter_id, started_at=started) monkeypatch.setattr(Session, "flush", lambda *_args, **_kwargs: (_ for _ in ()).throw(RuntimeError("flush boom"))) with pytest.raises(RuntimeError, match="flush boom"): client.post( f"/api/energy/meters/{meter_id}/close", headers={"X-CSRF-Token": _CSRF}, json={"ended_at": (started + timedelta(hours=1)).isoformat()}, ) monkeypatch.undo() with Session(engine) as observer: assert observer.get(Meter, meter_id).ended_at is None assert observer.scalar(select(MeterSourceBinding)).ended_at is None