tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 17 Wiki Activity

M8-R00: align tariff expose tests with source bindings

Browse Source
This commit is contained in:
tliu93
2026-08-24 00:15:48 +02:00
parent ebf96de4f1
commit 9db7f63274
1 changed files with 81 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_energy_expose.py
+71 -27
View File
@@ -38,8 +38,8 @@ from zoneinfo import ZoneInfo
import pytest
from alembic import command
from alembic.config import Config
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
from sqlalchemy import create_engine, select
from sqlalchemy.orm import Session, sessionmaker
# ---------------------------------------------------------------------------
@@ -1426,25 +1426,59 @@ _DUAL_TARIFF_PRICING = {
@pytest.fixture()
def reset_tariff(monkeypatch):
"""Reset dsmr_ingest._current_tariff to None before/after each tariff test."""
def reset_tariff(energy_db, monkeypatch):
"""Isolate legacy and source-scoped DSMR tariff state for each tariff test."""
from app.services import dsmr_ingest as _di
monkeypatch.setattr(_di, "_current_tariff", None)
test_session_local = sessionmaker(bind=energy_db)
monkeypatch.setattr(_di, "get_session_local", lambda: test_session_local)
_di._current_tariff = None
with _di._tariff_lock:
_di._tariffs.clear()
try:
yield
finally:
_di._current_tariff = None
with _di._tariff_lock:
_di._tariffs.clear()
def _insert_manual_period(energy_db) -> None:
"""Insert an active electricity meter + non-degraded manual pricing period into energy_db.
def _bind_migrated_dsmr_source(session: Session, meter: Any, *, now: datetime) -> int:
"""Bind the migration-created DSMR source to an active electricity meter."""
from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel
source = session.scalar(select(MeterSource).where(MeterSource.kind == "dsmr_mqtt"))
assert source is not None, "Alembic head must provide the migrated DSMR source"
channel = session.scalar(
select(MeterSourceChannel).where(MeterSourceChannel.source_id == source.id)
)
assert channel is not None, "Alembic head must provide the migrated DSMR channel"
session.add(
MeterSourceBinding(
meter_id=meter.id,
channel_id=channel.id,
started_at=meter.started_at,
ended_at=None,
created_at=now,
updated_at=now,
)
)
session.flush()
return source.id
def _insert_manual_period(energy_db) -> int:
"""Insert a bound active electricity meter and manual pricing period.
FUE-T05: the energy-cost provider requires an active electricity meter, so
this helper inserts one alongside the period so that build_catalog returns
the energy entities.
this helper binds its migration-created DSMR source to that meter. Return
the binding's source id so tariff tests exercise source-scoped state.
"""
t0 = datetime(2026, 3, 1, 10, 0, tzinfo=timezone.utc)
meter_start = datetime(2026, 3, 1, 0, 0, tzinfo=timezone.utc)
with Session(energy_db) as session:
_make_active_meter(session, started_at=meter_start, label="Tariff Test Meter")
meter = _make_active_meter(session, started_at=meter_start, label="Tariff Test Meter")
source_id = _bind_migrated_dsmr_source(session, meter, now=t0)
_make_period(
session,
period_start=t0,
@@ -1455,6 +1489,7 @@ def _insert_manual_period(energy_db) -> None:
degraded=False,
)
session.commit()
return source_id
def test_buy_price_tariff_1_returns_dal(energy_db, reset_tariff) -> None:
@@ -1462,8 +1497,8 @@ def test_buy_price_tariff_1_returns_dal(energy_db, reset_tariff) -> None:
from app.integrations.expose import build_catalog
from app.services.dsmr_ingest import set_current_tariff
_insert_manual_period(energy_db)
set_current_tariff(1)
source_id = _insert_manual_period(energy_db)
set_current_tariff(source_id, 1)
with Session(energy_db) as session:
catalog = build_catalog(session)
@@ -1478,8 +1513,8 @@ def test_buy_price_tariff_2_returns_normal(energy_db, reset_tariff) -> None:
from app.integrations.expose import build_catalog
from app.services.dsmr_ingest import set_current_tariff
_insert_manual_period(energy_db)
set_current_tariff(2)
source_id = _insert_manual_period(energy_db)
set_current_tariff(source_id, 2)
with Session(energy_db) as session:
catalog = build_catalog(session)
@@ -1494,8 +1529,8 @@ def test_buy_price_tariff_none_falls_back_to_normal(energy_db, reset_tariff) ->
from app.integrations.expose import build_catalog
from app.services.dsmr_ingest import set_current_tariff
_insert_manual_period(energy_db)
set_current_tariff(None) # explicitly None (no tariff received)
source_id = _insert_manual_period(energy_db)
set_current_tariff(source_id, None) # explicitly None (no tariff received)
with Session(energy_db) as session:
catalog = build_catalog(session)
@@ -1510,8 +1545,8 @@ def test_sell_price_tariff_1_returns_dal(energy_db, reset_tariff) -> None:
from app.integrations.expose import build_catalog
from app.services.dsmr_ingest import set_current_tariff
_insert_manual_period(energy_db)
set_current_tariff(1)
source_id = _insert_manual_period(energy_db)
set_current_tariff(source_id, 1)
with Session(energy_db) as session:
catalog = build_catalog(session)
@@ -1526,8 +1561,8 @@ def test_sell_price_tariff_2_returns_normal(energy_db, reset_tariff) -> None:
from app.integrations.expose import build_catalog
from app.services.dsmr_ingest import set_current_tariff
_insert_manual_period(energy_db)
set_current_tariff(2)
source_id = _insert_manual_period(energy_db)
set_current_tariff(source_id, 2)
with Session(energy_db) as session:
catalog = build_catalog(session)
@@ -1542,8 +1577,8 @@ def test_sell_price_tariff_none_falls_back_to_normal(energy_db, reset_tariff) ->
from app.integrations.expose import build_catalog
from app.services.dsmr_ingest import set_current_tariff
_insert_manual_period(energy_db)
set_current_tariff(None)
source_id = _insert_manual_period(energy_db)
set_current_tariff(source_id, None)
with Session(energy_db) as session:
catalog = build_catalog(session)
@@ -1573,7 +1608,8 @@ def test_tibber_buy_price_not_affected_by_tariff(energy_db, reset_tariff) -> Non
}
with Session(energy_db) as session:
_make_active_meter(session, started_at=meter_start)
meter = _make_active_meter(session, started_at=meter_start)
source_id = _bind_migrated_dsmr_source(session, meter, now=t0)
_make_period(
session,
period_start=t0,
@@ -1584,7 +1620,7 @@ def test_tibber_buy_price_not_affected_by_tariff(energy_db, reset_tariff) -> Non
# Tibber pricing must return the same value regardless of tariff.
for tariff_val in (1, 2, None):
set_current_tariff(tariff_val)
set_current_tariff(source_id, tariff_val)
with Session(energy_db) as session:
catalog = build_catalog(session)
buy_entry = next(e for e in catalog if e.entity.key == "energy.buy_price_now")
@@ -1614,7 +1650,8 @@ def test_tibber_sell_price_not_affected_by_tariff(energy_db, reset_tariff) -> No
}
with Session(energy_db) as session:
_make_active_meter(session, started_at=meter_start)
meter = _make_active_meter(session, started_at=meter_start)
source_id = _bind_migrated_dsmr_source(session, meter, now=t0)
_make_period(
session,
period_start=t0,
@@ -1624,7 +1661,7 @@ def test_tibber_sell_price_not_affected_by_tariff(energy_db, reset_tariff) -> No
session.commit()
for tariff_val in (1, 2, None):
set_current_tariff(tariff_val)
set_current_tariff(source_id, tariff_val)
with Session(energy_db) as session:
catalog = build_catalog(session)
sell_entry = next(e for e in catalog if e.entity.key == "energy.sell_price_now")
@@ -2798,8 +2835,10 @@ def test_m8_meter_today_uses_current_binding_and_never_invents_zero(energy_db) -
"""A binding is a half-open cumulative epoch, not merely a channel filter."""
from app.integrations.expose import build_catalog
from app.models.meter_source import WarmteLinkReading
from app.services import timezone as tz
now = datetime.now(timezone.utc)
now = datetime(2026, 1, 15, 10, tzinfo=timezone.utc)
local_tz = ZoneInfo("Europe/Amsterdam")
with Session(energy_db) as session:
_source, channel, meter = _make_thermal_source_and_meter(session, "heating", now)
binding = meter.source_bindings[0]
@@ -2816,6 +2855,11 @@ def test_m8_meter_today_uses_current_binding_and_never_invents_zero(energy_db) -
equipment_fingerprint="test"),
))
session.commit()
with (
patch("app.integrations.expose._utc_now", return_value=now),
patch.object(tz, "local_tz", return_value=local_tz),
patch.object(tz, "local_now", return_value=now.astimezone(local_tz)),
):
entity = next(item.entity for item in build_catalog(session)
if item.entity.key == f"meter.{meter.uuid}.today")
assert entity.value_getter(session) == Decimal("5")