tliu93
8b91146c29
- Store UTC, compute day boundaries in server local timezone (new app/services/timezone.py). - summarize: fixed fee / tax credit accrue only for elapsed whole local days, integrated across contract versions — no future-day inflation, no version-switch collapse. - MQTT import_cost_total / export_revenue_total getters delegate to summarize (no inline apportionment), anchored at the active contract's earliest version effective_from so the total sensors stay monotonic and never jump backward. - effective_from: naive datetimes interpreted as server-local wall-clock, then stored UTC. - frontend ContractForm sends local-midnight naive datetime (was UTC midnight). - get_costs_summary default window uses server-local "today".
1737 lines
72 KiB
Python
1737 lines
72 KiB
Python
"""Tests for M6-T07 + FU10: 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`` (Principle C): fixed costs & credits count only elapsed whole
|
||
*local* calendar days, cross-version. Sub-day windows → 0 fixed/credit.
|
||
One whole local day → full rate. Future dates → 0. Cross-version → sum.
|
||
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,
|
||
)
|
||
from app.services import timezone as tz_module
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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_zero_for_sub_day_window(self, energy_db: Session) -> None:
|
||
"""Principle C: a 30-minute window within the same local day counts 0 whole days.
|
||
|
||
Fixed costs are only added for elapsed whole local calendar days.
|
||
A window [10:00, 10:30) UTC stays within the same local calendar date
|
||
regardless of timezone offset (any UTC-11..UTC+14 range), so no whole
|
||
day is covered → fixed_costs = 0.
|
||
"""
|
||
from zoneinfo import ZoneInfo
|
||
self._setup_two_periods(energy_db)
|
||
# Pin the local timezone so the test is deterministic on any CI host.
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch.object(
|
||
tz_module, "local_tz", return_value=ZoneInfo("Europe/Amsterdam")
|
||
):
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert result["fixed_costs"] == 0.0, (
|
||
"Sub-day window should contribute 0 whole-day fixed costs under Principle C"
|
||
)
|
||
|
||
def test_credits_zero_for_sub_day_window(self, energy_db: Session) -> None:
|
||
"""Principle C: a 30-minute window within the same local day counts 0 whole days for credits."""
|
||
from zoneinfo import ZoneInfo
|
||
self._setup_two_periods(energy_db)
|
||
with __import__("unittest.mock", fromlist=["patch"]).patch.object(
|
||
tz_module, "local_tz", return_value=ZoneInfo("Europe/Amsterdam")
|
||
):
|
||
result = summarize(energy_db, _ts(10, 0), _ts(10, 30))
|
||
assert result["credits"] == 0.0, (
|
||
"Sub-day window should contribute 0 whole-day credits under Principle C"
|
||
)
|
||
|
||
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.
|
||
|
||
Principle C: the window [2026-06-23 00:00 UTC, 2026-06-24 00:00 UTC) spans
|
||
exactly one local calendar day in Europe/Amsterdam (CEST=UTC+2: 02:00→02:00).
|
||
effective_from = June 23 00:00 UTC = June 23 02:00 CEST = local date June 23.
|
||
Today (2026-06-25) ≥ June 23, so the day is elapsed → 1 day counted.
|
||
"""
|
||
from zoneinfo import ZoneInfo
|
||
from unittest.mock import patch
|
||
|
||
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 UTC day (no periods in DB — only standing/credits).
|
||
# Pin timezone to Europe/Amsterdam for deterministic local-date mapping.
|
||
start = datetime(2026, 6, 23, 0, 0, 0, tzinfo=_UTC)
|
||
end = datetime(2026, 6, 24, 0, 0, 0, tzinfo=_UTC)
|
||
|
||
with patch.object(tz_module, "local_tz", return_value=ZoneInfo("Europe/Amsterdam")):
|
||
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
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 7b. summarize — Principle C (whole-local-day counting, cross-version)
|
||
# ---------------------------------------------------------------------------
|
||
#
|
||
# All tests in this section pin the local timezone to Europe/Amsterdam via
|
||
# monkeypatch so assertions are deterministic regardless of CI host timezone.
|
||
# "Today" in these tests is 2026-06-25 (CEST = UTC+2).
|
||
#
|
||
# Reference rates:
|
||
# network_fee = 9.87 EUR/month, management_fee = 9.87 EUR/month
|
||
# daily_fixed = (9.87 + 9.87) / 30 = 0.658 EUR/day
|
||
# heffingskorting = 600.0 EUR/year
|
||
# daily_credit = 600.0 / 365 ≈ 1.6438... EUR/day
|
||
#
|
||
# Window semantics:
|
||
# start_local_date = local_date(start_utc) (inclusive)
|
||
# end_local_date = local_date(end_utc) (exclusive upper bound)
|
||
# cap = min(end_local_date, today_local + 1 day)
|
||
# counted_days = max(0, (cap - start_local_date).days)
|
||
# restricted to the version's [v_start, v_end) range
|
||
#
|
||
# CEST = UTC+2; local midnight June D = UTC June (D-1) 22:00.
|
||
# So:
|
||
# start "This Month" (June 1 local 00:00) = May 31 22:00 UTC
|
||
# end "Next Month" (July 1 local 00:00) = June 30 22:00 UTC
|
||
# In UTC we feed actual UTC boundaries.
|
||
#
|
||
# Table (today = June 25 local, effective_from = June 1 local 00:00 = May 31 22:00 UTC):
|
||
#
|
||
# | Window (UTC) | Local dates | Elapsed | Expected |
|
||
# |----------------------------------|-------------------------|---------|----------|
|
||
# | June 25 UTC+2 today (1 local day)| [June 25, June 26) | 1 day | 1 day |
|
||
# | June 25 UTC+2 → June 28 UTC+2 | [June 25, June 28) | 1 day* | 1 day |
|
||
# | June 1 CEST → July 1 CEST (mon) | [June 1, July 1) | 25 days*| 25 days |
|
||
# | May 1 → June 1 (all past) | [May 1, June 1) | 31 days | 31 days |
|
||
# | July 1 → Aug 1 (all future) | [July 1, Aug 1) | 0 days | 0 days |
|
||
#
|
||
# *today local = June 25: "today" counts but June 26+ does not.
|
||
|
||
_AMS = None # ZoneInfo resolved lazily
|
||
|
||
|
||
def _ams():
|
||
global _AMS
|
||
if _AMS is None:
|
||
from zoneinfo import ZoneInfo
|
||
_AMS = ZoneInfo("Europe/Amsterdam")
|
||
return _AMS
|
||
|
||
|
||
def _local_midnight_utc_ams(year: int, month: int, day: int) -> datetime:
|
||
"""Return UTC instant for Europe/Amsterdam local midnight on the given date."""
|
||
from zoneinfo import ZoneInfo
|
||
ams = ZoneInfo("Europe/Amsterdam")
|
||
return datetime(year, month, day, 0, 0, 0, tzinfo=ams).replace(tzinfo=None).replace(
|
||
tzinfo=ams
|
||
).astimezone(_UTC)
|
||
|
||
|
||
# Simpler helper using ZoneInfo directly:
|
||
def _ams_midnight(year: int, month: int, day: int) -> datetime:
|
||
"""UTC datetime corresponding to Europe/Amsterdam local midnight on year/month/day."""
|
||
from zoneinfo import ZoneInfo
|
||
ams = ZoneInfo("Europe/Amsterdam")
|
||
return datetime(year, month, day, 0, 0, 0, tzinfo=ams).astimezone(_UTC)
|
||
|
||
|
||
class TestSummarizePrincipleC:
|
||
"""Principle C whole-day counting with pinned Europe/Amsterdam timezone."""
|
||
|
||
# Effective_from: June 1 CEST local midnight = May 31 22:00 UTC.
|
||
# Today local = June 25 CEST (since today is 2026-06-25).
|
||
|
||
def _make_single_version_contract(self, session: Session, *, effective_from_utc: datetime) -> None:
|
||
"""Create an active manual contract with a single version."""
|
||
contract = _make_contract(session, kind="manual", active=True)
|
||
_make_version(session, contract, _MANUAL_VALUES, effective_from=effective_from_utc)
|
||
session.commit()
|
||
|
||
def _run_summarize_ams(self, session: Session, start_utc: datetime, end_utc: datetime) -> dict:
|
||
"""Run summarize with Europe/Amsterdam local_tz monkeypatched."""
|
||
from unittest.mock import patch
|
||
with patch.object(tz_module, "local_tz", return_value=_ams()):
|
||
return summarize(session, start_utc, end_utc)
|
||
|
||
def test_today_window_counts_1_day(self, energy_db: Session) -> None:
|
||
"""Today's window (local today 00:00 → local tomorrow 00:00) counts 1 day.
|
||
|
||
Matches table row: Today 6/25→6/26 → 1 day.
|
||
"""
|
||
eff_utc = _ams_midnight(2026, 6, 1)
|
||
self._make_single_version_contract(energy_db, effective_from_utc=eff_utc)
|
||
|
||
start = _ams_midnight(2026, 6, 25)
|
||
end = _ams_midnight(2026, 6, 26)
|
||
result = self._run_summarize_ams(energy_db, start, end)
|
||
|
||
daily_fixed = (9.87 + 9.87) / 30
|
||
daily_credit = 600.0 / 365
|
||
assert abs(result["fixed_costs"] - daily_fixed * 1) < 1e-9, (
|
||
f"Expected 1 day of fixed costs, got {result['fixed_costs']}"
|
||
)
|
||
assert abs(result["credits"] - daily_credit * 1) < 1e-9, (
|
||
f"Expected 1 day of credits, got {result['credits']}"
|
||
)
|
||
|
||
def test_future_window_counts_0_days(self, energy_db: Session) -> None:
|
||
"""A fully future window (all local dates > today) counts 0 days.
|
||
|
||
Matches table row: 7/1→8/1 (all future) → 0 days.
|
||
"""
|
||
eff_utc = _ams_midnight(2026, 6, 1)
|
||
self._make_single_version_contract(energy_db, effective_from_utc=eff_utc)
|
||
|
||
start = _ams_midnight(2026, 7, 1)
|
||
end = _ams_midnight(2026, 8, 1)
|
||
result = self._run_summarize_ams(energy_db, start, end)
|
||
|
||
assert result["fixed_costs"] == 0.0, (
|
||
f"All-future window must count 0 days; got fixed_costs={result['fixed_costs']}"
|
||
)
|
||
assert result["credits"] == 0.0, (
|
||
f"All-future window must count 0 days; got credits={result['credits']}"
|
||
)
|
||
|
||
def test_this_month_window_counts_25_days(self, energy_db: Session) -> None:
|
||
"""This Month (June 1 → July 1 local) counts 25 elapsed days (June 1..25).
|
||
|
||
Today = June 25 local, so June 26–30 are not yet elapsed.
|
||
Matches table row: 6/1→7/1 (This Month) → 25 days.
|
||
"""
|
||
eff_utc = _ams_midnight(2026, 6, 1)
|
||
self._make_single_version_contract(energy_db, effective_from_utc=eff_utc)
|
||
|
||
start = _ams_midnight(2026, 6, 1)
|
||
end = _ams_midnight(2026, 7, 1)
|
||
result = self._run_summarize_ams(energy_db, start, end)
|
||
|
||
daily_fixed = (9.87 + 9.87) / 30
|
||
daily_credit = 600.0 / 365
|
||
assert abs(result["fixed_costs"] - daily_fixed * 25) < 1e-9, (
|
||
f"Expected 25 days of fixed costs (June 1-25 elapsed), got {result['fixed_costs']}"
|
||
)
|
||
assert abs(result["credits"] - daily_credit * 25) < 1e-9, (
|
||
f"Expected 25 days of credits, got {result['credits']}"
|
||
)
|
||
|
||
def test_past_month_window_counts_all_31_days(self, energy_db: Session) -> None:
|
||
"""Last month (May 1 → June 1 local) counts all 31 days (fully past).
|
||
|
||
Matches table row: 5/1→6/1 (all past) → 31 days.
|
||
effective_from is June 1 so May is before the contract — 0 days from contract.
|
||
Use an earlier effective_from to cover May.
|
||
"""
|
||
eff_utc = _ams_midnight(2026, 1, 1) # contract starts Jan 1 → covers May
|
||
self._make_single_version_contract(energy_db, effective_from_utc=eff_utc)
|
||
|
||
start = _ams_midnight(2026, 5, 1)
|
||
end = _ams_midnight(2026, 6, 1)
|
||
result = self._run_summarize_ams(energy_db, start, end)
|
||
|
||
daily_fixed = (9.87 + 9.87) / 30
|
||
daily_credit = 600.0 / 365
|
||
assert abs(result["fixed_costs"] - daily_fixed * 31) < 1e-9, (
|
||
f"Expected 31 days of fixed costs (May 1-31 all past), got {result['fixed_costs']}"
|
||
)
|
||
assert abs(result["credits"] - daily_credit * 31) < 1e-9, (
|
||
f"Expected 31 days of credits, got {result['credits']}"
|
||
)
|
||
|
||
def test_partial_future_window_caps_at_today(self, energy_db: Session) -> None:
|
||
"""A window partially in the future caps at today (only elapsed days counted).
|
||
|
||
Window: June 25 → June 28 local (4 dates, but June 26/27 are future).
|
||
Today = June 25 → only 1 day elapsed (June 25).
|
||
Matches table row: 6/25→6/28 → 1 day.
|
||
"""
|
||
eff_utc = _ams_midnight(2026, 6, 1)
|
||
self._make_single_version_contract(energy_db, effective_from_utc=eff_utc)
|
||
|
||
start = _ams_midnight(2026, 6, 25)
|
||
end = _ams_midnight(2026, 6, 28)
|
||
result = self._run_summarize_ams(energy_db, start, end)
|
||
|
||
daily_fixed = (9.87 + 9.87) / 30
|
||
daily_credit = 600.0 / 365
|
||
assert abs(result["fixed_costs"] - daily_fixed * 1) < 1e-9, (
|
||
f"Expected 1 day of fixed costs (only June 25 elapsed), got {result['fixed_costs']}"
|
||
)
|
||
assert abs(result["credits"] - daily_credit * 1) < 1e-9, (
|
||
f"Expected 1 day of credits, got {result['credits']}"
|
||
)
|
||
|
||
def test_cross_version_integration(self, energy_db: Session) -> None:
|
||
"""Cross-version integration: V1 June 1-24, V2 June 25+ (today's boundary).
|
||
|
||
Window: This Month (June 1 → July 1 local) = 25 days total.
|
||
- V1: June 1-24 (24 days) at rate1
|
||
- V2: June 25-25 (1 day, today) at rate2
|
||
|
||
V1 rates: network_fee=6.0, management_fee=6.0 → daily = 12/30 = 0.4
|
||
heffingskorting=300 → daily = 300/365
|
||
V2 rates: network_fee=12.0, management_fee=12.0 → daily = 24/30 = 0.8
|
||
heffingskorting=600 → daily = 600/365
|
||
|
||
Expected:
|
||
fixed_costs = 24 × 0.4 + 1 × 0.8 = 9.6 + 0.8 = 10.4
|
||
credits = 24 × (300/365) + 1 × (600/365) = (7200+600)/365 = 7800/365
|
||
"""
|
||
_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": 6.0, "management_fee": 6.0},
|
||
"credits": {"heffingskorting": 300.0},
|
||
}
|
||
_VALUES_V2 = {
|
||
"energy": {"buy": {"normal": 0.20, "dal": 0.20}, "sell": {"normal": 0.08, "dal": 0.08},
|
||
"energy_tax": 0.0, "ode": 0.0},
|
||
"standing": {"network_fee": 12.0, "management_fee": 12.0},
|
||
"credits": {"heffingskorting": 600.0},
|
||
}
|
||
|
||
# V1: effective June 1 CEST local. V2: effective June 25 CEST local.
|
||
v1_from = _ams_midnight(2026, 6, 1)
|
||
v2_from = _ams_midnight(2026, 6, 25)
|
||
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _VALUES_V1, effective_from=v1_from,
|
||
effective_to=v2_from)
|
||
_make_version(energy_db, contract, _VALUES_V2, effective_from=v2_from)
|
||
energy_db.commit()
|
||
|
||
start = _ams_midnight(2026, 6, 1)
|
||
end = _ams_midnight(2026, 7, 1)
|
||
from unittest.mock import patch
|
||
with patch.object(tz_module, "local_tz", return_value=_ams()):
|
||
result = summarize(energy_db, start, end)
|
||
|
||
# V1: 24 days × (6+6)/30; V2: 1 day × (12+12)/30
|
||
expected_fixed = 24 * 12 / 30 + 1 * 24 / 30
|
||
expected_credits = 24 * 300 / 365 + 1 * 600 / 365
|
||
assert abs(result["fixed_costs"] - expected_fixed) < 1e-9, (
|
||
f"Cross-version fixed_costs wrong: expected {expected_fixed}, got {result['fixed_costs']}"
|
||
)
|
||
assert abs(result["credits"] - expected_credits) < 1e-9, (
|
||
f"Cross-version credits wrong: expected {expected_credits}, got {result['credits']}"
|
||
)
|
||
|
||
def test_version_switch_does_not_reset_counter(self, energy_db: Session) -> None:
|
||
"""Adding a new version does not reset the cumulative counter.
|
||
|
||
With V1 from June 1 and V2 from June 25, querying June 25 → July 1
|
||
still returns V2's 1-day rate without losing the historical V1 days
|
||
(the MQTT getter anchors at V1.effective_from to accumulate everything).
|
||
This test just checks that V2-only window returns V2 rate × 1 day.
|
||
"""
|
||
_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": 6.0, "management_fee": 6.0},
|
||
"credits": {"heffingskorting": 300.0},
|
||
}
|
||
_VALUES_V2 = {
|
||
"energy": {"buy": {"normal": 0.20, "dal": 0.20}, "sell": {"normal": 0.08, "dal": 0.08},
|
||
"energy_tax": 0.0, "ode": 0.0},
|
||
"standing": {"network_fee": 12.0, "management_fee": 12.0},
|
||
"credits": {"heffingskorting": 600.0},
|
||
}
|
||
v1_from = _ams_midnight(2026, 6, 1)
|
||
v2_from = _ams_midnight(2026, 6, 25)
|
||
|
||
contract = _make_contract(energy_db, kind="manual", active=True)
|
||
_make_version(energy_db, contract, _VALUES_V1, effective_from=v1_from,
|
||
effective_to=v2_from)
|
||
_make_version(energy_db, contract, _VALUES_V2, effective_from=v2_from)
|
||
energy_db.commit()
|
||
|
||
# Window = June 25 only (today, 1 day elapsed at V2 rate)
|
||
start = _ams_midnight(2026, 6, 25)
|
||
end = _ams_midnight(2026, 7, 1)
|
||
from unittest.mock import patch
|
||
with patch.object(tz_module, "local_tz", return_value=_ams()):
|
||
result = summarize(energy_db, start, end)
|
||
|
||
# Only 1 day at V2 rate
|
||
expected_fixed = 1 * 24 / 30
|
||
expected_credits = 1 * 600 / 365
|
||
assert abs(result["fixed_costs"] - expected_fixed) < 1e-9
|
||
assert abs(result["credits"] - expected_credits) < 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
|