tliu93
9d3e28a529
Root cause (found in the live DB): register_at() returned the latest reading for any boundary beyond the DSMR data range (a future instant or a long gap), so future periods got start==end -> delta 0 and were written as degraded=False 'ok' rows with cost 0. recompute_range() over a future end then filled the whole month with such fake rows, which then occupied the slots so the per-minute tick (immutability guard skips non-degraded rows) never recomputed the real data. - register_at: ignore a reading older than one period (15min) before the boundary -> returns None -> period is marked degraded instead of a fake 0. - recompute_range: only process closed periods (t1 <= now); never future ones. - Tests cover staleness window, out-of-range -> degraded, recompute skips future, and a regression that normal closed periods are unaffected.
1418 lines
57 KiB
Python
1418 lines
57 KiB
Python
"""Tests for M6-T07: billing engine (app/services/energy_cost.py).
|
||
|
||
Acceptance criteria covered
|
||
----------------------------
|
||
1. ``compute_period`` with a ``manual`` dual-tariff contract: correct
|
||
import_cost / export_revenue / net_cost; pricing snapshot and
|
||
contract_version_id stored.
|
||
2. ``compute_period`` idempotency: calling twice with overwrite=False leaves
|
||
the row unchanged; overwrite=True re-computes and updates it.
|
||
3. ``compute_period`` with a ``tibber`` contract + a matching TibberPrice:
|
||
correct calculation using ``total`` as buy price.
|
||
4. Missing Tibber price → period is **skipped** (no row written).
|
||
5. Missing readings (no DsmrReading covering a boundary) → degraded row.
|
||
6. Cross-version selection: two contract versions with different effective
|
||
dates; each t0 picks the correct version (asserts contract_version_id).
|
||
7. ``summarize``: Σnet + standing charges (month/30 × days) − heffingskorting
|
||
(year/365 × days) — hand-calculated against known values.
|
||
8. ``compute_closed_periods``: only processes closed and uncalculated periods;
|
||
does not touch already-computed (non-degraded) rows.
|
||
9. ``recompute_range``: explicitly overwrites all periods in [start, end)
|
||
including already-computed rows.
|
||
10. ``floor_to_quarter`` helper returns correct UTC quarter-hour boundaries.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from datetime import UTC, datetime, timedelta
|
||
from decimal import Decimal
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from alembic import command
|
||
from alembic.config import Config
|
||
from sqlalchemy import create_engine, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.integrations.pricing.strategies import PeriodDeltas # noqa: F401
|
||
from app.models.energy import (
|
||
DsmrReading,
|
||
EnergyContract,
|
||
EnergyContractVersion,
|
||
EnergyCostPeriod,
|
||
TibberPrice,
|
||
)
|
||
from app.services.energy_cost import (
|
||
compute_closed_periods,
|
||
compute_period,
|
||
floor_to_quarter,
|
||
register_at,
|
||
recompute_range,
|
||
summarize,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Fixtures
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def _make_app_alembic_config(database_url: str) -> Config:
|
||
cfg = Config("alembic_app.ini")
|
||
cfg.set_main_option("sqlalchemy.url", database_url)
|
||
return cfg
|
||
|
||
|
||
@pytest.fixture()
|
||
def energy_db(tmp_path: Path):
|
||
"""Temporary SQLite DB upgraded to head; yields an open Session."""
|
||
db_path = tmp_path / "energy_cost_test.db"
|
||
db_url = f"sqlite:///{db_path}"
|
||
alembic_cfg = _make_app_alembic_config(db_url)
|
||
command.upgrade(alembic_cfg, "head")
|
||
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
||
session = Session(engine)
|
||
yield session
|
||
session.close()
|
||
engine.dispose()
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Data-builder helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
_UTC = UTC
|
||
|
||
|
||
def _ts(hour: int, minute: int = 0, second: int = 0) -> datetime:
|
||
"""Return a UTC datetime on 2026-06-23 at the given time."""
|
||
return datetime(2026, 6, 23, hour, minute, second, tzinfo=_UTC)
|
||
|
||
|
||
# Manual contract values used across most tests.
|
||
_MANUAL_VALUES = {
|
||
"energy": {
|
||
"buy": {"normal": 0.133, "dal": 0.127},
|
||
"sell": {"normal": 0.05, "dal": 0.05},
|
||
"energy_tax": 0.11,
|
||
"ode": 0.0,
|
||
},
|
||
"standing": {
|
||
"network_fee": 9.87,
|
||
"management_fee": 9.87,
|
||
},
|
||
"credits": {
|
||
"heffingskorting": 600.0,
|
||
},
|
||
}
|
||
|
||
|
||
def _make_contract(
|
||
session: Session,
|
||
*,
|
||
kind: str = "manual",
|
||
active: bool = True,
|
||
currency: str = "EUR",
|
||
) -> EnergyContract:
|
||
"""Insert and flush an EnergyContract; return the ORM object."""
|
||
now = datetime.now(_UTC)
|
||
c = EnergyContract(
|
||
name=f"Test Contract ({kind})",
|
||
kind=kind,
|
||
active=active,
|
||
currency=currency,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
session.add(c)
|
||
session.flush()
|
||
return c
|
||
|
||
|
||
def _make_version(
|
||
session: Session,
|
||
contract: EnergyContract,
|
||
values: dict,
|
||
*,
|
||
effective_from: datetime,
|
||
effective_to: datetime | None = None,
|
||
) -> EnergyContractVersion:
|
||
"""Insert and flush an EnergyContractVersion; return the ORM object."""
|
||
v = EnergyContractVersion(
|
||
contract_id=contract.id,
|
||
effective_from=effective_from,
|
||
effective_to=effective_to,
|
||
values=values,
|
||
created_at=datetime.now(_UTC),
|
||
)
|
||
session.add(v)
|
||
session.flush()
|
||
return v
|
||
|
||
|
||
def _make_reading(
|
||
session: Session,
|
||
*,
|
||
recorded_at: datetime,
|
||
d1: str = "20000.000",
|
||
d2: str = "10000.000",
|
||
r1: str = "5000.000",
|
||
r2: str = "3000.000",
|
||
source_id: int | None = None,
|
||
) -> DsmrReading:
|
||
"""Insert and flush a DsmrReading with the given register values."""
|
||
r = DsmrReading(
|
||
recorded_at=recorded_at,
|
||
source_id=source_id,
|
||
payload={
|
||
"electricity_delivered_1": d1,
|
||
"electricity_delivered_2": d2,
|
||
"electricity_returned_1": r1,
|
||
"electricity_returned_2": r2,
|
||
"current_electricity_usage": "1.234",
|
||
},
|
||
)
|
||
session.add(r)
|
||
session.flush()
|
||
return r
|
||
|
||
|
||
def _make_tibber_price(
|
||
session: Session,
|
||
*,
|
||
starts_at: datetime,
|
||
total: float,
|
||
energy: float = 0.18,
|
||
currency: str = "EUR",
|
||
) -> TibberPrice:
|
||
"""Insert and flush a TibberPrice row; return the ORM object."""
|
||
tax = round(total - energy, 6)
|
||
row = TibberPrice(
|
||
starts_at=starts_at,
|
||
resolution="QUARTER_HOURLY",
|
||
energy=energy,
|
||
tax=tax,
|
||
total=total,
|
||
level="NORMAL",
|
||
currency=currency,
|
||
fetched_at=datetime.now(_UTC),
|
||
)
|
||
session.add(row)
|
||
session.flush()
|
||
return row
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 10. floor_to_quarter helper
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestFloorToQuarter:
|
||
def test_already_on_boundary(self) -> None:
|
||
dt = _ts(10, 0)
|
||
assert floor_to_quarter(dt) == dt
|
||
|
||
def test_15_boundary(self) -> None:
|
||
dt = _ts(10, 15)
|
||
assert floor_to_quarter(dt) == dt
|
||
|
||
def test_floor_mid_period(self) -> None:
|
||
dt = _ts(10, 7, 30)
|
||
assert floor_to_quarter(dt) == _ts(10, 0)
|
||
|
||
def test_floor_minute_16(self) -> None:
|
||
dt = _ts(10, 16)
|
||
assert floor_to_quarter(dt) == _ts(10, 15)
|
||
|
||
def test_floor_minute_44(self) -> None:
|
||
dt = _ts(10, 44, 59)
|
||
assert floor_to_quarter(dt) == _ts(10, 30)
|
||
|
||
def test_floor_minute_45(self) -> None:
|
||
dt = _ts(10, 45)
|
||
assert floor_to_quarter(dt) == _ts(10, 45)
|
||
|
||
def test_seconds_zeroed(self) -> None:
|
||
dt = _ts(10, 3, 45)
|
||
result = floor_to_quarter(dt)
|
||
assert result.second == 0
|
||
assert result.microsecond == 0
|
||
|
||
def test_timezone_preserved(self) -> None:
|
||
dt = _ts(10, 7)
|
||
result = floor_to_quarter(dt)
|
||
assert result.tzinfo is _UTC
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# register_at tests
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRegisterAt:
|
||
def test_returns_none_when_no_readings(self, energy_db: Session) -> None:
|
||
result = register_at(energy_db, _ts(10, 0))
|
||
assert result is None
|
||
|
||
def test_returns_most_recent_at_or_before_boundary(self, energy_db: Session) -> None:
|
||
# Insert two readings: one before boundary, one after.
|
||
_make_reading(energy_db, recorded_at=_ts(9, 55), d1="100.0", d2="200.0", r1="10.0", r2="20.0", source_id=1)
|
||
_make_reading(energy_db, recorded_at=_ts(10, 5), d1="999.0", d2="999.0", r1="999.0", r2="999.0", source_id=2)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, _ts(10, 0))
|
||
assert result is not None
|
||
assert result["d1"] == Decimal("100.0")
|
||
assert result["d2"] == Decimal("200.0")
|
||
|
||
def test_exact_boundary_included(self, energy_db: Session) -> None:
|
||
_make_reading(energy_db, recorded_at=_ts(10, 0), d1="500.0", d2="600.0", r1="50.0", r2="60.0", source_id=1)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, _ts(10, 0))
|
||
assert result is not None
|
||
assert result["d1"] == Decimal("500.0")
|
||
|
||
def test_missing_register_key_returns_none(self, energy_db: Session) -> None:
|
||
r = DsmrReading(
|
||
recorded_at=_ts(10, 0),
|
||
source_id=99,
|
||
payload={"electricity_delivered_1": "100.0"}, # missing d2, r1, r2
|
||
)
|
||
energy_db.add(r)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, _ts(10, 0))
|
||
assert result is None
|
||
|
||
def test_null_register_value_returns_none(self, energy_db: Session) -> None:
|
||
r = DsmrReading(
|
||
recorded_at=_ts(10, 0),
|
||
source_id=88,
|
||
payload={
|
||
"electricity_delivered_1": None,
|
||
"electricity_delivered_2": "200.0",
|
||
"electricity_returned_1": "10.0",
|
||
"electricity_returned_2": "20.0",
|
||
},
|
||
)
|
||
energy_db.add(r)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, _ts(10, 0))
|
||
assert result is None
|
||
|
||
def test_values_are_decimal(self, energy_db: Session) -> None:
|
||
_make_reading(energy_db, recorded_at=_ts(10, 0), d1="20915.154", d2="18372.099",
|
||
r1="1234.567", r2="890.123", source_id=1)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, _ts(10, 0))
|
||
assert result is not None
|
||
assert isinstance(result["d1"], Decimal)
|
||
assert result["d1"] == Decimal("20915.154")
|
||
assert result["r2"] == Decimal("890.123")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 1-2. compute_period — manual dual-tariff
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Hand-calculation reference for the tests below:
|
||
#
|
||
# Start registers (t0 = 10:00): d1=20000.000, d2=10000.000, r1=5000.000, r2=3000.000
|
||
# End registers (t1 = 10:15): d1=20000.500, d2=10001.200, r1=5000.000, r2=3000.100
|
||
#
|
||
# Deltas: Δd1=0.500, Δd2=1.200, Δr1=0.000, Δr2=0.100
|
||
#
|
||
# buy_dal = 0.127 + 0.11 + 0.0 = 0.237
|
||
# buy_normal = 0.133 + 0.11 + 0.0 = 0.243
|
||
# sell_dal = 0.05
|
||
# sell_normal = 0.05
|
||
#
|
||
# import_cost = 0.500×0.237 + 1.200×0.243 = 0.1185 + 0.2916 = 0.4101
|
||
# export_revenue = 0.000×0.05 + 0.100×0.05 = 0.000 + 0.005 = 0.005
|
||
# net_cost = 0.4101 − 0.005 = 0.4051
|
||
|
||
_T0 = _ts(10, 0)
|
||
_T1 = _ts(10, 15)
|
||
|
||
_START_D1 = "20000.000"
|
||
_START_D2 = "10000.000"
|
||
_START_R1 = "5000.000"
|
||
_START_R2 = "3000.000"
|
||
|
||
_END_D1 = "20000.500"
|
||
_END_D2 = "10001.200"
|
||
_END_R1 = "5000.000"
|
||
_END_R2 = "3000.100"
|
||
|
||
|
||
def _setup_manual_scenario(session: Session) -> EnergyContractVersion:
|
||
"""Create active manual contract + two boundary readings; return the version."""
|
||
contract = _make_contract(session, kind="manual", active=True)
|
||
version = _make_version(
|
||
session, contract, _MANUAL_VALUES,
|
||
effective_from=_ts(0, 0), # covers t0=10:00
|
||
)
|
||
# Start reading (at t0)
|
||
_make_reading(session, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
# End reading (at t1)
|
||
_make_reading(session, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
session.commit()
|
||
return version
|
||
|
||
|
||
class TestComputePeriodManual:
|
||
def test_correct_import_cost(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# import_cost = 0.500×0.237 + 1.200×0.243 = 0.4101
|
||
assert abs(row.import_cost - 0.4101) < 1e-9
|
||
|
||
def test_correct_export_revenue(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# export_revenue = 0.000×0.05 + 0.100×0.05 = 0.005
|
||
assert abs(row.export_revenue - 0.005) < 1e-9
|
||
|
||
def test_correct_net_cost(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# net_cost = 0.4101 − 0.005 = 0.4051
|
||
assert abs(row.net_cost - 0.4051) < 1e-9
|
||
|
||
def test_pricing_snapshot_stored(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.pricing["kind"] == "manual"
|
||
assert "buy_dal" in row.pricing
|
||
assert "buy_normal" in row.pricing
|
||
|
||
def test_contract_version_id_stored(self, energy_db: Session) -> None:
|
||
version = _setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.contract_version_id == version.id
|
||
|
||
def test_not_degraded(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.degraded is False
|
||
|
||
def test_kwh_deltas_stored(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert abs(row.d1_kwh - 0.5) < 1e-9
|
||
assert abs(row.d2_kwh - 1.2) < 1e-9
|
||
assert abs(row.r1_kwh - 0.0) < 1e-9
|
||
assert abs(row.r2_kwh - 0.1) < 1e-9
|
||
|
||
def test_currency_stored(self, energy_db: Session) -> None:
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.currency == "EUR"
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 2. Idempotency (overwrite=False / overwrite=True)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestComputePeriodIdempotency:
|
||
def test_no_overwrite_does_not_change_row(self, energy_db: Session) -> None:
|
||
"""Calling compute_period twice with overwrite=False must leave the row unchanged."""
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row_before = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
first_computed_at = row_before.computed_at
|
||
|
||
# Modify an end reading — but overwrite=False should ignore it.
|
||
_make_reading(energy_db, recorded_at=_T1 + timedelta(seconds=1),
|
||
d1="99999.0", d2="99999.0", r1="99999.0", r2="99999.0",
|
||
source_id=99)
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0, overwrite=False)
|
||
assert result is False # did not overwrite
|
||
|
||
row_after = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# computed_at must be unchanged (row not touched).
|
||
assert row_after.computed_at == first_computed_at
|
||
|
||
def test_overwrite_true_updates_row(self, energy_db: Session) -> None:
|
||
"""compute_period with overwrite=True must write to an existing successful row.
|
||
|
||
We verify overwrite=True by checking that the function returns True
|
||
(a write occurred) even though a non-degraded row already exists.
|
||
The row's computed_at timestamp will change because we call compute_period
|
||
again — that is the observable side-effect of overwriting.
|
||
"""
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
|
||
row_before = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row_before.degraded is False, "Pre-condition: row must be non-degraded"
|
||
|
||
# overwrite=True: must return True even though row already exists
|
||
result = compute_period(energy_db, _T0, overwrite=True)
|
||
assert result is True
|
||
|
||
def test_only_one_row_per_period(self, energy_db: Session) -> None:
|
||
"""There must never be two EnergyCostPeriod rows for the same period_start."""
|
||
_setup_manual_scenario(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
compute_period(energy_db, _T0, overwrite=True)
|
||
|
||
rows = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalars().all()
|
||
assert len(rows) == 1
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 3. compute_period — tibber contract + tibber_price
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Hand-calculation:
|
||
# Δd1=0.500, Δd2=1.200 → total_delivered = 1.700 kWh
|
||
# Δr1=0.000, Δr2=0.100 → total_returned = 0.100 kWh
|
||
# tibber total = 0.25 EUR/kWh
|
||
# buy = 0.25; sell = 0.25 − 0.10 (energy_tax) − 0.0 (sell_adjust) = 0.15
|
||
# import_cost = 1.700 × 0.25 = 0.425
|
||
# export_revenue = 0.100 × 0.15 = 0.015
|
||
# net_cost = 0.425 − 0.015 = 0.410
|
||
|
||
_TIBBER_VALUES = {
|
||
"energy": {
|
||
"energy_tax": 0.10,
|
||
"sell_adjust": 0.0,
|
||
},
|
||
"standing": {
|
||
"management_fee": 5.99,
|
||
"network_fee": 9.87,
|
||
},
|
||
"credits": {
|
||
"heffingskorting": 600.0,
|
||
},
|
||
}
|
||
|
||
|
||
class TestComputePeriodTibber:
|
||
def _setup(self, session: Session, total: float = 0.25) -> tuple[EnergyContractVersion, TibberPrice]:
|
||
contract = _make_contract(session, kind="tibber", active=True)
|
||
version = _make_version(session, contract, _TIBBER_VALUES, effective_from=_ts(0, 0))
|
||
_make_reading(session, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(session, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
price = _make_tibber_price(session, starts_at=_ts(9, 45), total=total)
|
||
session.commit()
|
||
return version, price
|
||
|
||
def test_correct_import_cost(self, energy_db: Session) -> None:
|
||
self._setup(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# import_cost = 1.700 × 0.25 = 0.425
|
||
assert abs(row.import_cost - 0.425) < 1e-9
|
||
|
||
def test_correct_export_revenue(self, energy_db: Session) -> None:
|
||
self._setup(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# sell = 0.25 - 0.10 = 0.15; export_revenue = 0.100 × 0.15 = 0.015
|
||
assert abs(row.export_revenue - 0.015) < 1e-9
|
||
|
||
def test_correct_net_cost(self, energy_db: Session) -> None:
|
||
self._setup(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# net_cost = 0.425 - 0.015 = 0.410
|
||
assert abs(row.net_cost - 0.410) < 1e-9
|
||
|
||
def test_contract_version_id_stored(self, energy_db: Session) -> None:
|
||
version, _ = self._setup(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.contract_version_id == version.id
|
||
|
||
def test_pricing_snapshot_has_tibber_fields(self, energy_db: Session) -> None:
|
||
self._setup(energy_db)
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.pricing["kind"] == "tibber"
|
||
assert "buy" in row.pricing
|
||
assert "sell" in row.pricing
|
||
assert "tibber_price_starts_at" in row.pricing
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 4. Missing Tibber price → skip (no row written)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestComputePeriodMissingTibberPrice:
|
||
def test_no_row_written_when_price_missing(self, energy_db: Session) -> None:
|
||
"""When Tibber price is absent for the period, no EnergyCostPeriod is written."""
|
||
contract = _make_contract(energy_db, kind="tibber", active=True)
|
||
_make_version(energy_db, contract, _TIBBER_VALUES, effective_from=_ts(0, 0))
|
||
_make_reading(energy_db, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
# No TibberPrice inserted.
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0)
|
||
assert result is False
|
||
|
||
rows = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalars().all()
|
||
assert len(rows) == 0, "No row should be written when Tibber price is missing"
|
||
|
||
def test_tibber_price_after_t0_counts_as_missing(self, energy_db: Session) -> None:
|
||
"""A TibberPrice with starts_at > t0 must NOT be used; period is skipped."""
|
||
contract = _make_contract(energy_db, kind="tibber", active=True)
|
||
_make_version(energy_db, contract, _TIBBER_VALUES, effective_from=_ts(0, 0))
|
||
_make_reading(energy_db, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
# Price starts AFTER t0 — must not be used.
|
||
_make_tibber_price(energy_db, starts_at=_ts(10, 15), total=0.25)
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0)
|
||
assert result is False
|
||
|
||
rows = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalars().all()
|
||
assert len(rows) == 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 5. Missing readings → degraded row
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestComputePeriodMissingReadings:
|
||
def test_degraded_when_start_reading_missing(self, energy_db: Session) -> None:
|
||
"""No DsmrReading at or before t0 → degraded row written."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# Only an end reading; no start reading.
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0)
|
||
assert result is True
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.degraded is True
|
||
assert row.import_cost == 0.0
|
||
assert row.export_revenue == 0.0
|
||
assert row.net_cost == 0.0
|
||
|
||
def test_degraded_when_end_reading_missing(self, energy_db: Session) -> None:
|
||
"""No DsmrReading at or before t1 → degraded row written.
|
||
|
||
We place a reading BEFORE t0 (so t0 boundary has data) but the first
|
||
reading AT OR AFTER t1 is only after t1+5min, leaving the t1 boundary
|
||
without a reading ≤ t1. This forces ``register_at(t1)`` to return the
|
||
same row as ``register_at(t0)`` — both map to the same pre-t0 reading —
|
||
which means deltas = 0 but NOT degraded.
|
||
|
||
Actually the degraded condition for a *missing end reading* occurs when
|
||
there is NO DsmrReading in the DB at all with ``recorded_at ≤ t1``.
|
||
To achieve that, we only add a reading after t1.
|
||
"""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# Only a reading AFTER t1 — no reading at or before t1.
|
||
_make_reading(energy_db, recorded_at=_ts(10, 20), d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0)
|
||
assert result is True
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.degraded is True
|
||
|
||
def test_degraded_has_no_contract_version_id(self, energy_db: Session) -> None:
|
||
"""Degraded rows written due to missing readings have contract_version_id=None."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# No readings at all.
|
||
energy_db.commit()
|
||
|
||
compute_period(energy_db, _T0)
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.contract_version_id is None
|
||
|
||
def test_degraded_retried_on_second_compute(self, energy_db: Session) -> None:
|
||
"""A degraded row is retried (overwritten) when readings become available.
|
||
|
||
First pass: no readings at all → degraded row (both start and end missing).
|
||
Second pass: add both boundary readings → row transitions to non-degraded.
|
||
"""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# First compute: no readings at all → degraded.
|
||
energy_db.commit()
|
||
|
||
compute_period(energy_db, _T0)
|
||
row_degraded = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row_degraded.degraded is True
|
||
|
||
# Now add both boundary readings and retry with overwrite=False.
|
||
# Degraded rows are retried by compute_period (overwrite=False still re-tries degraded).
|
||
_make_reading(energy_db, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0, overwrite=False)
|
||
assert result is True
|
||
|
||
row_fixed = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row_fixed.degraded is False
|
||
assert row_fixed.import_cost > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 6. Cross-version selection
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestCrossVersionSelection:
|
||
"""Two contract versions with non-overlapping effective dates.
|
||
|
||
Version 1: effective [00:00, 08:00) → lower buy prices
|
||
Version 2: effective [08:00, ∞) → higher buy prices
|
||
|
||
Periods before 08:00 must use version 1; periods at or after 08:00 use v2.
|
||
"""
|
||
|
||
# Different rate sets so we can distinguish which version was used.
|
||
_VALUES_V1 = {
|
||
"energy": {
|
||
"buy": {"normal": 0.10, "dal": 0.10},
|
||
"sell": {"normal": 0.05, "dal": 0.05},
|
||
"energy_tax": 0.0,
|
||
"ode": 0.0,
|
||
},
|
||
"standing": {"network_fee": 5.0, "management_fee": 5.0},
|
||
"credits": {"heffingskorting": 300.0},
|
||
}
|
||
_VALUES_V2 = {
|
||
"energy": {
|
||
"buy": {"normal": 0.50, "dal": 0.50},
|
||
"sell": {"normal": 0.10, "dal": 0.10},
|
||
"energy_tax": 0.0,
|
||
"ode": 0.0,
|
||
},
|
||
"standing": {"network_fee": 5.0, "management_fee": 5.0},
|
||
"credits": {"heffingskorting": 300.0},
|
||
}
|
||
|
||
def _setup(self, session: Session) -> tuple[EnergyContractVersion, EnergyContractVersion]:
|
||
contract = _make_contract(session, kind="manual", active=True)
|
||
v1 = _make_version(
|
||
session, contract, self._VALUES_V1,
|
||
effective_from=_ts(0, 0),
|
||
effective_to=_ts(8, 0),
|
||
)
|
||
v2 = _make_version(
|
||
session, contract, self._VALUES_V2,
|
||
effective_from=_ts(8, 0),
|
||
effective_to=None,
|
||
)
|
||
session.commit()
|
||
return v1, v2
|
||
|
||
def test_period_before_version_boundary_uses_v1(self, energy_db: Session) -> None:
|
||
v1, v2 = self._setup(energy_db)
|
||
t0_early = _ts(7, 0)
|
||
t1_early = _ts(7, 15)
|
||
_make_reading(energy_db, recorded_at=t0_early, d1="1000.0", d2="1000.0",
|
||
r1="0.0", r2="0.0", source_id=10)
|
||
_make_reading(energy_db, recorded_at=t1_early, d1="1001.0", d2="1000.0",
|
||
r1="0.0", r2="0.0", source_id=11)
|
||
energy_db.commit()
|
||
|
||
compute_period(energy_db, t0_early)
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == t0_early)
|
||
).scalar_one()
|
||
assert row.contract_version_id == v1.id, (
|
||
f"Expected version v1 (id={v1.id}), got {row.contract_version_id}"
|
||
)
|
||
# import_cost = 1.0 kWh × buy_normal(v1)=0.10 = 0.10
|
||
assert abs(row.import_cost - 0.10) < 1e-9
|
||
|
||
def test_period_at_version_boundary_uses_v2(self, energy_db: Session) -> None:
|
||
v1, v2 = self._setup(energy_db)
|
||
t0_late = _ts(8, 0)
|
||
t1_late = _ts(8, 15)
|
||
_make_reading(energy_db, recorded_at=t0_late, d1="2000.0", d2="2000.0",
|
||
r1="0.0", r2="0.0", source_id=20)
|
||
_make_reading(energy_db, recorded_at=t1_late, d1="2001.0", d2="2000.0",
|
||
r1="0.0", r2="0.0", source_id=21)
|
||
energy_db.commit()
|
||
|
||
compute_period(energy_db, t0_late)
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == t0_late)
|
||
).scalar_one()
|
||
assert row.contract_version_id == v2.id, (
|
||
f"Expected version v2 (id={v2.id}), got {row.contract_version_id}"
|
||
)
|
||
# import_cost = 1.0 kWh × buy_normal(v2)=0.50 = 0.50
|
||
assert abs(row.import_cost - 0.50) < 1e-9
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7. summarize — hand-verified
|
||
# ---------------------------------------------------------------------------
|
||
|
||
# Hand-calculation for TestSummarize:
|
||
#
|
||
# Interval: 2026-06-23 00:00 UTC to 2026-06-24 00:00 UTC (exactly 1 day)
|
||
# days = 1.0
|
||
#
|
||
# Active contract values (from _MANUAL_VALUES):
|
||
# standing: network_fee=9.87, management_fee=9.87
|
||
# credits: heffingskorting=600.0
|
||
#
|
||
# Fixed costs per day = (9.87 + 9.87) / 30 × 1.0 = 19.74 / 30 = 0.658
|
||
# Credits per day = 600.0 / 365 × 1.0 = 600.0 / 365 ≈ 1.6438...
|
||
#
|
||
# Metered net (from two periods, each net_cost=0.4051):
|
||
# Σnet = 2 × 0.4051 = 0.8102
|
||
# (slightly approximate: actual computed floats from compute_period)
|
||
#
|
||
# total_payable = 0.8102 + 0.658 − 1.6438... ≈ −0.1756...
|
||
|
||
|
||
class TestSummarize:
|
||
def _setup_two_periods(self, session: Session) -> None:
|
||
"""Insert contract + readings for two consecutive 15-min periods and compute them."""
|
||
contract = _make_contract(session, kind="manual", active=True)
|
||
_make_version(session, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
|
||
# Period 1: [10:00, 10:15)
|
||
_make_reading(session, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(session, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
|
||
# Period 2: [10:15, 10:30) — same delta as period 1
|
||
t2 = _ts(10, 30)
|
||
_make_reading(session, recorded_at=t2,
|
||
d1=str(float(_END_D1) + 0.5),
|
||
d2=str(float(_END_D2) + 1.2),
|
||
r1=str(float(_END_R1)),
|
||
r2=str(float(_END_R2) + 0.1),
|
||
source_id=3)
|
||
|
||
session.commit()
|
||
compute_period(session, _T0)
|
||
compute_period(session, _T1)
|
||
session.commit()
|
||
|
||
def test_metered_net_sum(self, energy_db: Session) -> None:
|
||
self._setup_two_periods(energy_db)
|
||
# Summarise over the two computed periods.
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
# Σnet ≈ 2 × 0.4051 = 0.8102
|
||
assert abs(result["metered_net"] - 0.8102) < 1e-6
|
||
|
||
def test_period_count(self, energy_db: Session) -> None:
|
||
self._setup_two_periods(energy_db)
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert result["period_count"] == 2
|
||
assert result["degraded_count"] == 0
|
||
|
||
def test_fixed_costs_formula(self, energy_db: Session) -> None:
|
||
"""Fixed costs = (network_fee + management_fee) / 30 × days."""
|
||
self._setup_two_periods(energy_db)
|
||
# Interval is 30 minutes = 0.5/48 day = 0.020833... days
|
||
days = Decimal("1800") / Decimal("86400")
|
||
expected_fixed = (Decimal("9.87") + Decimal("9.87")) / Decimal("30") * days
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert abs(Decimal(str(result["fixed_costs"])) - expected_fixed) < Decimal("1e-9")
|
||
|
||
def test_credits_formula(self, energy_db: Session) -> None:
|
||
"""Credits = heffingskorting / 365 × days."""
|
||
self._setup_two_periods(energy_db)
|
||
days = Decimal("1800") / Decimal("86400")
|
||
expected_credits = Decimal("600.0") / Decimal("365") * days
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert abs(Decimal(str(result["credits"])) - expected_credits) < Decimal("1e-9")
|
||
|
||
def test_total_payable_formula(self, energy_db: Session) -> None:
|
||
"""total_payable = metered_net + fixed_costs − credits."""
|
||
self._setup_two_periods(energy_db)
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
expected = result["metered_net"] + result["fixed_costs"] - result["credits"]
|
||
assert abs(result["total_payable"] - expected) < 1e-9
|
||
|
||
def test_no_active_contract_returns_zero_standing(self, energy_db: Session) -> None:
|
||
"""When no active contract exists, fixed_costs and credits are both 0."""
|
||
# No contract at all — just compute a period manually and summarise.
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert result["fixed_costs"] == 0.0
|
||
assert result["credits"] == 0.0
|
||
assert result["metered_net"] == 0.0
|
||
assert result["total_payable"] == 0.0
|
||
|
||
def test_currency_from_contract(self, energy_db: Session) -> None:
|
||
self._setup_two_periods(energy_db)
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert result["currency"] == "EUR"
|
||
|
||
def test_degraded_excluded_from_metered_sum(self, energy_db: Session) -> None:
|
||
"""Degraded rows must not contribute to the metered sums.
|
||
|
||
We produce a degraded row by calling compute_period when no readings
|
||
exist at all for the period boundaries.
|
||
"""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# No readings at all → degraded period.
|
||
energy_db.commit()
|
||
compute_period(energy_db, _T0)
|
||
energy_db.commit()
|
||
|
||
result = summarize(energy_db, _T0, _T1)
|
||
assert result["metered_net"] == 0.0
|
||
assert result["degraded_count"] == 1
|
||
assert result["period_count"] == 0
|
||
|
||
def test_one_day_summarize_hand_calc(self, energy_db: Session) -> None:
|
||
"""Full 1-day hand-calculation: fixed_costs/30 and credits/365 with 1-day interval."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
energy_db.commit()
|
||
|
||
# Summarize over exactly 1 day (no periods in DB — only standing/credits).
|
||
start = datetime(2026, 6, 23, 0, 0, 0, tzinfo=_UTC)
|
||
end = datetime(2026, 6, 24, 0, 0, 0, tzinfo=_UTC)
|
||
result = summarize(energy_db, start, end)
|
||
|
||
# fixed_costs = (9.87 + 9.87) / 30 × 1.0 = 0.658
|
||
assert abs(result["fixed_costs"] - (9.87 + 9.87) / 30) < 1e-9
|
||
# credits = 600 / 365
|
||
assert abs(result["credits"] - 600.0 / 365) < 1e-9
|
||
# total = 0 + 0.658 − (600/365)
|
||
expected_total = (9.87 + 9.87) / 30 - 600.0 / 365
|
||
assert abs(result["total_payable"] - expected_total) < 1e-9
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 8. compute_closed_periods
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestComputeClosedPeriods:
|
||
def test_skips_future_periods(self, energy_db: Session) -> None:
|
||
"""Periods whose t1 > now must not be computed."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=datetime.now(_UTC))
|
||
energy_db.commit()
|
||
|
||
# The period [now, now+15min) is still open — should not be computed.
|
||
written = compute_closed_periods(energy_db)
|
||
assert written == 0 # nothing written (no closed periods with data)
|
||
|
||
def test_does_not_overwrite_successful_period(self, energy_db: Session) -> None:
|
||
"""A non-degraded row must not be overwritten by the regular tick.
|
||
|
||
Uses a period from 2 days ago so it is definitely closed and within
|
||
the 7-day lookback window.
|
||
"""
|
||
past_t0 = floor_to_quarter(datetime.now(_UTC) - timedelta(days=2))
|
||
past_t1 = past_t0 + timedelta(minutes=15)
|
||
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=past_t0 - timedelta(hours=1))
|
||
_make_reading(energy_db, recorded_at=past_t0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=past_t1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
# First compute: direct call.
|
||
compute_period(energy_db, past_t0)
|
||
energy_db.commit()
|
||
|
||
row_before = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == past_t0)
|
||
).scalar_one()
|
||
original_computed_at = row_before.computed_at
|
||
|
||
# compute_closed_periods runs — the row should NOT be touched.
|
||
compute_closed_periods(energy_db)
|
||
|
||
row_after = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == past_t0)
|
||
).scalar_one()
|
||
assert row_after.computed_at == original_computed_at
|
||
|
||
def test_retries_degraded_rows(self, energy_db: Session) -> None:
|
||
"""A degraded row is retried when readings become available on the next tick.
|
||
|
||
We use a period from 2 days ago so it is definitely closed and within the
|
||
7-day lookback window, regardless of what time the test runs today.
|
||
"""
|
||
# Use a period that is definitely closed (2 days ago, early morning UTC).
|
||
past_t0 = floor_to_quarter(datetime.now(_UTC) - timedelta(days=2))
|
||
past_t1 = past_t0 + timedelta(minutes=15)
|
||
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=past_t0 - timedelta(hours=1))
|
||
# First compute: no readings → degraded.
|
||
energy_db.commit()
|
||
compute_period(energy_db, past_t0)
|
||
energy_db.commit()
|
||
|
||
row_before = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == past_t0)
|
||
).scalar_one()
|
||
assert row_before.degraded is True
|
||
|
||
# Now add both boundary readings.
|
||
_make_reading(energy_db, recorded_at=past_t0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=past_t1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
# compute_closed_periods should retry the degraded row.
|
||
compute_closed_periods(energy_db)
|
||
|
||
row_after = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == past_t0)
|
||
).scalar_one()
|
||
assert row_after.degraded is False
|
||
assert row_after.import_cost > 0
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 9. recompute_range
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRecomputeRange:
|
||
def test_overwrites_existing_rows(self, energy_db: Session) -> None:
|
||
"""recompute_range must overwrite non-degraded rows (explicit opt-in)."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
_make_reading(energy_db, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
# Initial compute.
|
||
compute_period(energy_db, _T0)
|
||
energy_db.commit()
|
||
|
||
# Simulate "new" end reading with higher values.
|
||
_make_reading(energy_db, recorded_at=_T1 - timedelta(seconds=5),
|
||
d1="20020.0", d2="10020.0", r1="5000.0", r2="3000.0",
|
||
source_id=77)
|
||
energy_db.commit()
|
||
|
||
count = recompute_range(energy_db, _T0, _T1)
|
||
assert count == 1
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# The row must have been updated (import_cost changes because deltas changed).
|
||
assert row.degraded is False
|
||
|
||
def test_returns_count_of_written_periods(self, energy_db: Session) -> None:
|
||
"""recompute_range returns the number of periods actually written."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
|
||
# Add readings for two consecutive periods: [10:00,10:15), [10:15,10:30).
|
||
_make_reading(energy_db, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
t2 = _ts(10, 30)
|
||
_make_reading(energy_db, recorded_at=t2,
|
||
d1=str(float(_END_D1) + 0.5),
|
||
d2=str(float(_END_D2) + 1.2),
|
||
r1=_END_R1, r2=str(float(_END_R2) + 0.1),
|
||
source_id=3)
|
||
energy_db.commit()
|
||
|
||
count = recompute_range(energy_db, _T0, t2)
|
||
assert count == 2
|
||
|
||
def test_idempotent_on_multiple_calls(self, energy_db: Session) -> None:
|
||
"""Calling recompute_range twice must not create duplicate rows."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
_make_reading(energy_db, recorded_at=_T0, d1=_START_D1, d2=_START_D2,
|
||
r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_T1, d1=_END_D1, d2=_END_D2,
|
||
r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
recompute_range(energy_db, _T0, _T1)
|
||
recompute_range(energy_db, _T0, _T1)
|
||
|
||
rows = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalars().all()
|
||
assert len(rows) == 1
|
||
|
||
def test_recompute_downgrades_successful_row_when_readings_disappear(
|
||
self, energy_db: Session
|
||
) -> None:
|
||
"""recompute_range must downgrade a previously successful row to degraded
|
||
when boundary readings no longer exist.
|
||
|
||
Scenario (reproduces REWORK 1 from reviewer probe3.py):
|
||
1. Compute period [10:00, 10:15) successfully — row is non-degraded, costs > 0.
|
||
2. Delete both boundary readings (simulate data loss / correction).
|
||
3. Call recompute_range over that window (overwrite=True path).
|
||
4. Row must now be degraded=True, all cost/kWh fields = 0,
|
||
contract_version_id = None, pricing = {}.
|
||
|
||
This verifies the "缺读数→degraded" contract holds even for the explicit
|
||
recompute path, i.e. stale successful values are never silently preserved.
|
||
"""
|
||
# Step 1: successful compute.
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
start_reading = _make_reading(
|
||
energy_db, recorded_at=_T0,
|
||
d1=_START_D1, d2=_START_D2, r1=_START_R1, r2=_START_R2, source_id=1,
|
||
)
|
||
end_reading = _make_reading(
|
||
energy_db, recorded_at=_T1,
|
||
d1=_END_D1, d2=_END_D2, r1=_END_R1, r2=_END_R2, source_id=2,
|
||
)
|
||
energy_db.commit()
|
||
|
||
compute_period(energy_db, _T0)
|
||
energy_db.commit()
|
||
|
||
row_initial = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
# Pre-condition: row is successful with non-zero import_cost.
|
||
assert row_initial.degraded is False
|
||
assert row_initial.import_cost > 0, "Pre-condition: import_cost must be non-zero"
|
||
assert row_initial.contract_version_id is not None
|
||
|
||
# Step 2: delete the boundary readings to simulate data loss.
|
||
energy_db.delete(start_reading)
|
||
energy_db.delete(end_reading)
|
||
energy_db.commit()
|
||
|
||
# Step 3: explicit recompute (overwrite=True path).
|
||
count = recompute_range(energy_db, _T0, _T1)
|
||
assert count == 1, "recompute_range must report one period written (degraded)"
|
||
|
||
# Step 4: row must now reflect degraded state — no stale costs.
|
||
energy_db.expire_all()
|
||
row_after = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row_after.degraded is True
|
||
assert row_after.import_cost == 0.0
|
||
assert row_after.export_revenue == 0.0
|
||
assert row_after.net_cost == 0.0
|
||
assert row_after.d1_kwh == 0.0
|
||
assert row_after.d2_kwh == 0.0
|
||
assert row_after.r1_kwh == 0.0
|
||
assert row_after.r2_kwh == 0.0
|
||
assert row_after.contract_version_id is None
|
||
assert row_after.pricing == {}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Bug-fix regression tests: register_at freshness + recompute_range future guard
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
class TestRegisterAtFreshness:
|
||
"""Tests for the staleness guard added to register_at (Fix A).
|
||
|
||
Normal DSMR readings arrive every ~10 seconds. A reading more than
|
||
``_READING_MAX_STALENESS`` (15 minutes) old at a given boundary is treated
|
||
as absent so the period is marked degraded rather than producing a
|
||
zero-delta fake-success row.
|
||
"""
|
||
|
||
def test_stale_reading_returns_none(self, energy_db: Session) -> None:
|
||
"""A reading older than 15 min before the boundary must be rejected."""
|
||
reading_time = _ts(10, 0) # 10:00
|
||
boundary = _ts(10, 30) # 10:30 — 30 min later (> 15 min staleness)
|
||
_make_reading(energy_db, recorded_at=reading_time, source_id=1)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, boundary)
|
||
assert result is None, (
|
||
"register_at must return None when the closest reading is more than "
|
||
"15 minutes before the boundary"
|
||
)
|
||
|
||
def test_fresh_reading_within_window_returned(self, energy_db: Session) -> None:
|
||
"""A reading within 15 min of the boundary must be returned normally."""
|
||
reading_time = _ts(10, 5) # 10:05
|
||
boundary = _ts(10, 15) # 10:15 — only 10 min gap (within window)
|
||
_make_reading(energy_db, recorded_at=reading_time,
|
||
d1="30000.0", d2="15000.0", r1="1000.0", r2="500.0",
|
||
source_id=1)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, boundary)
|
||
assert result is not None, (
|
||
"register_at must return the reading when it is within the 15-min staleness window"
|
||
)
|
||
assert result["d1"] == Decimal("30000.0")
|
||
|
||
def test_exact_15min_boundary_is_accepted(self, energy_db: Session) -> None:
|
||
"""A reading exactly 15 min before the boundary sits at the edge — accepted."""
|
||
reading_time = _ts(10, 0) # 10:00
|
||
boundary = _ts(10, 15) # 10:15 — exactly 15 min gap
|
||
_make_reading(energy_db, recorded_at=reading_time,
|
||
d1="40000.0", d2="20000.0", r1="2000.0", r2="1000.0",
|
||
source_id=1)
|
||
energy_db.commit()
|
||
|
||
result = register_at(energy_db, boundary)
|
||
# boundary - reading == 15 min == staleness limit → NOT stale (strict <)
|
||
assert result is not None
|
||
|
||
def test_future_boundary_returns_none(self, energy_db: Session) -> None:
|
||
"""A far-future boundary must return None (not the latest historical reading).
|
||
|
||
This is the root cause of the production bug: without the freshness
|
||
guard, register_at(far_future) returned the last available reading,
|
||
causing both period boundaries to resolve to the same row and producing
|
||
a zero-delta fake-success period.
|
||
"""
|
||
# Insert a reading at a realistic "now" time.
|
||
reading_time = _ts(10, 0) # 10:00 on 2026-06-23
|
||
_make_reading(energy_db, recorded_at=reading_time, source_id=1)
|
||
energy_db.commit()
|
||
|
||
# Query with a far-future boundary (e.g. end of month, same day +8 hours)
|
||
far_future_boundary = _ts(18, 0) # 18:00 — 8 hours later
|
||
result = register_at(energy_db, far_future_boundary)
|
||
assert result is None, (
|
||
"register_at must return None for a far-future boundary rather than "
|
||
"the latest historical reading"
|
||
)
|
||
|
||
|
||
class TestFuturePeriodDegraded:
|
||
"""Ensure that a period whose boundaries fall entirely outside DSMR data
|
||
is written as degraded=True (not as a fake zero-cost success).
|
||
|
||
This covers the production scenario where recompute_range was called with
|
||
a future end date, causing compute_period to be called on not-yet-closed
|
||
periods. With Fix A (register_at freshness) the boundary lookups return
|
||
None → degraded; with Fix B (recompute_range t1<=now guard) such periods
|
||
are skipped entirely. This test exercises the Fix A path directly via
|
||
compute_period(overwrite=True) on a future-like period with no nearby data.
|
||
"""
|
||
|
||
def test_out_of_range_period_is_degraded(self, energy_db: Session) -> None:
|
||
"""A period whose boundaries have no fresh DSMR readings → degraded=True.
|
||
|
||
We insert a reading at 10:00 and call compute_period for a period
|
||
starting at 14:00 (4 hours later). The boundary readings (14:00 and
|
||
14:15) are both more than 15 min from any available data, so
|
||
register_at returns None for both → degraded row written.
|
||
"""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# Only a reading at 10:00 — far from the 14:00/14:15 boundaries.
|
||
_make_reading(energy_db, recorded_at=_ts(10, 0), source_id=1)
|
||
energy_db.commit()
|
||
|
||
future_t0 = _ts(14, 0)
|
||
result = compute_period(energy_db, future_t0, overwrite=True)
|
||
assert result is True # a record was written
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == future_t0)
|
||
).scalar_one()
|
||
assert row.degraded is True, (
|
||
"compute_period must write degraded=True when both boundaries lack fresh readings"
|
||
)
|
||
assert row.import_cost == 0.0
|
||
assert row.net_cost == 0.0
|
||
|
||
|
||
class TestRecomputeRangeNoFuturePeriods:
|
||
"""Verify that recompute_range never writes periods whose t1 > now (Fix B)."""
|
||
|
||
def test_recompute_range_skips_future_periods(self, energy_db: Session) -> None:
|
||
"""When end is in the future, recompute_range must not write any future rows.
|
||
|
||
This is the direct test for Fix B. We call recompute_range with a
|
||
future end datetime; the only periods that may be written are those
|
||
where t1 <= now. No row with period_start > now should appear in the DB.
|
||
"""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
# No readings needed — we only care that future rows are NOT written.
|
||
energy_db.commit()
|
||
|
||
now = datetime.now(UTC)
|
||
far_future_end = now + timedelta(days=7)
|
||
# Use a past start so there are some candidate periods.
|
||
past_start = floor_to_quarter(now - timedelta(minutes=30))
|
||
|
||
recompute_range(energy_db, past_start, far_future_end)
|
||
|
||
# Assert no row has period_start beyond now.
|
||
rows = energy_db.execute(select(EnergyCostPeriod)).scalars().all()
|
||
future_rows = [
|
||
r for r in rows
|
||
if (r.period_start if r.period_start.tzinfo else r.period_start.replace(tzinfo=UTC)) > now
|
||
]
|
||
assert future_rows == [], (
|
||
f"recompute_range must not write future periods; "
|
||
f"found {len(future_rows)} row(s) with period_start > now"
|
||
)
|
||
|
||
def test_recompute_range_chain_no_slot_squatting(self, energy_db: Session) -> None:
|
||
"""Simulate the full production failure scenario and verify it is fixed.
|
||
|
||
Scenario:
|
||
1. recompute_range is called with end=far_future (reproduces the bug trigger).
|
||
2. compute_closed_periods is then called (reproduces the normal tick).
|
||
3. After the fix, step 1 must not have written any future rows, so the
|
||
tick in step 2 is free to compute periods normally.
|
||
|
||
We only verify that no future rows exist after step 1 (Fix B), because
|
||
that is the core slot-squatting prevention.
|
||
"""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
# Contract effective from far past so all periods in scope have a version.
|
||
_make_version(
|
||
energy_db, contract, _MANUAL_VALUES,
|
||
effective_from=datetime(2020, 1, 1, tzinfo=UTC),
|
||
)
|
||
energy_db.commit()
|
||
|
||
now = datetime.now(UTC)
|
||
far_future = now + timedelta(days=30)
|
||
past_start = floor_to_quarter(now - timedelta(hours=1))
|
||
|
||
# Step 1: buggy call with future end.
|
||
recompute_range(energy_db, past_start, far_future)
|
||
|
||
# Verify no future rows were squatted.
|
||
all_rows = energy_db.execute(select(EnergyCostPeriod)).scalars().all()
|
||
future_rows = [
|
||
r for r in all_rows
|
||
if (r.period_start if r.period_start.tzinfo else r.period_start.replace(tzinfo=UTC)) > now
|
||
]
|
||
assert future_rows == [], (
|
||
f"After recompute_range with future end, found {len(future_rows)} future row(s). "
|
||
"These would block the real tick from computing those periods."
|
||
)
|
||
|
||
|
||
class TestNormalPeriodsUnaffected:
|
||
"""Regression: the freshness guard must not break computation of normal
|
||
closed periods where readings are available close to the boundaries.
|
||
"""
|
||
|
||
def test_normal_closed_period_still_computes_ok(self, energy_db: Session) -> None:
|
||
"""A normal period with readings seconds before each boundary → non-degraded."""
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _MANUAL_VALUES, effective_from=_ts(0, 0))
|
||
|
||
# Readings placed very close to boundaries (as DSMR normally delivers them).
|
||
# t0=10:00, reading at 09:59:50 (10 s before); t1=10:15, reading at 10:14:55 (5 s before)
|
||
_make_reading(energy_db, recorded_at=_ts(9, 59, 50),
|
||
d1=_START_D1, d2=_START_D2, r1=_START_R1, r2=_START_R2, source_id=1)
|
||
_make_reading(energy_db, recorded_at=_ts(10, 14, 55),
|
||
d1=_END_D1, d2=_END_D2, r1=_END_R1, r2=_END_R2, source_id=2)
|
||
energy_db.commit()
|
||
|
||
result = compute_period(energy_db, _T0)
|
||
assert result is True
|
||
|
||
row = energy_db.execute(
|
||
select(EnergyCostPeriod).where(EnergyCostPeriod.period_start == _T0)
|
||
).scalar_one()
|
||
assert row.degraded is False, (
|
||
"A normal period with readings just before each boundary must not be degraded"
|
||
)
|
||
# import_cost matches hand-calc: same deltas as standard scenario
|
||
assert abs(row.import_cost - 0.4101) < 1e-6
|