"""Tests for energy pricing, DSMR metering, and meter epoch tables and ORM models. Covers: 1. Migration shape: upgrade to head creates all six tables with correct columns, constraints, and indexes; downgrade -1 cleanly removes them. 2. ORM metadata: Base.metadata.tables contains all six tables; FKs are RESTRICT; JSON columns are present; unique and index constraints are correct. 3. Baseline constant: APP_BASELINE_REVISION matches the actual Alembic head. 4. Basic ORM round-trip: insert + retrieve for each table, JSON payload round-trip. 5. FK RESTRICT: deleting a referenced parent row must fail when children exist. 6. Meter model: fields, nullability, FK from energy_cost_period. 7. Migration backfill: initial meter creation, idempotency, audit. """ from __future__ import annotations from datetime import datetime, timezone from pathlib import Path import pytest import sqlalchemy.exc from alembic import command from alembic.config import Config from sqlalchemy import create_engine, event as sa_event, inspect, text from sqlalchemy.orm import Session from app.db import Base from app.models.energy import ( DsmrReading, EnergyContract, EnergyContractVersion, Meter, TibberPrice, EnergyCostPeriod, ) from scripts.app_db_adopt import APP_BASELINE_REVISION # --------------------------------------------------------------------------- # 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 _engine_with_fk(db_url: str): """Create a SQLAlchemy engine with SQLite FK enforcement enabled.""" engine = create_engine(db_url, connect_args={"check_same_thread": False}) @sa_event.listens_for(engine, "connect") def _enable_fk(dbapi_conn, _rec): cursor = dbapi_conn.cursor() cursor.execute("PRAGMA foreign_keys = ON") cursor.close() return engine # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def energy_db(tmp_path: Path): """Temporary SQLite DB upgraded to the current Alembic head.""" db_path = tmp_path / "energy_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) yield engine engine.dispose() @pytest.fixture() def energy_db_with_fk(tmp_path: Path): """Temporary SQLite DB upgraded to head with FK enforcement enabled.""" db_path = tmp_path / "energy_fk_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = _engine_with_fk(db_url) yield engine engine.dispose() # --------------------------------------------------------------------------- # 1. Migration shape tests # --------------------------------------------------------------------------- def test_energy_tables_exist_after_upgrade(energy_db): """All six energy tables must exist after upgrade to head.""" inspector = inspect(energy_db) table_names = inspector.get_table_names() expected_tables = { "meter", "dsmr_reading", "energy_contract", "energy_contract_version", "tibber_price", "energy_cost_period", } for table in expected_tables: assert table in table_names, f"{table!r} missing after upgrade to head" def test_dsmr_reading_columns(energy_db): """dsmr_reading must have id, recorded_at (NOT NULL), source_id (nullable), payload (NOT NULL).""" inspector = inspect(energy_db) columns = {col["name"]: col for col in inspector.get_columns("dsmr_reading")} assert "id" in columns and not columns["id"]["nullable"] assert "recorded_at" in columns and not columns["recorded_at"]["nullable"] assert "source_id" in columns and columns["source_id"]["nullable"] assert "payload" in columns and not columns["payload"]["nullable"] def test_dsmr_reading_recorded_at_unique(energy_db): """dsmr_reading.recorded_at is the UNIQUE de-dup key (telegram-id-independent).""" inspector = inspect(energy_db) unique_constraints = inspector.get_unique_constraints("dsmr_reading") unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] assert "recorded_at" in unique_cols, "recorded_at must have a unique constraint" def test_dsmr_reading_source_id_not_unique(energy_db): """dsmr_reading.source_id (telegram id) must NOT be unique — it overflows/resets, so it is kept only as a reference value and never relied on for dedup.""" inspector = inspect(energy_db) unique_constraints = inspector.get_unique_constraints("dsmr_reading") unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] assert "source_id" not in unique_cols, "source_id must NOT have a unique constraint" def test_energy_contract_columns(energy_db): """energy_contract must have all required columns.""" inspector = inspect(energy_db) columns = {col["name"]: col for col in inspector.get_columns("energy_contract")} required_non_nullable = {"id", "name", "kind", "active", "currency", "created_at", "updated_at"} for col_name in required_non_nullable: assert col_name in columns, f"Missing column: {col_name}" assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" def test_energy_contract_version_columns(energy_db): """energy_contract_version must have all required columns with correct nullability.""" inspector = inspect(energy_db) columns = {col["name"]: col for col in inspector.get_columns("energy_contract_version")} non_nullable = {"id", "contract_id", "effective_from", "values", "created_at"} nullable = {"effective_to"} for col_name in non_nullable: assert col_name in columns, f"Missing column: {col_name}" assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" for col_name in nullable: assert col_name in columns, f"Missing column: {col_name}" assert columns[col_name]["nullable"], f"{col_name} should be nullable" def test_tibber_price_columns(energy_db): """tibber_price must have all required columns.""" inspector = inspect(energy_db) columns = {col["name"]: col for col in inspector.get_columns("tibber_price")} non_nullable = { "id", "starts_at", "resolution", "energy", "tax", "total", "currency", "fetched_at", } nullable = {"level"} for col_name in non_nullable: assert col_name in columns, f"Missing column: {col_name}" assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" for col_name in nullable: assert col_name in columns, f"Missing column: {col_name}" assert columns[col_name]["nullable"], f"{col_name} should be nullable" def test_tibber_price_starts_at_unique(energy_db): """tibber_price.starts_at must have a unique constraint.""" inspector = inspect(energy_db) unique_constraints = inspector.get_unique_constraints("tibber_price") unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] assert "starts_at" in unique_cols, "starts_at must have a unique constraint" def test_energy_cost_period_columns(energy_db): """energy_cost_period must have all required columns with correct nullability.""" inspector = inspect(energy_db) columns = {col["name"]: col for col in inspector.get_columns("energy_cost_period")} non_nullable = { "id", "period_start", "d1_kwh", "d2_kwh", "r1_kwh", "r2_kwh", "import_cost", "export_revenue", "net_cost", "currency", "pricing", "degraded", "computed_at", } nullable = {"contract_version_id", "meter_id"} for col_name in non_nullable: assert col_name in columns, f"Missing column: {col_name}" assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" for col_name in nullable: assert col_name in columns, f"Missing column: {col_name}" assert columns[col_name]["nullable"], f"{col_name} should be nullable" def test_energy_cost_period_period_start_unique(energy_db): """energy_cost_period.period_start must have a unique constraint.""" inspector = inspect(energy_db) unique_constraints = inspector.get_unique_constraints("energy_cost_period") unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] assert "period_start" in unique_cols, "period_start must have a unique constraint" def test_energy_contract_version_fk_to_contract(energy_db): """energy_contract_version.contract_id must have a FK referencing energy_contract.id.""" inspector = inspect(energy_db) fks = inspector.get_foreign_keys("energy_contract_version") assert len(fks) == 1, f"Expected 1 FK on energy_contract_version, got {len(fks)}" fk = fks[0] assert fk["referred_table"] == "energy_contract" assert "contract_id" in fk["constrained_columns"] assert "id" in fk["referred_columns"] def test_energy_cost_period_fk_to_contract_version(energy_db): """energy_cost_period.contract_version_id must have a FK referencing energy_contract_version.id.""" inspector = inspect(energy_db) fks = inspector.get_foreign_keys("energy_cost_period") # Two FKs: contract_version_id → energy_contract_version.id # meter_id → meter.id fk_by_col = { col: fk for fk in fks for col in fk["constrained_columns"] } assert "contract_version_id" in fk_by_col, ( "energy_cost_period must have a FK on contract_version_id" ) fk = fk_by_col["contract_version_id"] assert fk["referred_table"] == "energy_contract_version" assert "id" in fk["referred_columns"] def test_energy_cost_period_fk_to_meter(energy_db): """energy_cost_period.meter_id must have a FK referencing meter.id.""" inspector = inspect(energy_db) fks = inspector.get_foreign_keys("energy_cost_period") fk_by_col = { col: fk for fk in fks for col in fk["constrained_columns"] } assert "meter_id" in fk_by_col, ( "energy_cost_period must have a FK on meter_id" ) fk = fk_by_col["meter_id"] assert fk["referred_table"] == "meter" assert "id" in fk["referred_columns"] def test_downgrade_removes_energy_tables(tmp_path: Path): """Downgrading to 20260622_10_exposed_entities must cleanly remove all five energy tables.""" db_path = tmp_path / "downgrade_energy_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) for table in ("dsmr_reading", "energy_contract", "energy_contract_version", "tibber_price", "energy_cost_period"): assert table in inspector.get_table_names(), f"{table} should exist after upgrade" engine.dispose() # Downgrade to the revision just before ours — the explicit down_revision. command.downgrade(alembic_cfg, "20260622_10_exposed_entities") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) table_names = inspector.get_table_names() for table in ("dsmr_reading", "energy_contract", "energy_contract_version", "tibber_price", "energy_cost_period"): assert table not in table_names, f"{table} should be gone after downgrade" engine.dispose() def test_dsmr_decouple_migration_downgrade_restores_source_id_unique(tmp_path: Path): """Downgrading the telegram-id-decouple migration to 20260623_11 must restore the original constraints: source_id UNIQUE, recorded_at non-unique index.""" db_path = tmp_path / "dsmr_decouple_downgrade.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") # At head: recorded_at is unique, source_id is not. engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) head_unique = [ c for uc in inspector.get_unique_constraints("dsmr_reading") for c in uc["column_names"] ] assert "recorded_at" in head_unique assert "source_id" not in head_unique engine.dispose() # Downgrade just this migration. command.downgrade(alembic_cfg, "20260623_11_energy_tables") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) pre_unique = [ c for uc in inspector.get_unique_constraints("dsmr_reading") for c in uc["column_names"] ] assert "source_id" in pre_unique, "source_id unique must be restored on downgrade" assert "recorded_at" not in pre_unique index_names = {idx["name"] for idx in inspector.get_indexes("dsmr_reading")} assert "ix_dsmr_reading_recorded_at" in index_names engine.dispose() # --------------------------------------------------------------------------- # 2. ORM metadata checks (Base.metadata) # --------------------------------------------------------------------------- def test_base_metadata_contains_energy_tables(): """Base.metadata.tables must include all six energy tables.""" expected = { "meter", "dsmr_reading", "energy_contract", "energy_contract_version", "tibber_price", "energy_cost_period", } for table in expected: assert table in Base.metadata.tables, f"{table!r} not in Base.metadata" def test_dsmr_reading_payload_is_json(): """DsmrReading.payload must be mapped as a JSON column in ORM metadata.""" from sqlalchemy.types import JSON as JSON_TYPE table = Base.metadata.tables["dsmr_reading"] col = table.columns["payload"] assert isinstance(col.type, JSON_TYPE), "payload column must be JSON type" def test_energy_contract_version_values_is_json(): """EnergyContractVersion.values must be mapped as a JSON column in ORM metadata.""" from sqlalchemy.types import JSON as JSON_TYPE table = Base.metadata.tables["energy_contract_version"] col = table.columns["values"] assert isinstance(col.type, JSON_TYPE), "values column must be JSON type" def test_energy_cost_period_pricing_is_json(): """EnergyCostPeriod.pricing must be mapped as a JSON column in ORM metadata.""" from sqlalchemy.types import JSON as JSON_TYPE table = Base.metadata.tables["energy_cost_period"] col = table.columns["pricing"] assert isinstance(col.type, JSON_TYPE), "pricing column must be JSON type" def test_energy_contract_version_fk_ondelete_restrict(): """The FK from energy_contract_version.contract_id must be ON DELETE RESTRICT.""" table = Base.metadata.tables["energy_contract_version"] contract_id_col = table.columns["contract_id"] assert contract_id_col.foreign_keys, "contract_id must have a foreign key" fk = next(iter(contract_id_col.foreign_keys)) assert fk.ondelete == "RESTRICT", ( f"FK ondelete must be RESTRICT, got: {fk.ondelete!r}" ) def test_energy_cost_period_fk_ondelete_restrict(): """The FK from energy_cost_period.contract_version_id must be ON DELETE RESTRICT.""" table = Base.metadata.tables["energy_cost_period"] col = table.columns["contract_version_id"] assert col.foreign_keys, "contract_version_id must have a foreign key" fk = next(iter(col.foreign_keys)) assert fk.ondelete == "RESTRICT", ( f"FK ondelete must be RESTRICT, got: {fk.ondelete!r}" ) def test_energy_cost_period_meter_id_fk_ondelete_restrict(): """The FK from energy_cost_period.meter_id must be ON DELETE RESTRICT.""" table = Base.metadata.tables["energy_cost_period"] col = table.columns["meter_id"] assert col.foreign_keys, "meter_id must have a foreign key" fk = next(iter(col.foreign_keys)) assert fk.ondelete == "RESTRICT", ( f"meter_id FK ondelete must be RESTRICT, got: {fk.ondelete!r}" ) def test_dsmr_reading_recorded_at_unique_in_metadata(): """DsmrReading.recorded_at must be the unique de-dup key in ORM metadata, and source_id must NOT be unique (decoupled from the telegram id).""" table = Base.metadata.tables["dsmr_reading"] assert table.columns["recorded_at"].unique, "recorded_at must be declared unique" assert not table.columns["source_id"].unique, "source_id must NOT be unique" def test_tibber_price_starts_at_unique_in_metadata(): """TibberPrice.starts_at must be declared unique in ORM metadata.""" table = Base.metadata.tables["tibber_price"] col = table.columns["starts_at"] assert col.unique, "starts_at must be declared unique" # --------------------------------------------------------------------------- # 3. Baseline constant # --------------------------------------------------------------------------- def test_app_baseline_revision_matches_head(tmp_path: Path): """APP_BASELINE_REVISION must equal the Alembic head revision.""" from alembic.script import ScriptDirectory db_url = f"sqlite:///{tmp_path / 'rev_check.db'}" alembic_cfg = _make_app_alembic_config(db_url) script = ScriptDirectory.from_config(alembic_cfg) heads = script.get_heads() assert len(heads) == 1, f"Expected exactly 1 Alembic head, got {heads}" head = heads[0] assert APP_BASELINE_REVISION == head, ( f"APP_BASELINE_REVISION={APP_BASELINE_REVISION!r} does not match " f"Alembic head={head!r}" ) # --------------------------------------------------------------------------- # 4. ORM round-trip tests # --------------------------------------------------------------------------- def test_dsmr_reading_insert_and_retrieve(energy_db): """A DsmrReading can be inserted with a JSON payload and retrieved correctly.""" now = datetime.now(tz=timezone.utc) sample_payload = { "timestamp": "2026-06-23T10:00:00+00:00", "electricity_delivered_1": "20915.154", "electricity_delivered_2": "18372.099", "electricity_returned_1": "1234.567", "electricity_returned_2": "890.123", "current_electricity_usage": "1.234", "extra_device_timestamp": None, } with Session(energy_db) as session: reading = DsmrReading( recorded_at=now, source_id=42, payload=sample_payload, ) session.add(reading) session.commit() reading_id = reading.id with Session(energy_db) as session: fetched = session.get(DsmrReading, reading_id) assert fetched is not None assert fetched.source_id == 42 assert fetched.payload == sample_payload assert fetched.payload["electricity_delivered_1"] == "20915.154" assert fetched.payload["extra_device_timestamp"] is None def test_dsmr_reading_source_id_nullable(energy_db): """DsmrReading can be inserted without a source_id (nullable).""" now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: reading = DsmrReading( recorded_at=now, source_id=None, payload={"test": True}, ) session.add(reading) session.commit() reading_id = reading.id with Session(energy_db) as session: fetched = session.get(DsmrReading, reading_id) assert fetched is not None assert fetched.source_id is None def test_energy_contract_insert_and_retrieve(energy_db): """An EnergyContract can be inserted and retrieved with correct defaults.""" now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: contract = EnergyContract( name="My Manual Contract", kind="manual", active=True, currency="EUR", created_at=now, updated_at=now, ) session.add(contract) session.commit() contract_id = contract.id with Session(energy_db) as session: fetched = session.get(EnergyContract, contract_id) assert fetched is not None assert fetched.name == "My Manual Contract" assert fetched.kind == "manual" assert fetched.active is True assert fetched.currency == "EUR" def test_energy_contract_version_insert_and_retrieve(energy_db): """An EnergyContractVersion can be inserted with a JSON values blob.""" now = datetime.now(tz=timezone.utc) pricing_values = { "energy": { "buy": {"normal": 0.133, "dal": 0.127}, "sell": {"normal": 0.05, "dal": 0.05}, "energy_tax": 0.1108, "ode": 0.0, }, "standing": { "network_fee": 9.87, "management_fee": 9.87, }, "credits": { "heffingskorting": 600.0, }, } with Session(energy_db) as session: contract = EnergyContract( name="Fixed Rate 2026", kind="manual", active=False, currency="EUR", created_at=now, updated_at=now, ) session.add(contract) session.flush() version = EnergyContractVersion( contract_id=contract.id, effective_from=now, effective_to=None, values=pricing_values, created_at=now, ) session.add(version) session.commit() version_id = version.id with Session(energy_db) as session: fetched = session.get(EnergyContractVersion, version_id) assert fetched is not None assert fetched.effective_to is None assert fetched.values == pricing_values assert fetched.values["energy"]["buy"]["normal"] == 0.133 def test_tibber_price_insert_and_retrieve(energy_db): """A TibberPrice can be inserted and retrieved correctly.""" now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: price = TibberPrice( starts_at=now, resolution="QUARTER_HOURLY", energy=0.08, tax=0.05, total=0.13, level="NORMAL", currency="EUR", fetched_at=now, ) session.add(price) session.commit() price_id = price.id with Session(energy_db) as session: fetched = session.get(TibberPrice, price_id) assert fetched is not None assert fetched.total == 0.13 assert fetched.level == "NORMAL" assert fetched.currency == "EUR" def test_tibber_price_level_nullable(energy_db): """TibberPrice.level can be None.""" now = datetime.now(tz=timezone.utc) from datetime import timedelta other_time = now + timedelta(minutes=15) with Session(energy_db) as session: price = TibberPrice( starts_at=other_time, resolution="QUARTER_HOURLY", energy=0.07, tax=0.04, total=0.11, level=None, currency="EUR", fetched_at=now, ) session.add(price) session.commit() price_id = price.id with Session(energy_db) as session: fetched = session.get(TibberPrice, price_id) assert fetched is not None assert fetched.level is None def test_energy_cost_period_insert_and_retrieve(energy_db): """An EnergyCostPeriod can be inserted with a JSON pricing snapshot.""" now = datetime.now(tz=timezone.utc) pricing_snapshot = { "kind": "manual", "buy_normal": 0.133, "buy_dal": 0.127, "sell_normal": 0.05, "sell_dal": 0.05, } with Session(energy_db) as session: period = EnergyCostPeriod( period_start=now, d1_kwh=0.5, d2_kwh=1.2, r1_kwh=0.0, r2_kwh=0.0, import_cost=0.225, export_revenue=0.0, net_cost=0.225, currency="EUR", pricing=pricing_snapshot, contract_version_id=None, degraded=False, computed_at=now, ) session.add(period) session.commit() period_id = period.id with Session(energy_db) as session: fetched = session.get(EnergyCostPeriod, period_id) assert fetched is not None assert fetched.d1_kwh == 0.5 assert fetched.d2_kwh == 1.2 assert fetched.net_cost == 0.225 assert fetched.degraded is False assert fetched.contract_version_id is None assert fetched.pricing == pricing_snapshot # --------------------------------------------------------------------------- # 5. FK RESTRICT enforcement tests # --------------------------------------------------------------------------- def test_contract_version_restrict_prevents_contract_deletion( tmp_path: Path, ): """Deleting an EnergyContract with versions must fail due to ON DELETE RESTRICT.""" db_path = tmp_path / "restrict_contract_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = _engine_with_fk(db_url) try: now = datetime.now(tz=timezone.utc) with Session(engine) as session: contract = EnergyContract( name="RESTRICT Test Contract", kind="manual", active=False, currency="EUR", created_at=now, updated_at=now, ) session.add(contract) session.flush() version = EnergyContractVersion( contract_id=contract.id, effective_from=now, effective_to=None, values={"test": True}, created_at=now, ) session.add(version) session.commit() contract_id = contract.id with pytest.raises(sqlalchemy.exc.IntegrityError): with Session(engine) as session: session.execute( text("DELETE FROM energy_contract WHERE id = :cid"), {"cid": contract_id}, ) session.commit() finally: engine.dispose() def test_cost_period_restrict_prevents_version_deletion(tmp_path: Path): """Deleting an EnergyContractVersion with cost periods must fail due to ON DELETE RESTRICT.""" db_path = tmp_path / "restrict_version_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = _engine_with_fk(db_url) try: now = datetime.now(tz=timezone.utc) with Session(engine) as session: contract = EnergyContract( name="RESTRICT Version Test", kind="manual", active=False, currency="EUR", created_at=now, updated_at=now, ) session.add(contract) session.flush() version = EnergyContractVersion( contract_id=contract.id, effective_from=now, effective_to=None, values={"test": True}, created_at=now, ) session.add(version) session.flush() period = EnergyCostPeriod( period_start=now, d1_kwh=0.1, d2_kwh=0.2, r1_kwh=0.0, r2_kwh=0.0, import_cost=0.05, export_revenue=0.0, net_cost=0.05, currency="EUR", pricing={"kind": "manual"}, contract_version_id=version.id, degraded=False, computed_at=now, ) session.add(period) session.commit() version_id = version.id with pytest.raises(sqlalchemy.exc.IntegrityError): with Session(engine) as session: session.execute( text("DELETE FROM energy_contract_version WHERE id = :vid"), {"vid": version_id}, ) session.commit() finally: engine.dispose() # --------------------------------------------------------------------------- # 6. Meter model tests (M7-T01) # --------------------------------------------------------------------------- def test_meter_table_exists_after_upgrade(energy_db): """meter table must exist after upgrade to head.""" inspector = inspect(energy_db) assert "meter" in inspector.get_table_names(), "meter table missing after upgrade to head" def test_meter_columns(energy_db): """meter must have all required columns with correct nullability.""" inspector = inspect(energy_db) columns = {col["name"]: col for col in inspector.get_columns("meter")} non_nullable = {"id", "label", "commodity", "started_at", "reason", "created_at"} nullable = {"ended_at", "note"} for col_name in non_nullable: assert col_name in columns, f"Missing column: {col_name}" assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" for col_name in nullable: assert col_name in columns, f"Missing column: {col_name}" assert columns[col_name]["nullable"], f"{col_name} should be nullable" def test_meter_orm_metadata(): """Meter must be registered in Base.metadata with correct field types.""" assert "meter" in Base.metadata.tables, "meter not in Base.metadata" table = Base.metadata.tables["meter"] assert "id" in table.columns assert "label" in table.columns assert "commodity" in table.columns assert "started_at" in table.columns assert "ended_at" in table.columns assert "reason" in table.columns assert "note" in table.columns assert "created_at" in table.columns def test_meter_insert_and_retrieve(energy_db): """A Meter row can be inserted and retrieved with all fields intact.""" now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: m = Meter( label="Test meter @ Dorpsstraat 1", commodity="electricity", started_at=now, ended_at=None, reason="initial", note="2G smart meter", created_at=now, ) session.add(m) session.commit() meter_id = m.id with Session(energy_db) as session: fetched = session.get(Meter, meter_id) assert fetched is not None assert fetched.label == "Test meter @ Dorpsstraat 1" assert fetched.commodity == "electricity" assert fetched.reason == "initial" assert fetched.note == "2G smart meter" assert fetched.ended_at is None def test_meter_ended_at_nullable(energy_db): """Meter.ended_at and Meter.note can both be None (active/ongoing meter).""" now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: m = Meter( label="Active meter", commodity="electricity", started_at=now, ended_at=None, reason="home_move", note=None, created_at=now, ) session.add(m) session.commit() meter_id = m.id with Session(energy_db) as session: fetched = session.get(Meter, meter_id) assert fetched is not None assert fetched.ended_at is None assert fetched.note is None def test_energy_cost_period_meter_id_nullable(energy_db): """EnergyCostPeriod.meter_id must be nullable (no meter FK required).""" now = datetime.now(tz=timezone.utc) with Session(energy_db) as session: period = EnergyCostPeriod( period_start=now, d1_kwh=0.5, d2_kwh=1.2, r1_kwh=0.0, r2_kwh=0.0, import_cost=0.225, export_revenue=0.0, net_cost=0.225, currency="EUR", pricing={"kind": "manual"}, contract_version_id=None, meter_id=None, degraded=False, computed_at=now, ) session.add(period) session.commit() period_id = period.id with Session(energy_db) as session: fetched = session.get(EnergyCostPeriod, period_id) assert fetched is not None assert fetched.meter_id is None def test_energy_cost_period_meter_id_fk_enforced(tmp_path: Path): """Inserting an EnergyCostPeriod with a non-existent meter_id must fail (RESTRICT FK).""" db_path = tmp_path / "meter_fk_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = _engine_with_fk(db_url) try: now = datetime.now(tz=timezone.utc) with pytest.raises(sqlalchemy.exc.IntegrityError): with Session(engine) as session: period = EnergyCostPeriod( period_start=now, d1_kwh=0.0, d2_kwh=0.0, r1_kwh=0.0, r2_kwh=0.0, import_cost=0.0, export_revenue=0.0, net_cost=0.0, currency="EUR", pricing={}, contract_version_id=None, meter_id=9999, # non-existent meter degraded=False, computed_at=now, ) session.add(period) session.commit() finally: engine.dispose() def test_meter_restrict_prevents_deletion_with_cost_periods(tmp_path: Path): """Deleting a Meter that has cost periods attributed to it must fail (ON DELETE RESTRICT).""" db_path = tmp_path / "meter_restrict_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = _engine_with_fk(db_url) try: now = datetime.now(tz=timezone.utc) with Session(engine) as session: m = Meter( label="RESTRICT Test Meter", commodity="electricity", started_at=now, ended_at=None, reason="initial", note=None, created_at=now, ) session.add(m) session.flush() period = EnergyCostPeriod( period_start=now, d1_kwh=0.1, d2_kwh=0.2, r1_kwh=0.0, r2_kwh=0.0, import_cost=0.05, export_revenue=0.0, net_cost=0.05, currency="EUR", pricing={"kind": "manual"}, contract_version_id=None, meter_id=m.id, degraded=False, computed_at=now, ) session.add(period) session.commit() meter_id = m.id with pytest.raises(sqlalchemy.exc.IntegrityError): with Session(engine) as session: session.execute( text("DELETE FROM meter WHERE id = :mid"), {"mid": meter_id}, ) session.commit() finally: engine.dispose() # --------------------------------------------------------------------------- # 7. Migration backfill tests (M7-T01) # --------------------------------------------------------------------------- def test_migration_empty_db_no_initial_meter(tmp_path: Path): """Upgrading an empty database must not create any meter rows.""" db_path = tmp_path / "empty_backfill_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) with Session(engine) as session: count = session.execute(text("SELECT COUNT(*) FROM meter")).scalar() assert count == 0, f"Expected 0 meter rows in empty DB after upgrade, got {count}" engine.dispose() def test_migration_backfill_creates_initial_meter_from_readings(tmp_path: Path): """Upgrading with existing dsmr_reading rows must create exactly one initial meter and back-fill all energy_cost_period rows with its id.""" from datetime import timedelta db_path = tmp_path / "backfill_readings_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) # Upgrade only to the revision just before M7-T01 (i.e. pre-meter). command.upgrade(alembic_cfg, "20260624_12_dsmr_decouple_telegram_id") engine = create_engine(db_url, connect_args={"check_same_thread": False}) now = datetime.now(tz=timezone.utc) t0 = now - timedelta(hours=2) t1 = now - timedelta(hours=1) with Session(engine) as session: # Insert two dsmr_reading rows. session.execute( text( "INSERT INTO dsmr_reading (recorded_at, source_id, payload) " "VALUES (:ts, NULL, '{}')" ), {"ts": t0.strftime("%Y-%m-%dT%H:%M:%S")}, ) session.execute( text( "INSERT INTO dsmr_reading (recorded_at, source_id, payload) " "VALUES (:ts, NULL, '{}')" ), {"ts": t1.strftime("%Y-%m-%dT%H:%M:%S")}, ) # Insert two energy_cost_period rows (no meter_id column yet at this revision). session.execute( text( "INSERT INTO energy_cost_period " "(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, " "import_cost, export_revenue, net_cost, currency, pricing, " "contract_version_id, degraded, computed_at) " "VALUES (:ps, 0.1, 0.2, 0.0, 0.0, 0.05, 0.0, 0.05, 'EUR', '{}', " "NULL, 0, :now)" ), {"ps": t0.strftime("%Y-%m-%dT%H:%M:%S"), "now": now.strftime("%Y-%m-%dT%H:%M:%S")}, ) session.execute( text( "INSERT INTO energy_cost_period " "(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, " "import_cost, export_revenue, net_cost, currency, pricing, " "contract_version_id, degraded, computed_at) " "VALUES (:ps, 0.1, 0.2, 0.0, 0.0, 0.05, 0.0, 0.05, 'EUR', '{}', " "NULL, 0, :now)" ), {"ps": t1.strftime("%Y-%m-%dT%H:%M:%S"), "now": now.strftime("%Y-%m-%dT%H:%M:%S")}, ) session.commit() engine.dispose() # Now upgrade to head (applies M7-T01 migration with backfill). command.upgrade(alembic_cfg, "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) with Session(engine) as session: # Exactly one initial meter must exist. meter_count = session.execute( text("SELECT COUNT(*) FROM meter WHERE reason = 'initial' AND ended_at IS NULL") ).scalar() assert meter_count == 1, f"Expected 1 initial meter, got {meter_count}" # All non-degraded energy_cost_period rows must have meter_id set. null_count = session.execute( text( "SELECT COUNT(*) FROM energy_cost_period " "WHERE meter_id IS NULL AND degraded = 0" ) ).scalar() assert null_count == 0, ( f"Expected 0 non-degraded periods with NULL meter_id, got {null_count}" ) # The initial meter started_at must match the earliest dsmr_reading.recorded_at. meter_row = session.execute( text("SELECT started_at FROM meter WHERE reason = 'initial' LIMIT 1") ).fetchone() assert meter_row is not None meter_started = meter_row[0] # Both meter_started and t0 are now UTC strings / datetimes — just verify # the date portion matches (second-level precision is sufficient). assert t0.strftime("%Y-%m-%dT%H:%M:%S") in str(meter_started), ( f"initial meter started_at {meter_started!r} should match earliest reading {t0}" ) engine.dispose() def test_migration_backfill_idempotent(tmp_path: Path): """Running upgrade head twice must not create duplicate initial meters or corrupt data.""" db_path = tmp_path / "idempotent_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) # Upgrade to pre-meter revision and insert a reading. command.upgrade(alembic_cfg, "20260624_12_dsmr_decouple_telegram_id") now = datetime.now(tz=timezone.utc) engine = create_engine(db_url, connect_args={"check_same_thread": False}) with Session(engine) as session: session.execute( text( "INSERT INTO dsmr_reading (recorded_at, source_id, payload) " "VALUES (:ts, NULL, '{}')" ), {"ts": now.strftime("%Y-%m-%dT%H:%M:%S")}, ) session.commit() engine.dispose() # First upgrade to head. command.upgrade(alembic_cfg, "head") # Downgrade and re-upgrade to simulate idempotency (tests the guard condition). command.downgrade(alembic_cfg, "20260624_12_dsmr_decouple_telegram_id") command.upgrade(alembic_cfg, "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) with Session(engine) as session: meter_count = session.execute( text("SELECT COUNT(*) FROM meter WHERE reason = 'initial' AND ended_at IS NULL") ).scalar() assert meter_count == 1, ( f"Expected exactly 1 initial meter after repeated upgrade, got {meter_count}" ) engine.dispose() def test_migration_downgrade_removes_meter_table(tmp_path: Path): """Downgrading from M7-T01 must remove the meter table and the meter_id column.""" db_path = tmp_path / "downgrade_meter_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) command.upgrade(alembic_cfg, "head") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) assert "meter" in inspector.get_table_names(), "meter must exist before downgrade" ecp_cols = {col["name"] for col in inspector.get_columns("energy_cost_period")} assert "meter_id" in ecp_cols, "meter_id must exist in energy_cost_period before downgrade" engine.dispose() # Downgrade one step (removes the meter table + meter_id column). command.downgrade(alembic_cfg, "20260624_12_dsmr_decouple_telegram_id") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) assert "meter" not in inspector.get_table_names(), "meter must be gone after downgrade" ecp_cols_after = {col["name"] for col in inspector.get_columns("energy_cost_period")} assert "meter_id" not in ecp_cols_after, ( "meter_id must be removed from energy_cost_period after downgrade" ) engine.dispose()