"""API coverage for the thermal meter-cost ledger (M8-T16).""" from __future__ import annotations from datetime import UTC, datetime, timedelta from decimal import Decimal from unittest.mock import patch import pytest from fastapi.testclient import TestClient from sqlalchemy import create_engine, select from sqlalchemy.orm import Session from app.models.energy import EnergyContract, EnergyContractVersion, Meter, MeterCostPeriod from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel _CSRF = "test-csrf-token" _T0 = datetime(2026, 6, 23, 10, tzinfo=UTC) def _login(client: TestClient) -> str: response = client.post("/api/auth/login", json={"username": "admin", "password": "test-password"}) assert response.status_code == 200 return response.json()["csrf_token"] @pytest.fixture() def meter_cost_client(auth_database): from app.main import create_app engine = create_engine(auth_database["app_url"], connect_args={"check_same_thread": False}) with TestClient(create_app()) as client: yield client, engine engine.dispose() def _row(start: datetime, *, commodity: str = "heating", degraded: bool = False) -> MeterCostPeriod: return MeterCostPeriod( commodity=commodity, period_start=start, period_end=start + timedelta(minutes=15), meter_id=None, source_binding_id=None, contract_version_id=None, quantity=Decimal("0"), cost=Decimal("0") if degraded else Decimal("1.250000000"), currency="EUR", cost_breakdown={} if degraded else {"heating": "1.250000000"}, pricing_snapshot={}, quality="invalid" if degraded else "valid", degraded=degraded, degraded_reason="missing_contract" if degraded else None, created_at=_T0, updated_at=_T0, ) def _normal_ids(db: Session, contract_version_id: int) -> tuple[int, int, int]: now = datetime.now(UTC) meter = Meter(label="heating", commodity="heating", started_at=_T0 - timedelta(days=1), reason="initial", created_at=now) source = MeterSource(name="source", kind="warmtelink_serial", enabled=True, config={}, status="online", created_at=now, updated_at=now) db.add_all((meter, source)) db.flush() channel = MeterSourceChannel(source_id=source.id, channel_key="heating", label="heating", suggested_commodity="heating", unit="GJ", latest_quality="valid", created_at=now, updated_at=now) db.add(channel) db.flush() binding = MeterSourceBinding(meter_id=meter.id, channel_id=channel.id, started_at=meter.started_at, created_at=now, updated_at=now) db.add(binding) db.flush() return meter.id, binding.id, contract_version_id def test_meter_costs_require_auth_and_paginate_decimal_rows(meter_cost_client) -> None: client, engine = meter_cost_client assert client.get("/api/energy/meter-costs?scope=thermal").status_code == 401 with Session(engine) as db: db.add_all([_row(_T0, degraded=True), _row(_T0 + timedelta(minutes=15), degraded=True)]) db.commit() _login(client) response = client.get("/api/energy/meter-costs?scope=thermal&limit=1&offset=1") assert response.status_code == 200 body = response.json() assert body["total"] == 2 and len(body["items"]) == 1 assert body["items"][0]["cost"] == "0.000000000" assert body["items"][0]["degraded_reason"] == "missing_contract" assert client.get("/api/energy/meter-costs?scope=electricity").status_code == 422 def test_meter_costs_half_open_stable_pagination_and_deep_decimal_audit(meter_cost_client) -> None: client, engine = meter_cost_client with Session(engine) as db: normal = _row(_T0, degraded=True) normal.quantity = Decimal("0.050000") normal.cost = Decimal("1.123456789") normal.cost_breakdown = {"heating": Decimal("1.123456789")} normal.pricing_snapshot = {"variable": {"heating": Decimal("22.46913578")}} water = _row(_T0, commodity="hot_water", degraded=True) later = _row(_T0 + timedelta(minutes=15), degraded=True) db.add_all((normal, water, later)) db.commit() _login(client) base = "/api/energy/meter-costs?scope=thermal&start=2026-06-23T10:00:00Z&end=2026-06-23T10:15:00Z" response = client.get(base + "&limit=1&offset=0") assert response.status_code == 200 assert response.json()["total"] == 2 item = response.json()["items"][0] assert item["commodity"] == "heating" assert item["quantity"] == "0.050000" assert item["cost_breakdown"] == {"heating": "1.123456789"} assert item["pricing_snapshot"]["variable"]["heating"] == "22.46913578" assert client.get(base + "&commodity=hot_water").json()["items"][0]["commodity"] == "hot_water" # Tie-breaking includes id, so the second page deterministically returns water. assert client.get(base + "&limit=1&offset=1").json()["items"][0]["commodity"] == "hot_water" assert client.get(base + "&offset=2").json()["items"] == [] def test_meter_cost_summary_empty_and_recompute_csrf_window_validation(meter_cost_client) -> None: client, _engine = meter_cost_client csrf = _login(client) summary = client.get( "/api/energy/meter-costs/summary?scope=thermal&start=2026-06-23T00:00:00Z&end=2026-06-24T00:00:00Z" ) assert summary.status_code == 200 assert summary.json()["all_in"] == "0" assert summary.json()["fixed_breakdown"] == { "heating_network": "0", "metering": "0", "delivery_set": "0", "hot_water_network": "0", "other": "0", } url = "/api/energy/meter-costs/recompute?scope=thermal&start=2026-06-23T10:01:00Z&end=2026-06-23T10:15:00Z" assert client.post(url).status_code == 403 assert client.post(url, headers={"X-CSRF-Token": _CSRF}).status_code == 422 overlarge = ( "/api/energy/meter-costs/recompute?scope=thermal&start=2026-01-01T00:00:00Z" "&end=2026-02-02T00:00:00Z" ) assert client.post(overlarge, headers={"X-CSRF-Token": _CSRF}).status_code == 422 assert client.post( "/api/energy/meter-costs/recompute?scope=thermal&start=2026-06-23T10:15:00Z&end=2026-06-23T10:00:00Z", headers={"X-CSRF-Token": csrf}, ).status_code == 422 assert client.post( "/api/energy/meter-costs/recompute?scope=electricity&start=2026-06-23T10:00:00Z&end=2026-06-23T10:15:00Z", headers={"X-CSRF-Token": csrf}, ).status_code == 422 def test_meter_cost_summary_returns_five_fixed_components_and_totals(meter_cost_client) -> None: client, engine = meter_cost_client now = datetime.now(UTC) values = { "variable": {"heating": "20", "hot_water_heating": "4", "hot_water": "2", "hot_water_tax": "1"}, "standing": {"heating_network": "365", "metering": "73", "delivery_set": "0", "hot_water_network": "0", "other": "0"}, } with Session(engine) as db: contract = EnergyContract(name="thermal", kind="district_heating", scope="thermal", active=True, currency="EUR", created_at=now, updated_at=now) db.add(contract) db.flush() db.add(EnergyContractVersion(contract_id=contract.id, effective_from=_T0 - timedelta(days=2), values=values, created_at=now)) db.flush() # Allocate the contract-version id before normal ledger fixtures. meter_id, binding_id, version_id = _normal_ids(db, db.scalars(select(EnergyContractVersion.id)).one()) heating, water = _row(_T0), _row(_T0, commodity="hot_water") heating.meter_id = water.meter_id = meter_id heating.source_binding_id = water.source_binding_id = binding_id heating.contract_version_id = water.contract_version_id = version_id heating.cost = Decimal("1.000000000") heating.cost_breakdown = {"heating": "1.000000000"} water.cost = Decimal("1.400000000") water.cost_breakdown = {"hot_water_heating": "0.8", "hot_water": "0.4", "hot_water_tax": "0.2"} db.add_all((heating, water)) db.commit() _login(client) with patch("app.api.routes.api.meter_costs.local_now", return_value=datetime(2026, 6, 25, 2, tzinfo=UTC)): response = client.get("/api/energy/meter-costs/summary?scope=thermal&start=2026-06-23T00:00:00Z&end=2026-06-24T00:00:00Z") assert response.status_code == 200 body = response.json() assert body["heating"] == "1.000000000" and body["hot_water_heating"] == "0.8" assert body["fixed_breakdown"] == { "heating_network": "2", "metering": "0.4", "delivery_set": "0", "hot_water_network": "0", "other": "0", } assert body["variable_subtotal"] == "2.400000000" assert body["fixed_subtotal"] == "2.4" and body["all_in"] == "4.800000000" assert body["period_count"] == 2 and body["degraded_count"] == 0 def test_recompute_commits_once_and_rolls_back_service_or_count_failure(meter_cost_client) -> None: client, engine = meter_cost_client csrf = _login(client) url = "/api/energy/meter-costs/recompute?scope=thermal&start=2026-06-23T10:00:00Z&end=2026-06-23T10:15:00Z" headers = {"X-CSRF-Token": csrf} # An exception after a service-side mutation must leave no generated rows. from app.services import meter_cost as service original = service.compute_period calls = 0 def fail_after_first(*args, **kwargs): nonlocal calls calls += 1 if calls == 2: raise RuntimeError("service failure") return original(*args, **kwargs) with patch("app.services.meter_cost.compute_period", side_effect=fail_after_first): with pytest.raises(RuntimeError): client.post(url, headers=headers) with Session(engine) as db: assert db.scalars(select(MeterCostPeriod)).all() == [] # The query after recomputation is part of that same transaction as well. with patch("app.api.routes.api.meter_costs.select", side_effect=RuntimeError("count failure")): with pytest.raises(RuntimeError): client.post(url, headers=headers) with Session(engine) as db: assert db.scalars(select(MeterCostPeriod)).all() == []