403 lines
18 KiB
Python
403 lines
18 KiB
Python
"""M8-T15 tests for the binding-aware thermal ledger."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
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, WarmteLinkReading
|
|
from app.services import timezone as timezone_service
|
|
from app.services.meter_cost import compute_closed_periods, compute_period, recompute_range, summarize
|
|
|
|
|
|
@pytest.fixture()
|
|
def db(tmp_path: Path) -> Session:
|
|
url = f"sqlite:///{tmp_path / 'meter_cost.db'}"
|
|
cfg = Config("alembic_app.ini")
|
|
cfg.set_main_option("sqlalchemy.url", url)
|
|
command.upgrade(cfg, "head")
|
|
engine = create_engine(url)
|
|
session = Session(engine)
|
|
yield session
|
|
session.close()
|
|
engine.dispose()
|
|
|
|
|
|
T0 = datetime(2026, 6, 23, 10, tzinfo=UTC)
|
|
|
|
VALUES = {
|
|
"variable": {"heating": "20.0", "hot_water_heating": "4.0", "hot_water": "2.0", "hot_water_tax": "1.0"},
|
|
"standing": {"heating_network": "365", "metering": "73", "delivery_set": "0", "hot_water_network": "0", "other": "0"},
|
|
}
|
|
|
|
|
|
def _contract(session: Session, *, values: dict = VALUES, start: datetime = T0 - timedelta(days=1)) -> EnergyContractVersion:
|
|
now = datetime.now(UTC)
|
|
contract = EnergyContract(name="thermal", kind="district_heating", scope="thermal", active=True,
|
|
currency="EUR", created_at=now, updated_at=now)
|
|
session.add(contract)
|
|
session.flush()
|
|
version = EnergyContractVersion(contract_id=contract.id, effective_from=start, values=values, created_at=now)
|
|
session.add(version)
|
|
session.flush()
|
|
return version
|
|
|
|
|
|
def _domain(session: Session, commodity: str, *, start: datetime = T0 - timedelta(days=1), end: datetime | None = None):
|
|
now = datetime.now(UTC)
|
|
meter = Meter(label=commodity, commodity=commodity, started_at=start, ended_at=end,
|
|
reason="initial", created_at=now)
|
|
source = MeterSource(name=f"{commodity}-source", kind="warmtelink_serial", enabled=True, config={},
|
|
status="online", created_at=now, updated_at=now)
|
|
session.add_all((meter, source))
|
|
session.flush()
|
|
unit = "GJ" if commodity == "heating" else "m3"
|
|
channel = MeterSourceChannel(source_id=source.id, channel_key=commodity, label=commodity,
|
|
suggested_commodity=commodity, unit=unit, latest_quality="valid",
|
|
created_at=now, updated_at=now)
|
|
session.add(channel)
|
|
session.flush()
|
|
binding = MeterSourceBinding(meter_id=meter.id, channel_id=channel.id, started_at=start, ended_at=end,
|
|
created_at=now, updated_at=now)
|
|
session.add(binding)
|
|
session.flush()
|
|
return meter, channel, binding
|
|
|
|
|
|
def _reading(session: Session, channel: MeterSourceChannel, at: datetime, value: str, quality: str = "valid") -> None:
|
|
session.add(WarmteLinkReading(channel_id=channel.id, recorded_at=at, received_at=at, value=Decimal(value),
|
|
unit=channel.unit, quality=quality, equipment_fingerprint="test"))
|
|
|
|
|
|
def _setup_good(session: Session) -> None:
|
|
_contract(session)
|
|
_, heating, _ = _domain(session, "heating")
|
|
_, water, _ = _domain(session, "hot_water")
|
|
_reading(session, heating, T0, "10.000")
|
|
_reading(session, heating, T0 + timedelta(minutes=15), "10.050")
|
|
_reading(session, water, T0, "20.000")
|
|
_reading(session, water, T0 + timedelta(minutes=15), "20.200")
|
|
session.commit()
|
|
|
|
|
|
def _row(session: Session, commodity: str) -> MeterCostPeriod:
|
|
return session.execute(select(MeterCostPeriod).where(MeterCostPeriod.commodity == commodity)).scalar_one()
|
|
|
|
|
|
def _row_at(session: Session, commodity: str, start: datetime) -> MeterCostPeriod:
|
|
return session.execute(select(MeterCostPeriod).where(
|
|
MeterCostPeriod.commodity == commodity, MeterCostPeriod.period_start == start
|
|
)).scalar_one()
|
|
|
|
|
|
def _version(
|
|
session: Session, contract: EnergyContract, *, start: datetime, values: dict
|
|
) -> EnergyContractVersion:
|
|
version = EnergyContractVersion(
|
|
contract_id=contract.id, effective_from=start, values=values, created_at=datetime.now(UTC)
|
|
)
|
|
session.add(version)
|
|
session.flush()
|
|
return version
|
|
|
|
|
|
def test_two_commodity_decimal_breakdown_and_scheduler_idempotency(db: Session) -> None:
|
|
_setup_good(db)
|
|
assert compute_period(db, "heating", T0)
|
|
assert compute_period(db, "hot_water", T0)
|
|
db.commit()
|
|
heating, water = _row(db, "heating"), _row(db, "hot_water")
|
|
assert heating.quantity == Decimal("0.050000")
|
|
assert heating.cost == Decimal("1.000000000")
|
|
assert water.cost == Decimal("1.400000000")
|
|
assert {key: Decimal(value) for key, value in water.cost_breakdown.items()} == {
|
|
"hot_water_heating": Decimal("0.8"), "hot_water": Decimal("0.4"), "hot_water_tax": Decimal("0.2")
|
|
}
|
|
assert not compute_period(db, "heating", T0)
|
|
|
|
|
|
@pytest.mark.parametrize("commodity,start,end,reason", [
|
|
("heating", "10", "9.99", "negative_delta"),
|
|
("heating", "10", "10.101", "delta_limit_exceeded"),
|
|
("hot_water", "10", "11.001", "delta_limit_exceeded"),
|
|
])
|
|
def test_bad_deltas_are_degraded(db: Session, commodity: str, start: str, end: str, reason: str) -> None:
|
|
_contract(db)
|
|
_, channel, _ = _domain(db, commodity)
|
|
_reading(db, channel, T0, start)
|
|
_reading(db, channel, T0 + timedelta(minutes=15), end)
|
|
db.commit()
|
|
compute_period(db, commodity, T0)
|
|
db.commit()
|
|
row = _row(db, commodity)
|
|
assert row.degraded and row.degraded_reason == reason and row.contract_version_id is None
|
|
|
|
|
|
def test_freshness_quality_binding_and_contract_fail_closed(db: Session) -> None:
|
|
_contract(db)
|
|
meter, channel, binding = _domain(db, "heating")
|
|
_reading(db, channel, T0 - timedelta(seconds=121), "1")
|
|
_reading(db, channel, T0 + timedelta(minutes=15), "1.01")
|
|
db.commit()
|
|
compute_period(db, "heating", T0)
|
|
assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
|
|
# Recompute sees invalid quality independently at the end boundary.
|
|
db.query(WarmteLinkReading).delete()
|
|
_reading(db, channel, T0, "1")
|
|
_reading(db, channel, T0 + timedelta(minutes=15), "1.01", "invalid")
|
|
db.commit()
|
|
recompute_range(db, T0, T0 + timedelta(minutes=15))
|
|
assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
|
|
binding.ended_at = T0 + timedelta(minutes=15)
|
|
db.commit()
|
|
recompute_range(db, T0, T0 + timedelta(minutes=15))
|
|
assert _row(db, "heating").degraded_reason == "missing_or_ambiguous_binding"
|
|
assert meter.id is not None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("first_quality", "last_quality", "accepted", "period_quality"),
|
|
[
|
|
("valid", "valid", True, "valid"),
|
|
("valid", "unverifiable", True, "unverifiable"),
|
|
("unverifiable", "valid", True, "unverifiable"),
|
|
("unverifiable", "unverifiable", True, "unverifiable"),
|
|
("invalid", "valid", False, "invalid"),
|
|
("valid", "invalid", False, "invalid"),
|
|
],
|
|
)
|
|
def test_accepted_reading_quality_is_preserved_without_promotion(
|
|
db: Session, first_quality: str, last_quality: str, accepted: bool, period_quality: str
|
|
) -> None:
|
|
_contract(db)
|
|
_, channel, _ = _domain(db, "heating")
|
|
_reading(db, channel, T0, "10", first_quality)
|
|
_reading(db, channel, T0 + timedelta(minutes=15), "10.05", last_quality)
|
|
db.commit()
|
|
assert compute_period(db, "heating", T0)
|
|
db.commit()
|
|
row = _row(db, "heating")
|
|
assert row.degraded is not accepted
|
|
assert row.quality == period_quality
|
|
if not accepted:
|
|
assert row.degraded_reason == "missing_stale_or_invalid_reading"
|
|
|
|
|
|
def test_readings_must_be_inside_binding_and_meter_windows(db: Session) -> None:
|
|
_contract(db)
|
|
meter, channel, binding = _domain(db, "heating", start=T0)
|
|
_reading(db, channel, T0 - timedelta(seconds=60), "10")
|
|
_reading(db, channel, T0 + timedelta(minutes=15), "10.05")
|
|
db.commit()
|
|
compute_period(db, "heating", T0)
|
|
assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
|
|
|
|
# A candidate after a closed binding is equally outside the cumulative domain.
|
|
binding.started_at = T0 - timedelta(days=1)
|
|
binding.ended_at = T0 + timedelta(minutes=15, seconds=30)
|
|
meter.started_at = T0 - timedelta(days=1)
|
|
db.query(WarmteLinkReading).delete()
|
|
_reading(db, channel, T0, "10")
|
|
_reading(db, channel, T0 + timedelta(minutes=15, seconds=60), "10.05")
|
|
db.commit()
|
|
recompute_range(db, T0, T0 + timedelta(minutes=15))
|
|
assert _row(db, "heating").degraded_reason == "missing_stale_or_invalid_reading"
|
|
|
|
|
|
def test_freshness_is_independent_and_inclusive_with_nearest_candidate(db: Session) -> None:
|
|
_contract(db)
|
|
_, channel, _ = _domain(db, "heating")
|
|
_reading(db, channel, T0 - timedelta(seconds=120), "10")
|
|
_reading(db, channel, T0 + timedelta(minutes=15, seconds=120), "10.05")
|
|
db.commit()
|
|
compute_period(db, "heating", T0)
|
|
db.commit()
|
|
row = _row(db, "heating")
|
|
assert not row.degraded and row.quantity == Decimal("0.050000")
|
|
|
|
|
|
def test_closed_scheduler_retries_degraded_but_not_normal(db: Session) -> None:
|
|
_contract(db)
|
|
_, channel, _ = _domain(db, "heating")
|
|
# The first closed scheduler pass records a degraded row. Supplying the
|
|
# missing boundary inputs later must let its next pass repair that row.
|
|
db.commit()
|
|
assert compute_closed_periods(db, now=T0 + timedelta(minutes=16)) > 0
|
|
assert _row_at(db, "heating", T0).degraded
|
|
_reading(db, channel, T0, "10")
|
|
_reading(db, channel, T0 + timedelta(minutes=15), "10.05")
|
|
db.commit()
|
|
assert compute_closed_periods(db, now=T0 + timedelta(minutes=16)) > 0
|
|
normal = _row_at(db, "heating", T0)
|
|
assert not normal.degraded
|
|
frozen = (normal.cost, dict(normal.pricing_snapshot), normal.updated_at)
|
|
end_reading = db.execute(select(WarmteLinkReading).where(
|
|
WarmteLinkReading.channel_id == channel.id,
|
|
WarmteLinkReading.recorded_at == T0 + timedelta(minutes=15),
|
|
)).scalar_one()
|
|
end_reading.value = Decimal("10.090")
|
|
version = db.get(EnergyContractVersion, normal.contract_version_id)
|
|
assert version is not None
|
|
version.values = {**version.values, "variable": {**version.values["variable"], "heating": "99"}}
|
|
db.commit()
|
|
compute_closed_periods(db, now=T0 + timedelta(minutes=16))
|
|
db.expire_all()
|
|
unchanged = db.execute(select(MeterCostPeriod).where(MeterCostPeriod.id == normal.id)).scalar_one()
|
|
assert (unchanged.cost, unchanged.pricing_snapshot, unchanged.updated_at) == frozen
|
|
|
|
|
|
def test_explicit_recompute_overwrites_and_source_switch_degrades(db: Session) -> None:
|
|
_setup_good(db)
|
|
compute_period(db, "heating", T0)
|
|
db.commit()
|
|
row = _row(db, "heating")
|
|
old_cost = row.cost
|
|
channel = db.get(MeterSourceChannel, row.source_binding.channel_id)
|
|
end_reading = db.execute(select(WarmteLinkReading).where(
|
|
WarmteLinkReading.channel_id == channel.id,
|
|
WarmteLinkReading.recorded_at == T0 + timedelta(minutes=15),
|
|
)).scalar_one()
|
|
end_reading.value = Decimal("10.080")
|
|
db.commit()
|
|
assert recompute_range(db, T0, T0 + timedelta(minutes=15)) == 2
|
|
assert _row(db, "heating").cost != old_cost
|
|
# A hand-off precisely at the right boundary cannot form a single domain.
|
|
binding = _row(db, "heating").source_binding
|
|
binding.ended_at = T0 + timedelta(minutes=15)
|
|
db.commit()
|
|
recompute_range(db, T0, T0 + timedelta(minutes=15))
|
|
assert _row(db, "heating").degraded_reason == "missing_or_ambiguous_binding"
|
|
|
|
|
|
def test_real_source_binding_handoff_never_crosses_cumulative_registers(db: Session) -> None:
|
|
_contract(db)
|
|
meter, old_channel, old_binding = _domain(db, "heating")
|
|
handoff = T0 + timedelta(minutes=15)
|
|
old_binding.ended_at = handoff
|
|
new_meter, new_channel, new_binding = _domain(db, "heating", start=handoff, end=handoff)
|
|
# Keep the same Meter: this is a source/channel/binding handoff, not a meter swap.
|
|
new_binding.meter_id = meter.id
|
|
new_binding.ended_at = None
|
|
assert new_meter.id != meter.id
|
|
_reading(db, old_channel, T0, "1000")
|
|
_reading(db, old_channel, handoff - timedelta(seconds=1), "1000.01")
|
|
_reading(db, new_channel, handoff, "7")
|
|
_reading(db, new_channel, handoff + timedelta(seconds=1), "7.01")
|
|
db.commit()
|
|
compute_period(db, "heating", T0)
|
|
assert _row(db, "heating").degraded_reason == "cross_source_binding"
|
|
|
|
|
|
def test_real_meter_epoch_handoff_never_crosses_cumulative_registers(db: Session) -> None:
|
|
_contract(db)
|
|
handoff = T0 + timedelta(minutes=15)
|
|
_, old_channel, _ = _domain(db, "heating", end=handoff)
|
|
_, new_channel, _ = _domain(db, "heating", start=handoff)
|
|
_reading(db, old_channel, T0, "1000")
|
|
_reading(db, old_channel, handoff - timedelta(seconds=1), "1000.01")
|
|
_reading(db, new_channel, handoff, "2")
|
|
_reading(db, new_channel, handoff + timedelta(seconds=1), "2.01")
|
|
db.commit()
|
|
compute_period(db, "heating", T0)
|
|
assert _row(db, "heating").degraded_reason == "cross_meter_epoch"
|
|
|
|
|
|
def test_summary_fixed_once_per_contract_day_and_dst(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_setup_good(db)
|
|
compute_period(db, "heating", T0)
|
|
compute_period(db, "hot_water", T0)
|
|
db.commit()
|
|
monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
|
|
result = summarize(db, T0, T0 + timedelta(days=1), now=datetime(2026, 6, 24, 1, 6, tzinfo=UTC))
|
|
assert result["variable_cost"] == Decimal("2.400000000")
|
|
assert result["fixed_cost"] == Decimal("2.4") # two local days, each charged once not per commodity
|
|
# DST local day is charged once too, despite being 23 hours long.
|
|
db.execute(select(EnergyContractVersion)).scalar_one().effective_from = datetime(2026, 3, 1, tzinfo=UTC)
|
|
db.commit()
|
|
dst_start = datetime(2026, 3, 28, 23, tzinfo=UTC)
|
|
dst = summarize(db, dst_start, dst_start + timedelta(days=2), now=datetime(2026, 3, 31, tzinfo=UTC))
|
|
assert dst["fixed_cost"] == Decimal("3.6")
|
|
|
|
|
|
def test_summary_is_half_open_at_local_midnight_and_settlement(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
_contract(db)
|
|
monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
|
|
local = ZoneInfo("Europe/Amsterdam")
|
|
start = datetime(2026, 6, 24, 0, tzinfo=local).astimezone(UTC)
|
|
end = datetime(2026, 6, 25, 0, tzinfo=local).astimezone(UTC)
|
|
assert summarize(db, start, start, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("0")
|
|
assert summarize(db, end, start, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("0")
|
|
before = datetime(2026, 6, 24, 1, 4, 59, tzinfo=local).astimezone(UTC)
|
|
after = datetime(2026, 6, 24, 1, 5, tzinfo=local).astimezone(UTC)
|
|
assert summarize(db, start, end, now=before)["fixed_cost"] == Decimal("0")
|
|
assert summarize(db, start, end, now=after)["fixed_cost"] == Decimal("1.2")
|
|
|
|
|
|
def test_summary_versions_cover_first_day_intra_day_and_cross_day(db: Session, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
|
|
local = ZoneInfo("Europe/Amsterdam")
|
|
day = datetime(2026, 6, 24, tzinfo=local)
|
|
first = _contract(db, start=(day + timedelta(hours=6)).astimezone(UTC))
|
|
first.values = {**VALUES, "standing": {**VALUES["standing"], "heating_network": "365"}}
|
|
second_values = {**VALUES, "standing": {**VALUES["standing"], "heating_network": "730"}}
|
|
switch = (day + timedelta(hours=18)).astimezone(UTC)
|
|
first.effective_to = switch
|
|
_version(db, first.contract, start=switch, values=second_values)
|
|
db.commit()
|
|
start = day.astimezone(UTC)
|
|
next_day = (day + timedelta(days=1)).astimezone(UTC)
|
|
following = (day + timedelta(days=2)).astimezone(UTC)
|
|
# Initial-version day only owns 18h: V1 owns 12h, V2 6h; on the following
|
|
# day V2 owns the full daily fixed amount.
|
|
assert summarize(db, start, next_day, now=following + timedelta(hours=2))["fixed_cost"] == Decimal("1.15")
|
|
assert summarize(db, start, following, now=following + timedelta(hours=2))["fixed_cost"] == Decimal("3.35")
|
|
|
|
|
|
def test_normal_row_has_full_audit_snapshot_and_single_commodity_scheduler(db: Session) -> None:
|
|
version = _contract(db)
|
|
meter, channel, binding = _domain(db, "heating")
|
|
_reading(db, channel, T0, "10")
|
|
_reading(db, channel, T0 + timedelta(minutes=15), "10.05")
|
|
db.commit()
|
|
compute_closed_periods(db, now=T0 + timedelta(minutes=16))
|
|
row = _row_at(db, "heating", T0)
|
|
assert (row.meter_id, row.source_binding_id, row.contract_version_id) == (meter.id, binding.id, version.id)
|
|
assert row.pricing_snapshot == VALUES
|
|
assert row.currency == "EUR" and {
|
|
key: Decimal(value) for key, value in row.cost_breakdown.items()
|
|
} == {"heating": Decimal("1.0")}
|
|
assert not row.degraded
|
|
assert _row_at(db, "hot_water", T0).degraded
|
|
|
|
|
|
@pytest.mark.parametrize(("commodity", "end_value"), [("heating", "10.1"), ("hot_water", "11")])
|
|
def test_delta_limit_is_inclusive_at_exact_boundary(db: Session, commodity: str, end_value: str) -> None:
|
|
_contract(db)
|
|
_, channel, _ = _domain(db, commodity)
|
|
_reading(db, channel, T0, "10")
|
|
_reading(db, channel, T0 + timedelta(minutes=15), end_value)
|
|
db.commit()
|
|
compute_period(db, commodity, T0)
|
|
assert not _row(db, commodity).degraded
|
|
|
|
|
|
@pytest.mark.parametrize("day", [datetime(2026, 3, 29), datetime(2026, 10, 25)])
|
|
def test_summary_dst_local_midnights_charge_one_daily_rate(
|
|
db: Session, monkeypatch: pytest.MonkeyPatch, day: datetime
|
|
) -> None:
|
|
_contract(db, start=datetime(2026, 1, 1, tzinfo=UTC))
|
|
monkeypatch.setattr(timezone_service, "local_tz", lambda: ZoneInfo("Europe/Amsterdam"))
|
|
local = ZoneInfo("Europe/Amsterdam")
|
|
start = day.replace(tzinfo=local).astimezone(UTC)
|
|
end = (day + timedelta(days=1)).replace(tzinfo=local).astimezone(UTC)
|
|
assert summarize(db, start, end, now=end + timedelta(hours=2))["fixed_cost"] == Decimal("1.2")
|