219 lines
10 KiB
Python
219 lines
10 KiB
Python
"""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
|
|
from zoneinfo import ZoneInfo
|
|
|
|
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, monkeypatch: pytest.MonkeyPatch
|
|
) -> 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)
|
|
monkeypatch.setattr("app.services.timezone.local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
|
|
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() == []
|