"""Tests for M6-T08 + FU6 + FU11: energy_cost expose provider + state publish. Coverage: 1. build_catalog contains all 6 energy_cost entities (4 original + 2 daily). 2. Cumulative entities (import_cost_total / export_revenue_total) have state_class="total" and device_class="monetary". 3. Daily entities (import_cost_today / export_revenue_today) have state_class="total_increasing" and device_class="monetary". 4. All 6 energy_cost entities default to enabled=False (no toggle row). 5. value_getter: buy/sell price from pricing snapshot for tibber and manual kinds. 6. value_getter: cumulative SUM(import_cost) / SUM(export_revenue) from non-degraded rows; degraded rows excluded. 7. value_getter returns None when no non-degraded period exists. 8. MQTT not enabled → publish_states is a no-op (no raises, no publish calls). 9. Integration: build_discovery_payload on an energy_cost entity does NOT raise IndexError (validates 2-element identifiers). 10. Keys are stable fixed strings (not derived from mutable data or DB ids). 11. Provider registered: energy_cost entities appear alongside modbus entities in the full catalog. 12. Standing charges prorated into import_cost_total getter. 13. Tax credit (heffingskorting) prorated into export_revenue_total getter. 14. effective_from in future → standing/credit cumulative = 0. 15. No active contract → getters fall back to pure SUM (cumulative) or None (daily). 16. FU11: anchor = max(contract_start, recording_start) — pre-recording fixed costs excluded. 17. FU11: daily getter uses today's local window; returns None without active contract. 18. FU11: daily getter None when no non-degraded periods. """ from __future__ import annotations from datetime import datetime, timezone from pathlib import Path from typing import Any from unittest.mock import MagicMock, patch import pytest from alembic import command from alembic.config import Config from sqlalchemy import create_engine from sqlalchemy.orm import Session # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _make_app_alembic_config(database_url: str) -> Config: cfg = Config("alembic_app.ini") cfg.set_main_option("sqlalchemy.url", database_url) return cfg def _make_period( session: Session, *, period_start: datetime, import_cost: float = 0.10, export_revenue: float = 0.05, net_cost: float = 0.05, currency: str = "EUR", pricing: dict | None = None, degraded: bool = False, d1_kwh: float = 0.5, d2_kwh: float = 0.3, r1_kwh: float = 0.1, r2_kwh: float = 0.1, ) -> Any: """Insert an EnergyCostPeriod row and return it (session not committed).""" from app.models.energy import EnergyCostPeriod now = datetime.now(tz=timezone.utc) p = EnergyCostPeriod( period_start=period_start, d1_kwh=d1_kwh, d2_kwh=d2_kwh, r1_kwh=r1_kwh, r2_kwh=r2_kwh, import_cost=import_cost, export_revenue=export_revenue, net_cost=net_cost, currency=currency, pricing=pricing or {}, contract_version_id=None, degraded=degraded, computed_at=now, ) session.add(p) session.flush() return p def _make_active_meter( session: Session, *, started_at: datetime, label: str = "Test Meter", commodity: str = "electricity", reason: str = "initial", ) -> Any: """Insert an active (ended_at=None) Meter row and flush. Helper for M7-T04 tests: every cumulative-getter test that expects a non-None return value must declare an active electricity meter so the D2 anchor (meter.started_at) can be resolved. """ from app.models.energy import Meter now = datetime.now(tz=timezone.utc) m = Meter( label=label, commodity=commodity, started_at=started_at, ended_at=None, reason=reason, note=None, created_at=now, ) session.add(m) session.flush() return m def _make_settings( *, mqtt_enabled: bool = True, ha_discovery_enabled: bool = True, ha_discovery_prefix: str = "homeassistant", ha_state_topic_prefix: str = "home_automation", ) -> MagicMock: s = MagicMock() s.mqtt_enabled = mqtt_enabled s.ha_discovery_enabled = ha_discovery_enabled s.ha_discovery_prefix = ha_discovery_prefix s.ha_state_topic_prefix = ha_state_topic_prefix return s def _make_mock_manager(*, is_connected: bool = True) -> MagicMock: mgr = MagicMock() mgr.is_connected = is_connected return mgr # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def energy_db(tmp_path: Path): """Temporary SQLite DB at Alembic head for energy expose tests.""" db_path = tmp_path / "energy_expose_test.db" db_url = f"sqlite:///{db_path}" command.upgrade(_make_app_alembic_config(db_url), "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) yield engine engine.dispose() # --------------------------------------------------------------------------- # 1. build_catalog contains all 4 energy_cost entities # --------------------------------------------------------------------------- def test_build_catalog_contains_6_energy_cost_entities(energy_db) -> None: """build_catalog must include all 6 energy_cost sensor entities (4 original + 2 daily). FUE-T05: provider requires an active electricity meter. Insert one so the provider produces its 6 entities. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) energy_keys = {e.entity.key for e in catalog if e.entity.key.startswith("energy.")} expected_keys = { "energy.buy_price_now", "energy.sell_price_now", "energy.import_cost_total", "energy.export_revenue_total", "energy.import_cost_today", "energy.export_revenue_today", } assert expected_keys == energy_keys, ( f"Expected energy keys {expected_keys!r}, got {energy_keys!r}" ) # --------------------------------------------------------------------------- # 2. Cumulative entities have correct state_class and device_class # --------------------------------------------------------------------------- def test_cumulative_entities_have_total_state_class(energy_db) -> None: """import_cost_total and export_revenue_total must have state_class='total'. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) cumulative_keys = {"energy.import_cost_total", "energy.export_revenue_total"} for entry in catalog: if entry.entity.key in cumulative_keys: assert entry.entity.state_class == "total", ( f"Entity {entry.entity.key!r} must have state_class='total', " f"got {entry.entity.state_class!r}" ) def test_cumulative_entities_have_monetary_device_class(energy_db) -> None: """import_cost_total and export_revenue_total must have device_class='monetary'. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) cumulative_keys = {"energy.import_cost_total", "energy.export_revenue_total"} for entry in catalog: if entry.entity.key in cumulative_keys: assert entry.entity.device_class == "monetary", ( f"Entity {entry.entity.key!r} must have device_class='monetary', " f"got {entry.entity.device_class!r}" ) def test_all_energy_entities_are_sensors(energy_db) -> None: """All 6 energy_cost entities must have component='sensor'. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) energy_entries = [e for e in catalog if e.entity.key.startswith("energy.")] assert len(energy_entries) == 6 for entry in energy_entries: assert entry.entity.component == "sensor", ( f"Expected component='sensor' for {entry.entity.key!r}, " f"got {entry.entity.component!r}" ) # --------------------------------------------------------------------------- # 3. Default enabled=False (no toggle row) # --------------------------------------------------------------------------- def test_energy_cost_entities_default_to_disabled(energy_db) -> None: """All 6 energy_cost entities must default to enabled=False (no toggle row). FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) energy_entries = [e for e in catalog if e.entity.key.startswith("energy.")] assert len(energy_entries) == 6 for entry in energy_entries: assert entry.enabled is False, ( f"Entity {entry.entity.key!r} must default to enabled=False " f"when no toggle row exists, got enabled={entry.enabled!r}" ) # --------------------------------------------------------------------------- # 4. value_getter: current price from tibber pricing snapshot # --------------------------------------------------------------------------- def test_buy_price_getter_reads_tibber_snapshot(energy_db) -> None: """buy_price_now value_getter must return the 'buy' price from tibber snapshot. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 1, 1, 12, 0, tzinfo=timezone.utc) meter_start = datetime(2025, 1, 1, 0, 0, tzinfo=timezone.utc) tibber_pricing = { "kind": "tibber", "buy": "0.2850", "sell": "0.1200", "energy_tax": "0.1234", "sell_adjust": "0.0100", "total": "0.2850", "tibber_price_starts_at": t0.isoformat(), "tibber_price_id": 1, } with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period( session, period_start=t0, import_cost=0.10, export_revenue=0.05, currency="EUR", pricing=tibber_pricing, degraded=False, ) session.commit() 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" ) value = buy_entry.entity.value_getter(session) assert value == pytest.approx(0.2850), ( f"Expected buy price 0.2850, got {value!r}" ) def test_sell_price_getter_reads_tibber_snapshot(energy_db) -> None: """sell_price_now value_getter must return the 'sell' price from tibber snapshot. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 1, 1, 12, 15, tzinfo=timezone.utc) meter_start = datetime(2025, 1, 1, 0, 0, tzinfo=timezone.utc) tibber_pricing = { "kind": "tibber", "buy": "0.3100", "sell": "0.1500", "energy_tax": "0.1234", "sell_adjust": "0.0100", "total": "0.3100", "tibber_price_starts_at": t0.isoformat(), "tibber_price_id": 2, } with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period( session, period_start=t0, import_cost=0.12, export_revenue=0.07, currency="EUR", pricing=tibber_pricing, degraded=False, ) session.commit() 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" ) value = sell_entry.entity.value_getter(session) assert value == pytest.approx(0.1500), ( f"Expected sell price 0.1500, got {value!r}" ) def test_buy_price_getter_reads_manual_snapshot(energy_db) -> None: """buy_price_now value_getter must return 'buy_normal' from manual pricing snapshot. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 2, 1, 8, 0, tzinfo=timezone.utc) meter_start = datetime(2025, 2, 1, 0, 0, tzinfo=timezone.utc) manual_pricing = { "kind": "manual", "buy_dal": "0.2500", "buy_normal": "0.2700", "sell_dal": "0.0900", "sell_normal": "0.0950", "energy_tax": "0.1234", "ode": "0.0015", } with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period( session, period_start=t0, import_cost=0.08, export_revenue=0.03, currency="EUR", pricing=manual_pricing, degraded=False, ) session.commit() 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" ) value = buy_entry.entity.value_getter(session) # buy_normal is the representative buy price for manual strategy assert value == pytest.approx(0.2700), ( f"Expected buy_normal 0.2700, got {value!r}" ) def test_sell_price_getter_reads_manual_snapshot(energy_db) -> None: """sell_price_now value_getter must return 'sell_normal' from manual pricing snapshot. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 2, 1, 8, 15, tzinfo=timezone.utc) meter_start = datetime(2025, 2, 1, 0, 0, tzinfo=timezone.utc) manual_pricing = { "kind": "manual", "buy_dal": "0.2500", "buy_normal": "0.2700", "sell_dal": "0.0900", "sell_normal": "0.0950", "energy_tax": "0.1234", "ode": "0.0015", } with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period( session, period_start=t0, import_cost=0.09, export_revenue=0.04, currency="EUR", pricing=manual_pricing, degraded=False, ) session.commit() 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" ) value = sell_entry.entity.value_getter(session) # sell_normal is the representative sell price for manual strategy assert value == pytest.approx(0.0950), ( f"Expected sell_normal 0.0950, got {value!r}" ) # --------------------------------------------------------------------------- # 5. value_getter: cumulative SUM — degraded rows excluded # --------------------------------------------------------------------------- def test_import_cost_total_sums_non_degraded_rows(energy_db) -> None: """import_cost_total must return SUM of non-degraded import_cost values only. M7-T04 (D2): active electricity meter required so the getter can anchor on meter.started_at. The meter is started before t0 so both non-degraded rows fall within [anchor, now). """ from app.integrations.expose import build_catalog t0 = datetime(2025, 3, 1, 6, 0, tzinfo=timezone.utc) t1 = datetime(2025, 3, 1, 6, 15, tzinfo=timezone.utc) t2 = datetime(2025, 3, 1, 6, 30, tzinfo=timezone.utc) meter_start = datetime(2025, 3, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) # Two good rows: 0.10 + 0.20 = 0.30 _make_period(session, period_start=t0, import_cost=0.10, degraded=False) _make_period(session, period_start=t1, import_cost=0.20, degraded=False) # One degraded row (import_cost=0.0, should be excluded): _make_period(session, period_start=t2, import_cost=0.0, degraded=True) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # No active contract → fixed_costs = 0; only metered sum = 0.30. assert value == pytest.approx(0.30), ( f"Expected cumulative import_cost 0.30 (non-degraded only), got {value!r}" ) def test_export_revenue_total_sums_non_degraded_rows(energy_db) -> None: """export_revenue_total must return SUM of non-degraded export_revenue values only. M7-T04 (D2): active electricity meter required so the getter can anchor. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 3, 2, 6, 0, tzinfo=timezone.utc) t1 = datetime(2025, 3, 2, 6, 15, tzinfo=timezone.utc) t2 = datetime(2025, 3, 2, 6, 30, tzinfo=timezone.utc) meter_start = datetime(2025, 3, 2, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) # Two good rows: 0.05 + 0.07 = 0.12 _make_period(session, period_start=t0, export_revenue=0.05, degraded=False) _make_period(session, period_start=t1, export_revenue=0.07, degraded=False) # One degraded row (should be excluded): _make_period(session, period_start=t2, export_revenue=0.0, degraded=True) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) export_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_total" ) value = export_entry.entity.value_getter(session) # No active contract → credits = 0; only metered sum = 0.12. assert value == pytest.approx(0.12), ( f"Expected cumulative export_revenue 0.12 (non-degraded only), got {value!r}" ) def test_cumulative_getter_excludes_degraded_import_cost_row(energy_db) -> None: """A degraded row with non-zero import_cost must NOT contribute to the cumulative sum. This verifies the exclusion filter on degraded=True rows. (In practice compute_period writes 0.0 for degraded rows; but if a row was previously successful and then set degraded, its import_cost could be non-zero.) M7-T04 (D2): active electricity meter required. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 4, 1, 10, 0, tzinfo=timezone.utc) t1 = datetime(2025, 4, 1, 10, 15, tzinfo=timezone.utc) meter_start = datetime(2025, 4, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period(session, period_start=t0, import_cost=0.50, degraded=False) # A row that is degraded but somehow has a non-zero import_cost (edge case): _make_period(session, period_start=t1, import_cost=0.99, degraded=True) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # Only the non-degraded row should contribute: 0.50 (no contract → fixed_costs=0). assert value == pytest.approx(0.50), ( f"Degraded row must not be included in SUM; expected 0.50, got {value!r}" ) # --------------------------------------------------------------------------- # 6. value_getter returns None when no non-degraded period exists # --------------------------------------------------------------------------- def test_buy_price_getter_returns_none_with_no_periods(energy_db) -> None: """buy_price_now value_getter must return None when no non-degraded periods exist. FUE-T05: provider requires an active electricity meter. Insert one so the entity is present in the catalog; then verify the value_getter returns None (no periods → no pricing snapshot). """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() 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" ) value = buy_entry.entity.value_getter(session) assert value is None, f"Expected None with no periods, got {value!r}" def test_sell_price_getter_returns_none_with_no_periods(energy_db) -> None: """sell_price_now value_getter must return None when no non-degraded periods exist. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() 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" ) value = sell_entry.entity.value_getter(session) assert value is None, f"Expected None with no periods, got {value!r}" def test_import_cost_getter_returns_none_with_no_non_degraded_periods(energy_db) -> None: """import_cost_total value_getter must return None when only degraded rows exist. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2025, 5, 1, 0, 0, tzinfo=timezone.utc) meter_start = datetime(2025, 5, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) # Only a degraded row → SUM returns None (no rows to aggregate) _make_period(session, period_start=t0, import_cost=0.0, degraded=True) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) assert value is None, ( f"Expected None when only degraded rows exist, got {value!r}" ) def test_export_revenue_getter_returns_none_with_no_periods(energy_db) -> None: """export_revenue_total value_getter must return None when no periods at all. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) export_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_total" ) value = export_entry.entity.value_getter(session) assert value is None, f"Expected None with no periods, got {value!r}" # --------------------------------------------------------------------------- # 7. MQTT not enabled → publish_states is a no-op # --------------------------------------------------------------------------- def test_publish_states_noop_when_mqtt_disabled() -> None: """publish_states must be a no-op and not raise when MQTT is disabled.""" settings = _make_settings(mqtt_enabled=False, ha_discovery_enabled=True) mock_mgr = _make_mock_manager(is_connected=False) with ( patch("app.services.ha_discovery.build_runtime_settings", return_value=settings), patch("app.services.ha_discovery.mqtt_manager", mock_mgr), ): from app.services.ha_discovery import publish_states from sqlalchemy import create_engine as _ce eng = _ce("sqlite:///:memory:") with Session(eng) as session: publish_states(session) # must not raise mock_mgr.publish.assert_not_called() def test_publish_states_noop_when_ha_discovery_disabled() -> None: """publish_states must be a no-op when ha_discovery_enabled=False.""" settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=False) mock_mgr = _make_mock_manager(is_connected=True) with ( patch("app.services.ha_discovery.build_runtime_settings", return_value=settings), patch("app.services.ha_discovery.mqtt_manager", mock_mgr), ): from app.services.ha_discovery import publish_states from sqlalchemy import create_engine as _ce eng = _ce("sqlite:///:memory:") with Session(eng) as session: publish_states(session) # must not raise mock_mgr.publish.assert_not_called() def test_publish_states_noop_when_not_connected() -> None: """publish_states must be a no-op when MQTT is configured but not connected.""" settings = _make_settings(mqtt_enabled=True, ha_discovery_enabled=True) mock_mgr = _make_mock_manager(is_connected=False) with ( patch("app.services.ha_discovery.build_runtime_settings", return_value=settings), patch("app.services.ha_discovery.mqtt_manager", mock_mgr), ): from app.services.ha_discovery import publish_states from sqlalchemy import create_engine as _ce eng = _ce("sqlite:///:memory:") with Session(eng) as session: publish_states(session) # must not raise mock_mgr.publish.assert_not_called() # --------------------------------------------------------------------------- # 8. Integration: build_discovery_payload does not IndexError for energy entities # --------------------------------------------------------------------------- def test_build_discovery_payload_no_index_error_for_energy_entities(energy_db) -> None: """build_discovery_payload must NOT raise IndexError for energy_cost entities. FUE-T05: identifiers is now ('energy-cost', meter.uuid). Validates that the 2-element identifiers tuple satisfies ha_discovery.py's requirement to access identifiers[1] as node_id. """ from app.integrations.expose import build_catalog from app.services.ha_discovery import build_discovery_payload meter_uuid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" meter_label = "Test Meter" now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: from app.models.energy import Meter as _M m = _M( label=meter_label, commodity="electricity", started_at=now, ended_at=None, reason="initial", note=None, created_at=now, ) # Override the auto-generated uuid so we can assert on it deterministically. m.uuid = meter_uuid session.add(m) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) energy_entries = [e for e in catalog if e.entity.key.startswith("energy.")] assert len(energy_entries) == 6, "Expected 6 energy_cost entities in catalog" for entry in energy_entries: # identifiers[1] must be the meter uuid (not "energy-cost"). assert entry.entity.device.identifiers[1] == meter_uuid, ( f"identifiers[1] must be meter uuid {meter_uuid!r}, " f"got {entry.entity.device.identifiers[1]!r}" ) # Must not raise — specifically no IndexError from identifiers[1] topic, config = build_discovery_payload(entry.entity, "homeassistant") # node_id = identifiers[1] with hyphens → underscores node_id = meter_uuid.replace("-", "_") assert node_id in topic, ( f"Expected meter uuid node_id {node_id!r} in discovery topic, got {topic!r}" ) assert topic.endswith("/config"), ( f"Discovery topic must end with /config, got {topic!r}" ) assert "unique_id" in config # unique_id seed is identifiers[1] (meter uuid) + entity key assert meter_uuid in config["unique_id"], ( f"unique_id must contain meter uuid, got {config['unique_id']!r}" ) assert "device" in config assert "energy-cost" in config["device"]["identifiers"] assert meter_uuid in config["device"]["identifiers"] def test_energy_cost_entities_omit_availability_so_ha_shows_them(energy_db) -> None: """The energy-cost device has no online/offline heartbeat, so its discovery config must NOT declare an availability topic — otherwise HA marks the entities ``unavailable`` forever even though state is being published. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog from app.services.ha_discovery import build_discovery_payload now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) energy_entries = [e for e in catalog if e.entity.key.startswith("energy.")] assert len(energy_entries) == 6 for entry in energy_entries: assert entry.entity.device.provides_availability is False _, config = build_discovery_payload(entry.entity, "homeassistant") assert "availability" not in config, ( f"energy entity {entry.entity.key} must omit availability (always-available)" ) assert "availability_mode" not in config def test_energy_entity_discovery_topics_contain_correct_node_id() -> None: """Discovery topic node_id for energy entities must be derived from meter uuid. FUE-T05: identifiers[1] is now the active meter's uuid. ha_discovery._node_id() replaces hyphens with underscores in identifiers[1] to build the MQTT node_id. This test verifies that the topic reflects the meter uuid (not the old fixed 'energy-cost' string). """ from app.integrations.expose import DeviceInfo, ExposableEntity from app.services.ha_discovery import build_discovery_payload meter_uuid = "12345678-abcd-ef00-1234-567890abcdef" # identifiers[1] = meter uuid — this is what FUE-T05 sets. device = DeviceInfo( identifiers=("energy-cost", meter_uuid), name="Test Meter", provides_availability=False, ) entity = ExposableEntity( key="energy.import_cost_total", component="sensor", device=device, device_class="monetary", unit="EUR", name="Energy Import Cost (incl. standing)", state_class="total", ) topic, config = build_discovery_payload(entity, discovery_prefix="homeassistant") # node_id: identifiers[1] = meter_uuid, hyphens → underscores expected_node = meter_uuid.replace("-", "_") assert f"/{expected_node}/" in topic, ( f"Expected meter uuid node_id {expected_node!r} in topic {topic!r}" ) assert topic.startswith("homeassistant/sensor/"), ( f"Topic must start with homeassistant/sensor/, got {topic!r}" ) assert "state_class" in config assert config["state_class"] == "total" assert config.get("device_class") == "monetary" # unique_id must contain the meter uuid assert meter_uuid in config["unique_id"], ( f"unique_id must contain meter uuid, got {config['unique_id']!r}" ) # --------------------------------------------------------------------------- # 9. Key stability: fixed strings, not derived from mutable data # --------------------------------------------------------------------------- def test_energy_entity_keys_are_stable_fixed_strings(energy_db) -> None: """Entity keys must be fixed strings, not derived from DB ids or session state. FUE-T05: provider requires an active electricity meter. Keys remain the same stable strings (``energy.*``) regardless of which meter is active — only the DeviceInfo identifiers/name change, not the keys. """ from app.integrations.expose import build_catalog from datetime import timedelta t0 = datetime(2025, 6, 1, 0, 0, tzinfo=timezone.utc) t1 = t0 + timedelta(minutes=15) meter_start = datetime(2025, 6, 1, 0, 0, tzinfo=timezone.utc) # Insert active meter and first period with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start, label="Stable Key Meter") _make_period( session, period_start=t0, currency="EUR", pricing={"kind": "tibber", "buy": "0.25", "sell": "0.10"}, degraded=False, ) session.commit() with Session(energy_db) as session: catalog1 = build_catalog(session) keys1 = {e.entity.key for e in catalog1 if e.entity.key.startswith("energy.")} with Session(energy_db) as session: _make_period( session, period_start=t1, currency="EUR", pricing={"kind": "tibber", "buy": "0.30", "sell": "0.12"}, degraded=False, ) session.commit() with Session(energy_db) as session: catalog2 = build_catalog(session) keys2 = {e.entity.key for e in catalog2 if e.entity.key.startswith("energy.")} # Keys must be identical across both builds (price changed, key must not) assert keys1 == keys2, ( f"Entity keys must be stable across different data states. " f"First: {keys1!r}, Second: {keys2!r}" ) expected_keys = { "energy.buy_price_now", "energy.sell_price_now", "energy.import_cost_total", "energy.export_revenue_total", "energy.import_cost_today", "energy.export_revenue_today", } assert keys1 == expected_keys, ( f"Energy keys must be exactly {expected_keys!r}, got {keys1!r}" ) # --------------------------------------------------------------------------- # 10. Provider registered: energy_cost entities co-exist with modbus entities # --------------------------------------------------------------------------- def test_energy_provider_registered_alongside_modbus(energy_db) -> None: """Both modbus and energy_cost providers must be registered and produce entities. FUE-T05: energy-cost provider requires an active electricity meter. """ from app.models.modbus import ModbusDevice from app.integrations.expose import build_catalog now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now, label="Co-exist Electricity Meter") device = ModbusDevice( uuid="cccccccc-0000-0000-0000-000000000099", friendly_name="Co-exist Meter", host="10.0.0.1", port=502, unit_id=1, profile="sdm120", poll_interval_s=5, enabled=True, created_at=now, updated_at=now, ) session.add(device) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) all_keys = {e.entity.key for e in catalog} # Modbus entities present assert any(k.startswith("modbus.") for k in all_keys), ( "Expected modbus entities in catalog" ) # Energy entities present assert "energy.import_cost_total" in all_keys, ( "Expected energy.import_cost_total in catalog" ) assert "energy.export_revenue_total" in all_keys, ( "Expected energy.export_revenue_total in catalog" ) # --------------------------------------------------------------------------- # 11. Standing charges prorated into import_cost_total getter # --------------------------------------------------------------------------- def _make_contract_with_version( session: Session, *, kind: str = "manual", values: dict, effective_from: datetime, active: bool = True, ) -> None: """Insert an EnergyContract + one EnergyContractVersion and flush.""" from app.models.energy import EnergyContract, EnergyContractVersion now = datetime.now(timezone.utc) contract = EnergyContract( name=f"Test Contract ({kind})", kind=kind, currency="EUR", active=active, created_at=now, updated_at=now, ) session.add(contract) session.flush() version = EnergyContractVersion( contract_id=contract.id, effective_from=effective_from, effective_to=None, values=values, created_at=now, ) session.add(version) session.flush() _STANDING_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": 30.0, # EUR/month "management_fee": 60.0, # EUR/month → daily = (30+60)/30 = 3.0 }, "credits": { "heffingskorting": 365.0, # EUR/year → daily = 365/365 = 1.0 }, } def test_import_cost_total_includes_standing_charges(energy_db) -> None: """import_cost_total getter must add prorated standing charges from active contract. M7-T04 (D2): anchor = active electricity meter's started_at. The getter delegates to summarize(anchor → now), where anchor = meter.started_at. With network_fee=30 EUR/month + management_fee=60 EUR/month, daily_standing = 3.0 EUR/day. The getter uses summarize, which counts whole *local* elapsed days under Principle C. We pin the timezone to UTC for simplicity: local days = UTC days. Setup: - meter.started_at = effective_from = 10 full UTC days ago (exact UTC midnight). - First period placed AT meter.started_at → anchor = meter.started_at. - Under UTC pinned, Principle C counts days [meter.started_at.date(), today] = 11 days. """ from decimal import Decimal from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) # D2 anchor = meter.started_at = 10 UTC days ago at midnight meter_started_at = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=10) effective_from = meter_started_at # contract also starts at the same time # Under UTC timezone (pinned), Principle C counts whole days from meter.started_at.date() # to today (inclusive) = 11 days total (10 + today). expected_days = (now_utc.date() - meter_started_at.date()).days + 1 # = 11 import_cost_sum = 5.00 # EUR t0 = meter_started_at # first period at anchor t1 = t0 + timedelta(minutes=15) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_started_at) _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=effective_from, active=True, ) _make_period(session, period_start=t0, import_cost=3.00, degraded=False) _make_period(session, period_start=t1, import_cost=2.00, degraded=False) session.commit() with Session(energy_db) as session: # Pin to UTC so local days = UTC days (deterministic on any CI host). with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # daily_standing = (30 + 60) / 30 = 3.0 EUR/day daily_standing = Decimal("90") / Decimal("30") expected = float(Decimal(str(import_cost_sum)) + daily_standing * expected_days) assert value == pytest.approx(expected, rel=1e-6), ( f"Expected import_cost_total={expected} (sum={import_cost_sum} + " f"standing={float(daily_standing * expected_days)} over {expected_days} days), " f"got {value!r}" ) # --------------------------------------------------------------------------- # 12. Tax credit (heffingskorting) prorated into export_revenue_total getter # --------------------------------------------------------------------------- def test_export_revenue_total_includes_tax_credit(energy_db) -> None: """export_revenue_total getter must add prorated heffingskorting from active contract. M7-T04 (D2): anchor = active electricity meter's started_at. With heffingskorting=365 EUR/year, daily_credit = 365/365 = 1.0 EUR/day. Timezone pinned to UTC for deterministic local-day counting. Setup: meter.started_at = effective_from = 4 days ago → 5 days total (incl. today). """ from decimal import Decimal from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) meter_started_at = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=4) effective_from = meter_started_at # Under UTC pinned: expected_days = days from meter.started_at.date() to today inclusive. expected_days = (now_utc.date() - meter_started_at.date()).days + 1 # = 5 t0 = meter_started_at t1 = t0 + timedelta(minutes=15) export_sum = 2.50 # EUR with Session(energy_db) as session: _make_active_meter(session, started_at=meter_started_at) _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=effective_from, active=True, ) _make_period(session, period_start=t0, export_revenue=1.50, degraded=False) _make_period(session, period_start=t1, export_revenue=1.00, degraded=False) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) export_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_total" ) value = export_entry.entity.value_getter(session) # daily_credit = 365 / 365 = 1.0 EUR/day daily_credit = Decimal("365") / Decimal("365") expected = float(Decimal(str(export_sum)) + daily_credit * expected_days) assert value == pytest.approx(expected, rel=1e-6), ( f"Expected export_revenue_total={expected} (sum={export_sum} + " f"credit={float(daily_credit * expected_days)} over {expected_days} days), " f"got {value!r}" ) # --------------------------------------------------------------------------- # 13. effective_from in future → standing/credit cumulative = 0 # --------------------------------------------------------------------------- def test_import_cost_standing_zero_when_effective_from_in_future(energy_db) -> None: """When contract version effective_from is in the future, standing charges must be 0. M7-T04 (D2): anchor = active electricity meter's started_at (in the past). The summarize window [meter.started_at, now) DOES include the period data, but Principle C does not count future local days for fixed costs. Since the contract's effective_from is in the future, no days in [meter.started_at, today] fall under a valid contract version, so fixed_costs = 0. With the D2 anchor in the past, the metered sum IS returned (unlike the old FU11 design where the anchor was the future effective_from → empty window → 0). Expected: value = import_cost_sum + 0 (no standing days) """ from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) # Meter started 2 days ago; period is 1 day ago (within meter window). meter_started_at = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=2) future_effective = now_utc + timedelta(days=30) t0 = now_utc.replace(hour=6, minute=0, second=0, microsecond=0) - timedelta(days=1) import_cost_sum = 4.00 with Session(energy_db) as session: _make_active_meter(session, started_at=meter_started_at) _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=future_effective, active=True, ) _make_period(session, period_start=t0, import_cost=import_cost_sum, degraded=False) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # D2: anchor = meter.started_at (past) → metered_import = import_cost_sum. # Principle C: contract effective_from is in the future → 0 standing days → fixed_costs = 0. # Result = import_cost_sum + 0. assert value == pytest.approx(import_cost_sum, rel=1e-9), ( f"When effective_from is in future, standing = 0, value = metered sum; " f"expected {import_cost_sum}, got {value!r}" ) def test_export_revenue_credit_zero_when_effective_from_in_future(energy_db) -> None: """When contract version effective_from is in the future, tax credit must be 0. M7-T04 (D2): anchor = active electricity meter's started_at (in the past). Summarize window includes the period, but Principle C does not count future days, so credits = 0. Expected: value = export_sum + 0 credits. """ from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) meter_started_at = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=2) future_effective = now_utc + timedelta(days=60) t0 = now_utc.replace(hour=6, minute=0, second=0, microsecond=0) - timedelta(days=1) export_sum = 3.00 with Session(energy_db) as session: _make_active_meter(session, started_at=meter_started_at) _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=future_effective, active=True, ) _make_period(session, period_start=t0, export_revenue=export_sum, degraded=False) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) export_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_total" ) value = export_entry.entity.value_getter(session) # D2: metered_export = export_sum; Principle C: future effective_from → 0 credits. assert value == pytest.approx(export_sum, rel=1e-9), ( f"When effective_from is in future, credit = 0, value = metered sum; " f"expected {export_sum}, got {value!r}" ) # --------------------------------------------------------------------------- # 14. No active contract → getters fall back to pure SUM # --------------------------------------------------------------------------- def test_import_cost_total_returns_metered_sum_when_no_active_contract(energy_db) -> None: """When active meter exists but no active contract, import_cost_total returns metered sum. M7-T04 (D2): anchor = meter.started_at; summarize runs without a contract → fixed_costs = 0; the return value is the pure metered sum within the meter window. This is a behaviour change from FU11 (which returned a global SUM) — D2 only counts periods since meter.started_at. """ from app.integrations.expose import build_catalog t0 = datetime(2026, 1, 1, 6, 0, tzinfo=timezone.utc) t1 = datetime(2026, 1, 1, 6, 15, tzinfo=timezone.utc) meter_start = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) # Inactive contract — active=False _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=t0, active=False, ) _make_period(session, period_start=t0, import_cost=1.50, degraded=False) _make_period(session, period_start=t1, import_cost=2.50, degraded=False) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # D2: anchor = meter.started_at; both periods fall within [anchor, now]. # No active contract → fixed_costs = 0 → value = pure metered SUM = 4.00. assert value == pytest.approx(4.00, rel=1e-9), ( f"Expected metered SUM=4.00 (no contract → no standing), got {value!r}" ) def test_export_revenue_total_returns_metered_sum_when_no_active_contract(energy_db) -> None: """When active meter exists but no active contract, export_revenue_total returns metered sum. M7-T04 (D2): anchor = meter.started_at; summarize with no contract → credits = 0; return value = pure metered export sum within the meter window. """ from app.integrations.expose import build_catalog t0 = datetime(2026, 2, 1, 6, 0, tzinfo=timezone.utc) t1 = datetime(2026, 2, 1, 6, 15, tzinfo=timezone.utc) meter_start = datetime(2026, 2, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) # Inactive contract — active=False _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=t0, active=False, ) _make_period(session, period_start=t0, export_revenue=0.80, degraded=False) _make_period(session, period_start=t1, export_revenue=1.20, degraded=False) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) export_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_total" ) value = export_entry.entity.value_getter(session) # D2: anchor = meter.started_at; no contract → credits = 0 → value = metered sum = 2.00. assert value == pytest.approx(2.00, rel=1e-9), ( f"Expected metered SUM=2.00 (no contract → no credits), got {value!r}" ) # --------------------------------------------------------------------------- # 15. Tariff-aware pricing: buy_price_now / sell_price_now honour _current_tariff # --------------------------------------------------------------------------- _DUAL_TARIFF_PRICING = { "kind": "manual", "buy_dal": "0.2500", "buy_normal": "0.2700", "sell_dal": "0.0900", "sell_normal": "0.0950", "energy_tax": "0.1234", "ode": "0.0015", } @pytest.fixture() def reset_tariff(monkeypatch): """Reset dsmr_ingest._current_tariff to None before/after each tariff test.""" from app.services import dsmr_ingest as _di monkeypatch.setattr(_di, "_current_tariff", None) yield def _insert_manual_period(energy_db) -> None: """Insert an active electricity meter + non-degraded manual pricing period into energy_db. 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. """ 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") _make_period( session, period_start=t0, import_cost=0.10, export_revenue=0.05, currency="EUR", pricing=_DUAL_TARIFF_PRICING, degraded=False, ) session.commit() def test_buy_price_tariff_1_returns_dal(energy_db, reset_tariff) -> None: """buy_price_now must return buy_dal when tariff=1 (off-peak).""" 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) 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") value = buy_entry.entity.value_getter(session) assert value == pytest.approx(0.2500), f"Expected buy_dal=0.2500, got {value!r}" def test_buy_price_tariff_2_returns_normal(energy_db, reset_tariff) -> None: """buy_price_now must return buy_normal when tariff=2 (peak).""" 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) 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") value = buy_entry.entity.value_getter(session) assert value == pytest.approx(0.2700), f"Expected buy_normal=0.2700, got {value!r}" def test_buy_price_tariff_none_falls_back_to_normal(energy_db, reset_tariff) -> None: """buy_price_now must return buy_normal when tariff is None (no DSMR tariff received yet).""" 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) 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") value = buy_entry.entity.value_getter(session) assert value == pytest.approx(0.2700), f"Expected buy_normal=0.2700 fallback, got {value!r}" def test_sell_price_tariff_1_returns_dal(energy_db, reset_tariff) -> None: """sell_price_now must return sell_dal when tariff=1 (off-peak).""" 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) 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") value = sell_entry.entity.value_getter(session) assert value == pytest.approx(0.0900), f"Expected sell_dal=0.0900, got {value!r}" def test_sell_price_tariff_2_returns_normal(energy_db, reset_tariff) -> None: """sell_price_now must return sell_normal when tariff=2 (peak).""" 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) 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") value = sell_entry.entity.value_getter(session) assert value == pytest.approx(0.0950), f"Expected sell_normal=0.0950, got {value!r}" def test_sell_price_tariff_none_falls_back_to_normal(energy_db, reset_tariff) -> None: """sell_price_now must return sell_normal when tariff is 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(None) 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") value = sell_entry.entity.value_getter(session) assert value == pytest.approx(0.0950), f"Expected sell_normal=0.0950 fallback, got {value!r}" def test_tibber_buy_price_not_affected_by_tariff(energy_db, reset_tariff) -> None: """Tibber buy_price_now must NOT be affected by the DSMR tariff — always uses 'buy' key. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog from app.services.dsmr_ingest import set_current_tariff t0 = datetime(2026, 4, 1, 9, 0, tzinfo=timezone.utc) meter_start = datetime(2026, 4, 1, 0, 0, tzinfo=timezone.utc) tibber_pricing = { "kind": "tibber", "buy": "0.3100", "sell": "0.1500", "energy_tax": "0.1234", "sell_adjust": "0.0100", "total": "0.3100", } with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period( session, period_start=t0, pricing=tibber_pricing, degraded=False, ) session.commit() # Tibber pricing must return the same value regardless of tariff. for tariff_val in (1, 2, None): set_current_tariff(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") value = buy_entry.entity.value_getter(session) assert value == pytest.approx(0.3100), ( f"Tibber buy price must be 0.3100 regardless of tariff={tariff_val!r}, got {value!r}" ) def test_tibber_sell_price_not_affected_by_tariff(energy_db, reset_tariff) -> None: """Tibber sell_price_now must NOT be affected by the DSMR tariff — always uses 'sell' key. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog from app.services.dsmr_ingest import set_current_tariff t0 = datetime(2026, 4, 1, 9, 15, tzinfo=timezone.utc) meter_start = datetime(2026, 4, 1, 0, 0, tzinfo=timezone.utc) tibber_pricing = { "kind": "tibber", "buy": "0.3100", "sell": "0.1500", "energy_tax": "0.1234", "sell_adjust": "0.0100", "total": "0.3100", } with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start) _make_period( session, period_start=t0, pricing=tibber_pricing, degraded=False, ) session.commit() for tariff_val in (1, 2, None): set_current_tariff(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") value = sell_entry.entity.value_getter(session) assert value == pytest.approx(0.1500), ( f"Tibber sell price must be 0.1500 regardless of tariff={tariff_val!r}, got {value!r}" ) # --------------------------------------------------------------------------- # FU11 new tests: daily entities metadata + getter behaviour # --------------------------------------------------------------------------- def test_daily_entities_in_catalog(energy_db) -> None: """FU11: catalog must include import_cost_today and export_revenue_today. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) daily_keys = { "energy.import_cost_today", "energy.export_revenue_today", } found = {e.entity.key for e in catalog if e.entity.key in daily_keys} assert found == daily_keys, f"Expected daily keys {daily_keys!r} in catalog, got {found!r}" def test_daily_entities_have_total_increasing_state_class(energy_db) -> None: """FU11: daily entities must have state_class='total_increasing' (not 'total'). FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) daily_keys = {"energy.import_cost_today", "energy.export_revenue_today"} for entry in catalog: if entry.entity.key in daily_keys: assert entry.entity.state_class == "total_increasing", ( f"Daily entity {entry.entity.key!r} must have state_class='total_increasing', " f"got {entry.entity.state_class!r}" ) def test_daily_entities_have_monetary_device_class(energy_db) -> None: """FU11: daily entities must have device_class='monetary' and state_class='total_increasing'. HA developer docs prohibit MONETARY with state_class=measurement, but explicitly allow it with total and total_increasing. monetary+total_increasing is the correct combination for a non-negative, monotonically non-decreasing daily quantity that resets at local midnight. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) daily_keys = {"energy.import_cost_today", "energy.export_revenue_today"} for entry in catalog: if entry.entity.key in daily_keys: assert entry.entity.device_class == "monetary", ( f"Daily entity {entry.entity.key!r} must have device_class='monetary', " f"got {entry.entity.device_class!r}" ) def test_daily_entities_default_disabled(energy_db) -> None: """FU11: daily entities must default to enabled=False (no toggle row seeded). FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog now = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=now) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) daily_keys = {"energy.import_cost_today", "energy.export_revenue_today"} for entry in catalog: if entry.entity.key in daily_keys: assert entry.enabled is False, ( f"Daily entity {entry.entity.key!r} must default to enabled=False, " f"got enabled={entry.enabled!r}" ) def test_import_cost_today_getter_uses_local_day_window(energy_db) -> None: """FU11: import_cost_today getter returns value for today's local window. Setup: period at local 'today' (within today's UTC window), active contract. Timezone pinned to UTC. Verifies: value = metered_import + fixed_costs for 1 day (Principle C: today counts as elapsed if we're past local midnight — which UTC is). FUE-T05: provider requires an active electricity meter. """ from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) # Place a period in today's window (1 hour ago, well within today UTC midnight→tomorrow) # When TZ=UTC, local today = UTC today, so a period 1h ago is in today's window. t0 = now_utc.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1) # Ensure t0 is still today UTC (handle edge case where now is < 1h past UTC midnight) if t0.date() < now_utc.date(): t0 = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) # Contract starting well before today effective_from = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=30) meter_start = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=30) import_cost_today = 1.23 with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start, label="Today Window Meter") _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=effective_from, active=True, ) _make_period(session, period_start=t0, import_cost=import_cost_today, degraded=False) session.commit() with Session(energy_db) as session: # Pin timezone to UTC so today_local = UTC date (deterministic on any CI host). with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) today_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_today" ) value = today_entry.entity.value_getter(session) # Principle C: today counts as 1 day (we're within it, and it's "today"). # daily_standing = (30+60)/30 = 3.0 EUR/day × 1 day = 3.0 from decimal import Decimal daily_standing = Decimal("90") / Decimal("30") # = 3.0 expected = float(Decimal(str(import_cost_today)) + daily_standing * 1) assert value == pytest.approx(expected, rel=1e-6), ( f"import_cost_today expected {expected}, got {value!r}" ) def test_export_revenue_today_getter_uses_local_day_window(energy_db) -> None: """FU11: export_revenue_today getter returns value for today's local window. FUE-T05: provider requires an active electricity meter. """ from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) t0 = now_utc.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1) if t0.date() < now_utc.date(): t0 = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) effective_from = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=30) meter_start = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=30) export_today = 0.75 with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start, label="Today Export Meter") _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=effective_from, active=True, ) _make_period(session, period_start=t0, export_revenue=export_today, degraded=False) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) today_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_today" ) value = today_entry.entity.value_getter(session) from decimal import Decimal daily_credit = Decimal("365") / Decimal("365") # = 1.0 EUR/day × 1 day expected = float(Decimal(str(export_today)) + daily_credit * 1) assert value == pytest.approx(expected, rel=1e-6), ( f"export_revenue_today expected {expected}, got {value!r}" ) def test_daily_getter_returns_none_when_no_active_contract(energy_db) -> None: """FU11: daily getters return None when no active contract exists. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2026, 6, 25, 10, 0, tzinfo=timezone.utc) meter_start = datetime(2026, 6, 25, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start, label="No Contract Meter") # Inactive contract _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=t0, active=False, ) _make_period(session, period_start=t0, import_cost=1.0, export_revenue=0.5, degraded=False) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) import_today_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_today" ) export_today_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_today" ) import_val = import_today_entry.entity.value_getter(session) export_val = export_today_entry.entity.value_getter(session) assert import_val is None, ( f"import_cost_today must be None with no active contract, got {import_val!r}" ) assert export_val is None, ( f"export_revenue_today must be None with no active contract, got {export_val!r}" ) def test_daily_getter_returns_none_when_no_non_degraded_periods(energy_db) -> None: """FU11: daily getters return None when only degraded rows exist. FUE-T05: provider requires an active electricity meter. """ from app.integrations.expose import build_catalog t0 = datetime(2026, 6, 25, 10, 0, tzinfo=timezone.utc) meter_start = datetime(2026, 6, 25, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: _make_active_meter(session, started_at=meter_start, label="Degraded Only Meter") _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=t0 - datetime.resolution * 0, # same time is fine here active=True, ) _make_period(session, period_start=t0, import_cost=0.0, degraded=True) session.commit() with Session(energy_db) as session: catalog = build_catalog(session) import_today_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_today" ) value = import_today_entry.entity.value_getter(session) assert value is None, ( f"import_cost_today must be None with only degraded rows, got {value!r}" ) # --------------------------------------------------------------------------- # FU11 new test: anchor = max(contract_start, recording_start) # --------------------------------------------------------------------------- def test_cumulative_anchor_is_meter_started_at(energy_db) -> None: """M7-T04 (D2): anchor = active electricity meter's started_at, not recording_start. Scenario: contract effective_from = 180 days ago; meter.started_at = 7 days ago; first non-degraded period_start = 7 days ago. Expected: anchor = meter.started_at (7 days ago) → 8 days of standing charges. This replaces the old FU11 test (anchor=recording_start). Under D2 the anchor is always the active meter's started_at, irrespective of when data recording began. """ from decimal import Decimal from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) midnight_today = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) # Contract starts 180 days ago (far before the meter) contract_start = midnight_today - timedelta(days=180) # Active meter started 7 days ago — this is the D2 anchor meter_started_at = midnight_today - timedelta(days=7) t0 = meter_started_at # first period at meter start with Session(energy_db) as session: _make_active_meter(session, started_at=meter_started_at) _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=contract_start, active=True, ) _make_period(session, period_start=t0, import_cost=5.00, degraded=False) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # Anchor = meter.started_at (7 days ago). # Principle C (UTC pinned): count local days from meter_started_at.date() to today inclusive. # That's 7 + 1 = 8 days. expected_days = (now_utc.date() - meter_started_at.date()).days + 1 # = 8 daily_standing = Decimal("90") / Decimal("30") # = 3.0 EUR/day expected = float(Decimal("5.00") + daily_standing * expected_days) assert value == pytest.approx(expected, rel=1e-6), ( f"Expected anchor=meter.started_at anchor, " f"import_cost_total={expected} (metered=5.00 + standing={float(daily_standing*expected_days)} " f"over {expected_days} days), got {value!r}" ) # Verify that using the old FU11 anchor (contract start, 180 days ago) would give a # very different result (>€500 more standing), confirming we're NOT using it. old_expected_days = (now_utc.date() - contract_start.date()).days + 1 # ~181 days old_expected = float(Decimal("5.00") + daily_standing * old_expected_days) assert abs(value - old_expected) > 100, ( f"D2 anchor (meter.started_at) expected ~{expected:.2f}, " f"old FU11 anchor (contract_start) would yield ~{old_expected:.2f}; " f"difference must be >€100" ) # --------------------------------------------------------------------------- # M7-T04 new tests: per-meter cumulative reset (D2) # --------------------------------------------------------------------------- def test_cumulative_returns_none_when_no_active_meter(energy_db) -> None: """M7-T04 None protection: no active electricity meter → cumulative getters return None. FUE-T05 interaction: when no active meter exists, the provider returns [] and entities are not in the catalog. This test verifies the D2 None-protection of the value_getters themselves (the inner functions) by calling them directly via a catalog built with an active meter, then closing the meter (ended_at set) and calling the getter in that closed-meter session. This is the key D2 None protection: without an active meter the anchor cannot be resolved, so the getter returns None rather than a stale/wrong value. Periods exist (non-degraded), but no Meter row with ended_at IS NULL is present. """ from app.models.energy import Meter t0 = datetime(2026, 3, 1, 10, 0, tzinfo=timezone.utc) t1 = datetime(2026, 3, 1, 10, 15, tzinfo=timezone.utc) meter_start = datetime(2026, 3, 1, 0, 0, tzinfo=timezone.utc) closed_at = datetime(2026, 3, 1, 12, 0, tzinfo=timezone.utc) with Session(energy_db) as session: # Insert a meter that will later be closed. m = Meter( label="Will Be Closed", commodity="electricity", started_at=meter_start, ended_at=None, reason="initial", note=None, created_at=meter_start, ) session.add(m) _make_period(session, period_start=t0, import_cost=1.00, degraded=False) _make_period(session, period_start=t1, export_revenue=0.50, degraded=False) session.commit() meter_id = m.id # Build the catalog while the meter is active to capture the value_getter callables. from app.integrations.expose import build_catalog with Session(energy_db) as session: catalog = build_catalog(session) import_entry = next(e for e in catalog if e.entity.key == "energy.import_cost_total") export_entry = next(e for e in catalog if e.entity.key == "energy.export_revenue_total") import_getter = import_entry.entity.value_getter export_getter = export_entry.entity.value_getter # Now close the meter (no active meter remains). with Session(energy_db) as session: m = session.get(Meter, meter_id) m.ended_at = closed_at session.commit() # Call getters in a session where there is no active meter. with Session(energy_db) as session: import_val = import_getter(session) export_val = export_getter(session) assert import_val is None, ( f"import_cost_total must be None with no active meter, got {import_val!r}" ) assert export_val is None, ( f"export_revenue_total must be None with no active meter, got {export_val!r}" ) def test_cumulative_resets_after_meter_swap(energy_db) -> None: """M7-T04 (D2): after a meter swap the cumulative only counts the new meter's periods. Scenario: - Old meter: started_at = 30 days ago. Two non-degraded periods (€3.00 total). - Swap at: 7 days ago. Old meter closed, new active meter opened. - New meter: started_at = 7 days ago. One non-degraded period (€1.00). - Expected: import_cost_total = €1.00 + any standing charges from [7 days ago, now]. Old-meter periods (period_start < new_meter.started_at) fall outside the [anchor, now) window and are excluded, giving a clean reset to zero for the new meter epoch. """ from decimal import Decimal from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.models.energy import Meter as _Meter from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) midnight_today = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) old_meter_start = midnight_today - timedelta(days=30) swap_at = midnight_today - timedelta(days=7) # new meter anchor # Old-meter periods (before the swap) old_t0 = old_meter_start old_t1 = old_meter_start + timedelta(minutes=15) # New-meter period (after the swap) new_t0 = swap_at # first period of the new meter epoch new_meter_import = 1.00 # EUR — only this should be counted (old-meter 3.00 must be excluded) with Session(energy_db) as session: # Old meter: closed at swap_at old_m = _Meter( label="Old Meter", commodity="electricity", started_at=old_meter_start, ended_at=swap_at, # closed reason="initial", note=None, created_at=datetime.now(timezone.utc), ) session.add(old_m) session.flush() # New meter: active (ended_at IS NULL) — D2 anchor new_m = _Meter( label="New Meter", commodity="electricity", started_at=swap_at, ended_at=None, # active reason="meter_swap", note=None, created_at=datetime.now(timezone.utc), ) session.add(new_m) session.flush() # Old-meter periods (before swap): should NOT appear in post-swap cumulative _make_period(session, period_start=old_t0, import_cost=2.00, degraded=False) _make_period(session, period_start=old_t1, import_cost=1.00, degraded=False) # New-meter period (at swap point / after swap) _make_period(session, period_start=new_t0, import_cost=new_meter_import, degraded=False) # Active contract starting well before the old meter _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=old_meter_start, active=True, ) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) import_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_total" ) value = import_entry.entity.value_getter(session) # Anchor = new meter's started_at (7 days ago). # Old-meter periods (period_start < swap_at) fall outside [anchor, now) → excluded. # Only new_t0 (= swap_at = anchor) is within the window → metered = 1.00. # Standing charges: anchor = 7 days ago → 8 days (Principle C, UTC pinned). expected_days = (now_utc.date() - swap_at.date()).days + 1 # = 8 daily_standing = Decimal("90") / Decimal("30") # = 3.0 EUR/day expected = float(Decimal(str(new_meter_import)) + daily_standing * expected_days) assert value == pytest.approx(expected, rel=1e-6), ( f"Post-swap cumulative must only include new meter periods. " f"Expected {expected} (metered={new_meter_import} + " f"standing={float(daily_standing * expected_days)} over {expected_days} days), " f"got {value!r}" ) # Paranoia: if old-meter periods were accidentally included, value would be # much larger (old_meter_import = 3.00 would inflate it). assert value < new_meter_import + float(daily_standing * (expected_days + 1)) + 0.5, ( f"Value suspiciously large — old-meter periods may be included: {value!r}" ) def test_daily_getters_unaffected_by_d2_meter_anchor(energy_db) -> None: """M7-T04: daily import/export getters must still use the local-day window. D2 only changes the cumulative (*_total) anchor. The *_today getters use today's local-day window [local midnight, local tomorrow midnight) in UTC. They must work correctly regardless of which meter is active (meter started 30 days ago vs. 7 days ago — value must be the same: today's metered+fixed). FUE-T05: provider requires an active electricity meter to produce entities. The test verifies that the *_today value is NOT influenced by the meter's started_at (it uses today's window, not the meter anchor). """ from decimal import Decimal from datetime import timedelta from unittest.mock import patch from zoneinfo import ZoneInfo from app.integrations.expose import build_catalog from app.services import timezone as _tz_mod now_utc = datetime.now(timezone.utc) # Period 1h ago — in today's UTC window t0 = now_utc.replace(minute=0, second=0, microsecond=0) - timedelta(hours=1) if t0.date() < now_utc.date(): t0 = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) effective_from = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=30) # Meter started far in the past — daily value must equal today's window value, # NOT vary with the meter's started_at (that's the cumulative getter's job). meter_start = now_utc.replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=30) import_cost_today = 0.88 export_today = 0.44 with Session(energy_db) as session: # Active meter required by FUE-T05 provider; meter started_at must NOT # affect the today getter's window (today's local midnight → tomorrow). _make_active_meter(session, started_at=meter_start, label="Far-Past Meter") _make_contract_with_version( session, values=_STANDING_VALUES, effective_from=effective_from, active=True, ) _make_period( session, period_start=t0, import_cost=import_cost_today, export_revenue=export_today, degraded=False, ) session.commit() with Session(energy_db) as session: with patch.object(_tz_mod, "local_tz", return_value=ZoneInfo("UTC")): catalog = build_catalog(session) import_today_entry = next( e for e in catalog if e.entity.key == "energy.import_cost_today" ) export_today_entry = next( e for e in catalog if e.entity.key == "energy.export_revenue_today" ) import_val = import_today_entry.entity.value_getter(session) export_val = export_today_entry.entity.value_getter(session) # Principle C: today counts as 1 day. daily_standing = Decimal("90") / Decimal("30") # = 3.0 EUR/day daily_credit = Decimal("365") / Decimal("365") # = 1.0 EUR/day expected_import = float(Decimal(str(import_cost_today)) + daily_standing * 1) expected_export = float(Decimal(str(export_today)) + daily_credit * 1) assert import_val == pytest.approx(expected_import, rel=1e-6), ( f"import_cost_today must use local-day window regardless of meter started_at; " f"expected {expected_import}, got {import_val!r}" ) assert export_val == pytest.approx(expected_export, rel=1e-6), ( f"export_revenue_today must use local-day window regardless of meter started_at; " f"expected {expected_export}, got {export_val!r}" ) # --------------------------------------------------------------------------- # FUE-T05: energy-cost provider identity anchored to active electricity meter # --------------------------------------------------------------------------- def test_energy_cost_provider_returns_empty_when_no_active_meter(energy_db) -> None: """FUE-T05 ①: no active electricity meter → provider returns []. Without an active meter the energy-cost identifiers cannot be anchored, so the provider must return an empty list. HA will not see any energy-cost sensor until a meter is declared. """ from app.integrations.expose import _energy_cost_provider with Session(energy_db) as session: result = _energy_cost_provider(session) assert result == [], ( f"Expected [] when no active electricity meter, got {result!r}" ) def test_energy_cost_provider_identifiers_match_meter_uuid(energy_db) -> None: """FUE-T05 ②: with an active meter, identifiers[1] == meter.uuid. The HA device identity is anchored to the active meter's uuid. ha_discovery.py uses identifiers[1] as the MQTT node_id and unique_id seed; changing the active meter (meter swap) produces a new uuid → new node_id → new HA sensor. """ from app.integrations.expose import _energy_cost_provider meter_label = "Sunny Side Meter" now = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: m = _make_active_meter(session, started_at=now, label=meter_label) meter_uuid = m.uuid session.commit() with Session(energy_db) as session: entities = _energy_cost_provider(session) assert len(entities) == 6, f"Expected 6 entities, got {len(entities)}" for entity in entities: assert entity.device.identifiers == ("energy-cost", meter_uuid), ( f"identifiers must be ('energy-cost', meter.uuid); " f"expected ('energy-cost', {meter_uuid!r}), " f"got {entity.device.identifiers!r}" ) assert entity.device.name == meter_label, ( f"device.name must be exactly the meter label {meter_label!r}, " f"got {entity.device.name!r}" ) def test_energy_cost_entity_keys_do_not_contain_meter_uuid(energy_db) -> None: """FUE-T05 ③: entity keys remain stable 'energy.*' strings (no uuid injected). Keys are the anchor for the toggle table; they must NOT change when the meter changes. Only identifiers[1] (node_id / unique_id) changes on a meter swap. """ from app.integrations.expose import _energy_cost_provider now = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: m = _make_active_meter(session, started_at=now, label="Key Stability Meter") meter_uuid = m.uuid session.commit() with Session(energy_db) as session: entities = _energy_cost_provider(session) expected_keys = { "energy.buy_price_now", "energy.sell_price_now", "energy.import_cost_total", "energy.export_revenue_total", "energy.import_cost_today", "energy.export_revenue_today", } actual_keys = {e.key for e in entities} assert actual_keys == expected_keys, ( f"Expected stable keys {expected_keys!r}, got {actual_keys!r}" ) # None of the keys must embed the meter uuid. for key in actual_keys: assert meter_uuid not in key, ( f"Key {key!r} must NOT contain meter uuid {meter_uuid!r}" ) def test_energy_cost_toggle_survives_meter_swap(energy_db) -> None: """FUE-T05 ③ (toggle stability): enabled toggle on 'energy.buy_price_now' survives meter swap. After a meter swap the provider queries a new active meter → new identifiers[1] / unique_id / topic in HA. But the entity key stays 'energy.buy_price_now', so the existing toggle row (keyed by 'energy.buy_price_now') is still found → enabled=True. This is the critical property: the user does NOT need to re-tick toggles after a swap. """ from app.integrations.expose import build_catalog from app.models.energy import Meter from app.models.expose import ExposedEntityToggle t0 = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) swap_at = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: # Old meter (initial) old_meter = Meter( label="Old Meter", commodity="electricity", started_at=t0, ended_at=swap_at, reason="initial", note=None, created_at=t0, ) session.add(old_meter) # Enable buy_price_now toggle toggle = ExposedEntityToggle( key="energy.buy_price_now", enabled=True, updated_at=t0, ) session.add(toggle) session.commit() # Perform meter swap: new active meter with Session(energy_db) as session: new_meter = Meter( label="New Meter", commodity="electricity", started_at=swap_at, ended_at=None, reason="meter_swap", note=None, created_at=swap_at, ) session.add(new_meter) session.commit() new_meter_uuid = new_meter.uuid # After swap: catalog should use new meter uuid, but toggle state is preserved. 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"), None ) assert buy_entry is not None, "energy.buy_price_now must be present in catalog" # identifiers[1] must now be the NEW meter's uuid assert buy_entry.entity.device.identifiers[1] == new_meter_uuid, ( f"After swap, identifiers[1] must be new meter uuid {new_meter_uuid!r}, " f"got {buy_entry.entity.device.identifiers[1]!r}" ) # Toggle state must still be enabled (key unchanged → same toggle row found) assert buy_entry.enabled is True, ( "energy.buy_price_now toggle must remain enabled after meter swap " "(key is stable, toggle row survives)" ) def test_energy_cost_identifiers_change_after_meter_swap(energy_db) -> None: """FUE-T05 ④: after meter swap, provider produces new identifiers[1] (new meter uuid). Old uuid's entities are no longer produced → HA sensor for old uuid is frozen. New uuid's entities appear → HA creates fresh sensors for the new meter. """ from app.integrations.expose import _energy_cost_provider from app.models.energy import Meter t0 = datetime(2026, 1, 1, 0, 0, tzinfo=timezone.utc) swap_at = datetime(2026, 6, 1, 0, 0, tzinfo=timezone.utc) with Session(energy_db) as session: old_meter = Meter( label="Old Meter", commodity="electricity", started_at=t0, ended_at=swap_at, reason="initial", note=None, created_at=t0, ) session.add(old_meter) session.commit() old_uuid = old_meter.uuid # Before swap: old meter closed, new meter not yet created → no active meter → [] with Session(energy_db) as session: result_no_active = _energy_cost_provider(session) assert result_no_active == [], ( "Expected [] after old meter is closed but before new meter is created" ) # Create new active meter (simulates the swap completing) with Session(energy_db) as session: new_meter = Meter( label="New Meter", commodity="electricity", started_at=swap_at, ended_at=None, reason="meter_swap", note=None, created_at=swap_at, ) session.add(new_meter) session.commit() new_uuid = new_meter.uuid assert old_uuid != new_uuid, "Old and new meter uuids must differ" with Session(energy_db) as session: entities_after_swap = _energy_cost_provider(session) assert len(entities_after_swap) == 6, ( f"Expected 6 entities after swap, got {len(entities_after_swap)}" ) for entity in entities_after_swap: assert entity.device.identifiers[1] == new_uuid, ( f"After swap, identifiers[1] must be new uuid {new_uuid!r}, " f"got {entity.device.identifiers[1]!r}" ) # Old uuid must not appear in identifiers assert entity.device.identifiers[1] != old_uuid, ( f"After swap, old uuid {old_uuid!r} must not appear in identifiers" )