From 32a20785ee1c0efcad0f6c4df90892318469be70 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Thu, 25 Jun 2026 14:48:39 +0200 Subject: [PATCH] M7-T01: add meter table + model + energy_cost_period.meter_id + backfill migration --- .../versions/20260625_13_meter_table.py | 238 +++++++++ app/models/energy.py | 70 ++- scripts/app_db_adopt.py | 2 +- tests/test_energy_models.py | 452 +++++++++++++++++- 4 files changed, 751 insertions(+), 11 deletions(-) create mode 100644 alembic_app/versions/20260625_13_meter_table.py diff --git a/alembic_app/versions/20260625_13_meter_table.py b/alembic_app/versions/20260625_13_meter_table.py new file mode 100644 index 0000000..56020ad --- /dev/null +++ b/alembic_app/versions/20260625_13_meter_table.py @@ -0,0 +1,238 @@ +"""add meter table and energy_cost_period.meter_id + +Introduces the ``meter`` table (one row per physical meter installation epoch) +and a nullable FK column ``energy_cost_period.meter_id`` that attributes each +billing period to a specific physical meter. + +**Backfill logic (§3.7 of the M7 design doc)**: + +If the database already contains any ``dsmr_reading`` or +``energy_cost_period`` rows, one initial ``meter`` row is created: + + label = "Initial meter" + commodity = "electricity" + started_at = earliest dsmr_reading.recorded_at + (or, if none, earliest energy_cost_period.period_start, + or, if still none, the migration timestamp) + ended_at = NULL (still active) + reason = "initial" + +All existing ``energy_cost_period`` rows are then back-filled with that +initial meter's id. + +**Idempotency**: the backfill is guarded with a check for any existing +``meter`` row whose ``reason = 'initial'`` and ``commodity = 'electricity'`` +and ``ended_at IS NULL``, so repeating the upgrade does not create duplicate +meters or overwrite already-filled meter_id values. + +**Audit (on-non-degraded periods only)**: after backfilling, the number of +non-degraded ``energy_cost_period`` rows with ``meter_id IS NULL`` must be +zero; if it is not, the migration raises a ``RuntimeError`` and rolls back. + +**Data safety**: this migration is additive only — no existing rows are +deleted or overwritten; it only creates a new table, adds a nullable column, +and back-fills that column. + +Revision ID: 20260625_13_meter_table +Revises: 20260624_12_dsmr_decouple_telegram_id +Create Date: 2026-06-25 00:00:00.000000 +""" + +from datetime import datetime, timezone +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + +revision: str = "20260625_13_meter_table" +down_revision: Union[str, None] = "20260624_12_dsmr_decouple_telegram_id" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # ------------------------------------------------------------------ # + # 1. Create the meter table. # + # ------------------------------------------------------------------ # + op.create_table( + "meter", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("label", sa.String(length=255), nullable=False), + sa.Column("commodity", sa.String(length=32), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("reason", sa.String(length=64), nullable=False), + sa.Column("note", sa.String(length=1024), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + + # ------------------------------------------------------------------ # + # 2. Add meter_id column to energy_cost_period (nullable FK). # + # ------------------------------------------------------------------ # + with op.batch_alter_table("energy_cost_period", schema=None) as batch_op: + batch_op.add_column( + sa.Column("meter_id", sa.Integer(), nullable=True) + ) + batch_op.create_foreign_key( + "fk_energy_cost_period_meter_id", + "meter", + ["meter_id"], + ["id"], + ondelete="RESTRICT", + ) + + # ------------------------------------------------------------------ # + # 3. Backfill initial meter (idempotent). # + # ------------------------------------------------------------------ # + conn = op.get_bind() + + # Check if there is already an initial meter (idempotency guard). + existing_initial = conn.execute( + sa.text( + "SELECT id FROM meter " + "WHERE reason = 'initial' AND commodity = 'electricity' AND ended_at IS NULL " + "LIMIT 1" + ) + ).fetchone() + + if existing_initial is not None: + # Already backfilled — nothing to do. + return + + # Determine whether there is any historical data to create a meter for. + has_readings = conn.execute( + sa.text("SELECT 1 FROM dsmr_reading LIMIT 1") + ).fetchone() + has_periods = conn.execute( + sa.text("SELECT 1 FROM energy_cost_period LIMIT 1") + ).fetchone() + + if not has_readings and not has_periods: + # Empty database: no historical data, so no initial meter is needed. + # meter_id will remain NULL on any future rows until T02 service layer + # starts populating it. + return + + # Determine started_at: earliest dsmr_reading.recorded_at, falling back to + # earliest energy_cost_period.period_start, and finally to now(). + earliest_reading_row = conn.execute( + sa.text("SELECT MIN(recorded_at) AS ts FROM dsmr_reading") + ).fetchone() + earliest_period_row = conn.execute( + sa.text("SELECT MIN(period_start) AS ts FROM energy_cost_period") + ).fetchone() + + started_at_value: datetime | None = None + if earliest_reading_row and earliest_reading_row[0] is not None: + # SQLite returns ISO strings for datetime columns; parse to datetime. + raw = earliest_reading_row[0] + started_at_value = _parse_sqlite_datetime(raw) + if started_at_value is None and earliest_period_row and earliest_period_row[0] is not None: + raw = earliest_period_row[0] + started_at_value = _parse_sqlite_datetime(raw) + if started_at_value is None: + started_at_value = datetime.now(tz=timezone.utc) + + now_utc = datetime.now(tz=timezone.utc) + + # Insert the initial meter row. + conn.execute( + sa.text( + "INSERT INTO meter (label, commodity, started_at, ended_at, reason, note, created_at) " + "VALUES (:label, :commodity, :started_at, NULL, :reason, NULL, :created_at)" + ), + { + "label": "Initial meter", + "commodity": "electricity", + "started_at": _iso(started_at_value), + "reason": "initial", + "created_at": _iso(now_utc), + }, + ) + + # Retrieve the newly created meter id. + meter_row = conn.execute( + sa.text( + "SELECT id FROM meter " + "WHERE reason = 'initial' AND commodity = 'electricity' AND ended_at IS NULL " + "LIMIT 1" + ) + ).fetchone() + assert meter_row is not None, "Initial meter row not found after insert" + meter_id: int = meter_row[0] + + # Back-fill all existing energy_cost_period rows that have meter_id IS NULL. + conn.execute( + sa.text( + "UPDATE energy_cost_period SET meter_id = :mid WHERE meter_id IS NULL" + ), + {"mid": meter_id}, + ) + + # ------------------------------------------------------------------ # + # 4. Audit: verify all non-degraded periods have a meter_id. # + # ------------------------------------------------------------------ # + unmatched_row = conn.execute( + sa.text( + "SELECT COUNT(*) FROM energy_cost_period " + "WHERE meter_id IS NULL AND degraded = 0" + ) + ).fetchone() + unmatched_count: int = unmatched_row[0] if unmatched_row else 0 + + if unmatched_count != 0: + raise RuntimeError( + f"Meter backfill audit failed: {unmatched_count} non-degraded " + "energy_cost_period row(s) still have meter_id IS NULL after backfill. " + "Migration aborted to protect data integrity." + ) + + +def downgrade() -> None: + # Remove the FK column from energy_cost_period first (references meter). + with op.batch_alter_table("energy_cost_period", schema=None) as batch_op: + batch_op.drop_constraint("fk_energy_cost_period_meter_id", type_="foreignkey") + batch_op.drop_column("meter_id") + + # Drop the meter table. + op.drop_table("meter") + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _parse_sqlite_datetime(value: str | datetime) -> datetime: + """Parse a SQLite datetime value into an aware UTC datetime. + + SQLite stores datetimes as ISO 8601 strings. SQLAlchemy may return them + as plain strings or as naive datetimes (no tzinfo) depending on the driver + and column declaration. This helper normalises both forms to an aware UTC + ``datetime``. + """ + if isinstance(value, datetime): + if value.tzinfo is None: + return value.replace(tzinfo=timezone.utc) + return value + # String form — strip trailing Z or +00:00 variants, then attach UTC. + s = str(value).strip() + for suffix in ("+00:00", "Z", " UTC"): + if s.endswith(suffix): + s = s[: -len(suffix)] + # SQLite uses space as the T separator in some formats. + s = s.replace(" ", "T") + try: + dt = datetime.fromisoformat(s) + except ValueError: + # Fallback: strip subseconds if present to handle unusual formats. + dt = datetime.strptime(s[:19], "%Y-%m-%dT%H:%M:%S") + return dt.replace(tzinfo=timezone.utc) + + +def _iso(dt: datetime) -> str: + """Serialise a datetime to an ISO 8601 string for SQLite storage.""" + if dt.tzinfo is not None: + dt = dt.astimezone(timezone.utc).replace(tzinfo=None) + return dt.strftime("%Y-%m-%dT%H:%M:%S") diff --git a/app/models/energy.py b/app/models/energy.py index a9ecbc7..2fd8947 100644 --- a/app/models/energy.py +++ b/app/models/energy.py @@ -1,6 +1,7 @@ """SQLAlchemy models for the energy pricing and DSMR metering subsystem. -Five tables: +Six tables: + - meter: physical electricity meter lifecycle epoch. - dsmr_reading: raw DSMR telegram blobs (10-second down-sampled). - energy_contract: contract head (manual or tibber, one active at a time). - energy_contract_version: versioned pricing values; append-only for auditability. @@ -19,6 +20,62 @@ from sqlalchemy.types import JSON from app.db import Base +class Meter(Base): + """One physical electricity meter's installation epoch. + + A ``meter`` record represents the period ``[started_at, ended_at)`` during + which a particular physical meter was installed and active. Replacing a meter + (swap, home move, etc.) is modelled by closing the current record + (``ended_at = swap_timestamp``) and opening a new one + (``started_at = swap_timestamp``). + + **Invariant**: for each ``commodity`` there is at most one active meter + (``ended_at IS NULL``) at any point in time. The service layer enforces + this — no DB-level constraint is added to keep the migration simple and to + allow the application to return a meaningful error message. + + ``commodity`` defaults to ``"electricity"``; the field is a free-form string + (no CHECK constraint) so future commodities (``gas``, ``heating``) can be + added without a schema change. + + ``reason`` captures why this epoch started — one of ``initial``, + ``meter_swap``, ``home_move``, or ``other`` — stored as a plain string so + the application layer controls the allowed set. + """ + + __tablename__ = "meter" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # Human-readable label for this physical meter (e.g. address, serial, tariff zone). + label: Mapped[str] = mapped_column(String(255), nullable=False) + + # Energy commodity this meter measures. Defaults to "electricity". + commodity: Mapped[str] = mapped_column(String(32), nullable=False, default="electricity") + + # UTC timestamp when this meter epoch starts (inclusive). May be in the past + # (retroactive declaration); effective billing start = max(started_at, data start). + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + # UTC timestamp when this meter epoch ends (exclusive). NULL = currently active. + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + + # Why this epoch was created. Application-layer validation enforces the + # allowed set; no CHECK constraint to keep migrations simple. + reason: Mapped[str] = mapped_column(String(64), nullable=False) + + # Free-form note (e.g. location, physical meter id, reason details). + note: Mapped[str | None] = mapped_column(String(1024), nullable=True) + + # UTC timestamp of when this row was created. + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + # Relationship to cost periods attributed to this meter epoch (not loaded eagerly). + cost_periods: Mapped[list["EnergyCostPeriod"]] = relationship( + back_populates="meter", cascade="save-update, merge" + ) + + class DsmrReading(Base): """One down-sampled DSMR telegram stored as a full JSON blob. @@ -217,6 +274,14 @@ class EnergyCostPeriod(Base): ForeignKey("energy_contract_version.id", ondelete="RESTRICT"), nullable=True ) + # FK to the meter epoch this period belongs to. RESTRICT prevents deletion of + # a meter that still has attributed cost periods. Nullable for backwards + # compatibility (pre-M7 rows) and degraded periods where the meter was not + # determinable. + meter_id: Mapped[int | None] = mapped_column( + ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True + ) + # True when the period was computed with incomplete data (missing readings or # missing price); serves as a flag for later recomputation. degraded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) @@ -229,6 +294,9 @@ class EnergyCostPeriod(Base): back_populates="cost_periods" ) + # Relationship back to the meter epoch. + meter: Mapped["Meter | None"] = relationship(back_populates="cost_periods") + # Index on recorded_at for efficient time-range queries on DSMR readings. # (The ORM-level index=True on recorded_at already creates ix_dsmr_reading_recorded_at; diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index 70f0cad..4daa477 100644 --- a/scripts/app_db_adopt.py +++ b/scripts/app_db_adopt.py @@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: from app.config import get_settings -APP_BASELINE_REVISION = "20260624_12_dsmr_decouple_telegram_id" +APP_BASELINE_REVISION = "20260625_13_meter_table" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_energy_models.py b/tests/test_energy_models.py index de86793..d889de7 100644 --- a/tests/test_energy_models.py +++ b/tests/test_energy_models.py @@ -1,13 +1,15 @@ -"""Tests for M6-T01: energy pricing and DSMR metering tables and ORM models. +"""Tests for energy pricing, DSMR metering, and meter epoch tables and ORM models. Covers: - 1. Migration shape: upgrade to head creates all five tables with correct columns, + 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 five tables; FKs are RESTRICT; + 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 @@ -27,6 +29,7 @@ from app.models.energy import ( DsmrReading, EnergyContract, EnergyContractVersion, + Meter, TibberPrice, EnergyCostPeriod, ) @@ -92,10 +95,11 @@ def energy_db_with_fk(tmp_path: Path): def test_energy_tables_exist_after_upgrade(energy_db): - """All five energy tables must exist after upgrade to head.""" + """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", @@ -200,7 +204,7 @@ def test_energy_cost_period_columns(energy_db): "import_cost", "export_revenue", "net_cost", "currency", "pricing", "degraded", "computed_at", } - nullable = {"contract_version_id"} + nullable = {"contract_version_id", "meter_id"} for col_name in non_nullable: assert col_name in columns, f"Missing column: {col_name}" @@ -234,10 +238,35 @@ 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") - assert len(fks) == 1, f"Expected 1 FK on energy_cost_period, got {len(fks)}" - fk = fks[0] + # 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 "contract_version_id" in fk["constrained_columns"] + 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"] @@ -308,8 +337,9 @@ def test_dsmr_decouple_migration_downgrade_restores_source_id_unique(tmp_path: P def test_base_metadata_contains_energy_tables(): - """Base.metadata.tables must include all five new energy tables.""" + """Base.metadata.tables must include all six energy tables.""" expected = { + "meter", "dsmr_reading", "energy_contract", "energy_contract_version", @@ -366,6 +396,17 @@ def test_energy_cost_period_fk_ondelete_restrict(): ) +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).""" @@ -739,3 +780,396 @@ def test_cost_period_restrict_prevents_version_deletion(tmp_path: Path): 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()