tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity
Files
6f8cb05eab8c1d86573652b3ca7dcf44bdfe2e04
home-automation/tests/test_energy_cost.py
T

1418 lines
57 KiB
Python
Raw Normal View History

M6-T07: add metering billing engine + period job + summary + recompute
2026-06-23 22:26:56 +02:00
"""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 == {}
Billing fix: don't bill periods from stale/out-of-range DSMR readings
2026-06-24 12:44:05 +02:00
# ---------------------------------------------------------------------------
# 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
Reference in New Issue Copy Permalink