From 28486a83c7764385fc18f45b69c61e2734a1defe Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sat, 22 Aug 2026 23:23:43 +0200 Subject: [PATCH] M8-T03: adopt DSMR history into meter sources --- .../20260822_16_dsmr_source_adoption.py | 269 +++++++++++++++++ app/models/energy.py | 44 ++- docs/design/m8-warmtelink-energy.md | 7 +- scripts/app_db_adopt.py | 2 +- tests/test_dsmr_source_migration.py | 285 ++++++++++++++++++ tests/test_energy_models.py | 44 ++- tests/test_meter_sources.py | 6 +- 7 files changed, 628 insertions(+), 29 deletions(-) create mode 100644 alembic_app/versions/20260822_16_dsmr_source_adoption.py create mode 100644 tests/test_dsmr_source_migration.py diff --git a/alembic_app/versions/20260822_16_dsmr_source_adoption.py b/alembic_app/versions/20260822_16_dsmr_source_adoption.py new file mode 100644 index 0000000..d601fef --- /dev/null +++ b/alembic_app/versions/20260822_16_dsmr_source_adoption.py @@ -0,0 +1,269 @@ +"""adopt historical DSMR rows into the source and binding model + +Revision ID: 20260822_16_dsmr_source_adoption +Revises: 20260822_15_meter_sources +Create Date: 2026-08-22 00:00:00.000000 + +The upgrade is deliberately data-preserving: it creates one migration-owned +DSMR source/channel, moves the telegram identifier to ``telegram_id``, and +audits every reading and cost row before committing. Old ``app_config`` rows, +payload JSON, and cost snapshots are never deleted or rewritten. +""" + +from __future__ import annotations + +import uuid +from datetime import datetime, timezone +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "20260822_16_dsmr_source_adoption" +down_revision: Union[str, None] = "20260822_15_meter_sources" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _as_bool(value: str | None) -> bool: + return value is not None and value.strip().lower() in {"1", "true", "yes", "on"} + + +def _iso_now() -> str: + return datetime.now(tz=timezone.utc).replace(tzinfo=None).isoformat(sep=" ") + + +def _config(connection: sa.Connection) -> dict[str, object]: + rows = connection.execute(sa.text("SELECT key, value FROM app_config")).all() + values = {str(key): str(value) for key, value in rows} + # An unconfigured historical DSMR installation needs a disabled identity, + # not guessed connection details. Preserve every legacy value we model + # when any legacy DSMR/MQTT configuration was explicitly present. + legacy_keys = { + "MQTT_BROKER_HOST", "MQTT_BROKER_PORT", "MQTT_USERNAME", "MQTT_PASSWORD", + "MQTT_TLS_ENABLED", "DSMR_MQTT_TOPIC", "DSMR_TARIFF_TOPIC", "DSMR_SAMPLE_INTERVAL_S", + } + if not legacy_keys & values.keys(): + return {} + return { + "broker_host": values.get("MQTT_BROKER_HOST", ""), + "broker_port": int(values.get("MQTT_BROKER_PORT", "1883")), + "username": values.get("MQTT_USERNAME", ""), + "password": values.get("MQTT_PASSWORD", ""), + "tls_enabled": _as_bool(values.get("MQTT_TLS_ENABLED")), + "topic": values.get("DSMR_MQTT_TOPIC", "dsmr/json"), + "tariff_topic": values.get("DSMR_TARIFF_TOPIC", "dsmr/meter-stats/electricity_tariff"), + "sample_interval_s": int(values.get("DSMR_SAMPLE_INTERVAL_S", "10")), + } + + +def _count(connection: sa.Connection, table: str) -> int: + return int(connection.execute(sa.text(f"SELECT COUNT(*) FROM {table}")).scalar_one()) + + +def upgrade() -> None: + connection = op.get_bind() + readings_before = _count(connection, "dsmr_reading") + costs_before = _count(connection, "energy_cost_period") + sources_before = _count(connection, "meter_source") + channels_before = _count(connection, "meter_source_channel") + bindings_before = _count(connection, "meter_source_binding") + now = _iso_now() + + # A source exists even without historical configuration/readings. It stays + # disabled unless the old explicit DSMR switch was enabled, so no broker or + # topic is guessed at runtime. + config = _config(connection) + source_result = connection.execute( + sa.text( + "INSERT INTO meter_source " + "(uuid, name, kind, enabled, config, status, last_seen_at, last_error, created_at, updated_at) " + "VALUES (:uuid, :name, 'dsmr_mqtt', :enabled, :config, 'unknown', NULL, NULL, :now, :now)" + ), + { + "uuid": str(uuid.uuid4()), + "name": "Migrated DSMR source", + "enabled": _as_bool( + connection.execute( + sa.text("SELECT value FROM app_config WHERE key = 'DSMR_INGEST_ENABLED'") + ).scalar_one_or_none() + ), + "config": __import__("json").dumps(config), + "now": now, + }, + ) + source_id = source_result.lastrowid + if source_id is None: + raise RuntimeError("DSMR source adoption failed to create a source") + channel_result = connection.execute( + sa.text( + "INSERT INTO meter_source_channel " + "(uuid, source_id, channel_key, label, suggested_commodity, unit, device_type, fingerprint, " + "latest_value, latest_at, latest_quality, created_at, updated_at) " + "VALUES (:uuid, :source_id, 'electricity-total', 'DSMR electricity total', 'electricity', " + "'kWh', NULL, NULL, NULL, NULL, NULL, :now, :now)" + ), + {"uuid": str(uuid.uuid4()), "source_id": source_id, "now": now}, + ) + channel_id = channel_result.lastrowid + if channel_id is None: + raise RuntimeError("DSMR source adoption failed to create an electricity channel") + if _count(connection, "meter_source") != sources_before + 1: + raise RuntimeError("DSMR source adoption source row-count audit failed") + if _count(connection, "meter_source_channel") != channels_before + 1: + raise RuntimeError("DSMR source adoption channel row-count audit failed") + + # Rename/add while nullable, back-fill all rows, then make the FK non-null + # and replace the legacy timestamp-only uniqueness in a SQLite batch rebuild. + with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: + batch_op.alter_column("source_id", new_column_name="telegram_id") + batch_op.add_column(sa.Column("meter_source_id", sa.Integer(), nullable=True)) + connection.execute( + sa.text("UPDATE dsmr_reading SET meter_source_id = :source_id WHERE meter_source_id IS NULL"), + {"source_id": source_id}, + ) + with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: + batch_op.drop_constraint("uq_dsmr_reading_recorded_at", type_="unique") + batch_op.alter_column("meter_source_id", existing_type=sa.Integer(), nullable=False) + batch_op.create_foreign_key( + "fk_dsmr_reading_meter_source_id", "meter_source", ["meter_source_id"], ["id"], + ondelete="RESTRICT", + ) + batch_op.create_unique_constraint( + "uq_dsmr_reading_source_recorded_at", ["meter_source_id", "recorded_at"] + ) + batch_op.create_index("ix_dsmr_reading_meter_source_id", ["meter_source_id"]) + adopted_readings = int( + connection.execute( + sa.text("SELECT COUNT(*) FROM dsmr_reading WHERE meter_source_id = :source_id"), + {"source_id": source_id}, + ).scalar_one() + ) + if adopted_readings != readings_before: + raise RuntimeError("DSMR source adoption reading source audit failed") + + # Bind each electricity meter only where it overlaps the actual DSMR data. + data_window = connection.execute( + sa.text("SELECT MIN(recorded_at), MAX(recorded_at) FROM dsmr_reading") + ).one() + expected_binding_count = 0 + if data_window[0] is not None: + meters = connection.execute( + sa.text( + "SELECT id, started_at, ended_at FROM meter WHERE commodity = 'electricity' " + "ORDER BY started_at, id" + ) + ).all() + for meter_id, started_at, ended_at in meters: + # Intersect [meter start, meter end) with the inclusive historical + # samples. A closed boundary at the final sample remains valid for + # the preceding interval; an empty intersection gets no fake binding. + if started_at > data_window[1] or (ended_at is not None and ended_at <= data_window[0]): + continue + expected_binding_count += 1 + binding_start = max(started_at, data_window[0]) + binding_end = ended_at + connection.execute( + sa.text( + "INSERT INTO meter_source_binding " + "(uuid, meter_id, channel_id, started_at, ended_at, created_at, updated_at) " + "VALUES (:uuid, :meter_id, :channel_id, :started_at, :ended_at, :now, :now)" + ), + { + "uuid": str(uuid.uuid4()), "meter_id": meter_id, "channel_id": channel_id, + "started_at": binding_start, "ended_at": binding_end, "now": now, + }, + ) + + # A cost period may be linked only if exactly one binding covers both its + # start and end. Historical boundary/unknown rows remain auditable but are + # explicitly degraded instead of being silently attributed to a current meter. + periods = connection.execute( + sa.text("SELECT id, meter_id, period_start, degraded FROM energy_cost_period") + ).all() + resolvable_normal_periods: dict[int, int] = {} + unresolved_period_ids: set[int] = set() + for period_id, meter_id, period_start, degraded_before in periods: + candidates = [] + if meter_id is not None: + candidates = connection.execute( + sa.text( + "SELECT id FROM meter_source_binding " + "WHERE meter_id = :meter_id AND started_at <= :start " + "AND (ended_at IS NULL OR julianday(ended_at) > julianday(:start, '+15 minutes'))" + ), + {"meter_id": meter_id, "start": period_start}, + ).all() + if len(candidates) == 1: + if not degraded_before: + resolvable_normal_periods[period_id] = candidates[0][0] + connection.execute( + sa.text("UPDATE energy_cost_period SET source_binding_id = :binding_id WHERE id = :id"), + {"binding_id": candidates[0][0], "id": period_id}, + ) + else: + unresolved_period_ids.add(period_id) + connection.execute( + sa.text( + "UPDATE energy_cost_period SET degraded = 1, source_binding_id = NULL WHERE id = :id" + ), + {"id": period_id}, + ) + + readings_after = _count(connection, "dsmr_reading") + costs_after = _count(connection, "energy_cost_period") + if readings_after != readings_before or costs_after != costs_before: + raise RuntimeError("DSMR source adoption row-count audit failed") + if _count(connection, "meter_source_binding") != bindings_before + expected_binding_count: + raise RuntimeError("DSMR source adoption binding row-count audit failed") + if int( + connection.execute( + sa.text("SELECT COUNT(*) FROM meter_source_binding WHERE channel_id = :channel_id"), + {"channel_id": channel_id}, + ).scalar_one() + ) != expected_binding_count: + raise RuntimeError("DSMR source adoption binding channel audit failed") + for period_id, binding_id in resolvable_normal_periods.items(): + bound, degraded = connection.execute( + sa.text("SELECT source_binding_id, degraded FROM energy_cost_period WHERE id = :id"), + {"id": period_id}, + ).one() + if bound != binding_id or degraded: + raise RuntimeError("DSMR source adoption resolvable cost audit failed") + if unresolved_period_ids: + unresolved_count = int( + connection.execute( + sa.text( + "SELECT COUNT(*) FROM energy_cost_period " + "WHERE id IN :period_ids AND (degraded != 1 OR source_binding_id IS NOT NULL)" + ).bindparams(sa.bindparam("period_ids", expanding=True)), + {"period_ids": list(unresolved_period_ids)}, + ).scalar_one() + ) + if unresolved_count: + raise RuntimeError("DSMR source adoption unresolved cost audit failed") + orphan_rows = connection.execute(sa.text("PRAGMA foreign_key_check")).all() + if orphan_rows: + raise RuntimeError("DSMR source adoption foreign-key audit failed") + normal_unbound = int( + connection.execute( + sa.text("SELECT COUNT(*) FROM energy_cost_period WHERE degraded = 0 AND source_binding_id IS NULL") + ).scalar_one() + ) + if normal_unbound: + raise RuntimeError(f"DSMR source adoption left {normal_unbound} normal cost period(s) unbound") + if _count(connection, "meter_source") < 1: + raise RuntimeError("DSMR source adoption source audit failed") + + +def downgrade() -> None: + # Schema-only downgrade for isolated test databases. It intentionally does + # not delete migration-created source/channel/binding rows. + with op.batch_alter_table("dsmr_reading", schema=None) as batch_op: + batch_op.drop_index("ix_dsmr_reading_meter_source_id") + batch_op.drop_constraint("uq_dsmr_reading_source_recorded_at", type_="unique") + batch_op.drop_constraint("fk_dsmr_reading_meter_source_id", type_="foreignkey") + batch_op.drop_column("meter_source_id") + batch_op.alter_column("telegram_id", new_column_name="source_id") + batch_op.create_unique_constraint("uq_dsmr_reading_recorded_at", ["recorded_at"]) diff --git a/app/models/energy.py b/app/models/energy.py index 2dcf195..0dfaae2 100644 --- a/app/models/energy.py +++ b/app/models/energy.py @@ -13,8 +13,8 @@ from __future__ import annotations import uuid as _uuid from datetime import datetime -from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String -from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy import Boolean, DateTime, Float, ForeignKey, Integer, String, UniqueConstraint, event, text +from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym from sqlalchemy.types import JSON from app.db import Base @@ -97,8 +97,9 @@ class DsmrReading(Base): (that field overflows and must be manually reset to zero — a known DSMR quirk — so relying on it for uniqueness risks silently dropping new data). The table's own autoincrement ``id`` PK is the stable internal identity, and - ``recorded_at`` (the telegram timestamp) is the UNIQUE de-duplication key: a - single P1 meter emits exactly one telegram per timestamp. + ``(meter_source_id, recorded_at)`` is the UNIQUE de-duplication key: each + configured P1 source emits at most one telegram per timestamp, while + different sources may legitimately emit at the same instant. ``recorded_at`` is a real column (not inside the payload) so time-range queries are efficient. The entire telegram frame is stored verbatim in @@ -110,21 +111,44 @@ class DsmrReading(Base): id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) - # UTC timestamp of the sample — real column, UNIQUE (telegram-id-independent - # idempotency key). The unique index also serves time-range queries. - recorded_at: Mapped[datetime] = mapped_column( - DateTime(timezone=True), nullable=False, unique=True - ) + # UTC timestamp of the sample. Idempotency is per configured source, so + # distinct P1 sources may legitimately emit at the same instant. + recorded_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) # Telegram's own id (DSMR Reader assigns it). Stored only as a reference / # debugging aid — NOT used for uniqueness or idempotency (it overflows and # gets reset to zero). Nullable because some DSMR sources may not emit one. - source_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + telegram_id: Mapped[int | None] = mapped_column(Integer, nullable=True) + + # Compatibility for the pre-M8 ingest implementation. This is an ORM + # alias only; the physical database column is ``telegram_id``. + source_id = synonym("telegram_id") + + # The configured source is the durable identity of the cumulative reading + # stream. It is non-null after the revision-16 historical adoption. + meter_source_id: Mapped[int] = mapped_column( + ForeignKey("meter_source.id", ondelete="RESTRICT"), nullable=False, index=True + ) # Full telegram frame as a JSON object; values are typically JSON strings # (e.g. "20915.154") — callers must cast to Decimal before arithmetic. payload: Mapped[dict] = mapped_column(JSON, nullable=False) + __table_args__ = ( + UniqueConstraint( + "meter_source_id", "recorded_at", name="uq_dsmr_reading_source_recorded_at" + ), + ) + + +@event.listens_for(DsmrReading, "before_insert") +def _supply_legacy_dsmr_source(_mapper, connection, target: DsmrReading) -> None: + """Keep the pre-T04 single-source writer working during the schema handoff.""" + if target.meter_source_id is None: + target.meter_source_id = connection.execute( + text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt' ORDER BY id LIMIT 1") + ).scalar_one() + class EnergyContract(Base): """Contract head: a named energy contract with a chosen pricing strategy. diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 548a561..641e619 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -363,7 +363,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T03 — 将 DSMR 历史迁入 Source / Binding [structural] -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T02 - **Context**: 先把既有电力链路安全迁到统一模型，之后才能让 runtime 和计费真正按 source 工作。 @@ -373,6 +373,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 - `modify scripts/app_db_adopt.py` - `create tests/test_dsmr_source_migration.py` - `modify tests/test_energy_models.py` +- `modify tests/test_meter_sources.py` **Steps** 1. 把 ORM `DsmrReading.source_id` 重命名为 nullable `telegram_id`，新增 non-null @@ -385,6 +386,8 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 5. 用 revision 14 的历史 fixture 覆盖单 Meter、多次换表、跨界成本、无读数 Meter 和无旧 config。 6. 把 `APP_BASELINE_REVISION` 同步到 revision 16；所有升级 fixture 仅在 `tmp_path` 中构造，不读取 真实 app DB 或 volume。 +7. 把 T01 自身的 revision 14→15 schema 对账测试固定升级到明确 revision 15，不以可继续前进的 + `head` 作为 T01 终点；T03 的新 fixture 单独负责 revision 14/15→16 历史回填对账。 **Out of scope / 不要碰** - 不删除旧 DSMR `app_config` 行，不改变 MQTT subscription，不改 API 响应。 @@ -395,11 +398,13 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 - [ ] 两个 source 可在同一 timestamp 各存一条 DSMR reading；同 source 重复 timestamp 被拒绝。 - [ ] `telegram_id` 不参与幂等唯一键，旧 telegram id 完整保留。 - [ ] `APP_BASELINE_REVISION` 等于唯一 revision 16 head，升级重复运行幂等。 +- [ ] T01 schema-only fixture 固定停在 revision 15，T03 fixture 到 revision 16，二者职责不随 head 漂移。 - [ ] 历史升级 fixture、空库升级、`pytest`、`ruff check .` 全绿。 **Reviewer checklist** - 对账是否在 migration 中真实执行，而不只是测试断言；失败能否原子回滚。 - baseline 常量是否随 revision 16 同步，fixture 是否完全隔离于真实生产路径。 +- 前序 migration 测试是否使用明确 revision 边界，而非把历史阶段误写为永久 `head`。 - 是否存在“把所有历史强绑当前 Meter/source”的静默错误或任何 destructive cleanup。 ### M8-T04 — DSMR runtime 改为多 Source 配置 [structural] diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index b8afe86..f9c3101 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 = "20260822_15_meter_sources" +APP_BASELINE_REVISION = "20260822_16_dsmr_source_adoption" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_dsmr_source_migration.py b/tests/test_dsmr_source_migration.py new file mode 100644 index 0000000..d6f0817 --- /dev/null +++ b/tests/test_dsmr_source_migration.py @@ -0,0 +1,285 @@ +"""Isolated revision-14/15 fixtures for the DSMR source-adoption migration.""" + +from __future__ import annotations + +import json +from datetime import datetime, timedelta, 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, inspect, text + + +def _config(database_url: str) -> Config: + config = Config("alembic_app.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +def _engine(database_url: str): + engine = create_engine(database_url, connect_args={"check_same_thread": False}) + + @event.listens_for(engine, "connect") + def _foreign_keys(connection, _record) -> None: + connection.execute("PRAGMA foreign_keys=ON") + + return engine + + +def _insert_meter(connection, label: str, started: datetime, ended: datetime | None) -> int: + connection.execute( + text( + "INSERT INTO meter (uuid, label, commodity, started_at, ended_at, reason, note, created_at) " + "VALUES (:uuid, :label, 'electricity', :started, :ended, 'initial', NULL, :started)" + ), + {"uuid": f"{label:0<8}-0000-4000-8000-000000000000", "label": label, + "started": started, "ended": ended}, + ) + return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one()) + + +def _insert_contract_version(connection, start: datetime) -> int: + connection.execute( + text( + "INSERT INTO energy_contract (name, kind, active, currency, created_at, updated_at) " + "VALUES ('Historic contract', 'manual', 1, 'EUR', :at, :at)" + ), + {"at": start}, + ) + contract_id = int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one()) + connection.execute( + text( + "INSERT INTO energy_contract_version " + "(contract_id, effective_from, effective_to, \"values\", created_at) " + "VALUES (:contract_id, :at, NULL, :values, :at)" + ), + {"contract_id": contract_id, "at": start, "values": json.dumps({"historic": True})}, + ) + return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one()) + + +def _insert_cost( + connection, + period_start: datetime, + meter_id: int, + contract_version_id: int, + sequence: int, +) -> None: + amount = 2.5 + sequence + connection.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, meter_id, degraded, computed_at) " + "VALUES (:start, :d1, :d2, :r1, :r2, :import_cost, :export_revenue, :net_cost, " + "'EUR', :pricing, :contract_version_id, :meter, 0, :computed_at)" + ), + { + "start": period_start, + "d1": 1.0 + sequence, + "d2": 2.0 + sequence, + "r1": 3.0 + sequence, + "r2": 4.0 + sequence, + "import_cost": amount, + "export_revenue": 0.25 + sequence, + "net_cost": amount - (0.25 + sequence), + "pricing": json.dumps({"historic": True, "sequence": sequence}), + "contract_version_id": contract_version_id, + "meter": meter_id, + "computed_at": period_start + timedelta(seconds=sequence), + }, + ) + + +def test_populated_revision_14_adopts_dsmr_history_at_revision_16(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'revision_14_history.db'}" + config = _config(database_url) + command.upgrade(config, "20260625_14_meter_uuid") + start = datetime(2026, 8, 1, tzinfo=timezone.utc) + engine = _engine(database_url) + try: + with engine.begin() as connection: + connection.execute( + text("INSERT INTO app_config (key, value, updated_at) VALUES (:key, :value, :at)"), + [ + {"key": "DSMR_INGEST_ENABLED", "value": "true", "at": start}, + {"key": "DSMR_MQTT_TOPIC", "value": "historic/dsmr", "at": start}, + {"key": "DSMR_TARIFF_TOPIC", "value": "historic/tariff", "at": start}, + {"key": "DSMR_SAMPLE_INTERVAL_S", "value": "15", "at": start}, + {"key": "MQTT_BROKER_HOST", "value": "mqtt.example.invalid", "at": start}, + {"key": "MQTT_BROKER_PORT", "value": "1884", "at": start}, + {"key": "MQTT_USERNAME", "value": "historic-user", "at": start}, + {"key": "MQTT_PASSWORD", "value": "historic-password", "at": start}, + {"key": "MQTT_TLS_ENABLED", "value": "true", "at": start}, + {"key": "UNRELATED_CONFIG", "value": "untouched", "at": start}, + ], + ) + config_before = dict(connection.execute(text("SELECT key, value FROM app_config")).all()) + for offset, telegram_id in ((0, 77), (20, 78), (40, 77)): + connection.execute( + text("INSERT INTO dsmr_reading (recorded_at, source_id, payload) VALUES (:at, :id, :payload)"), + {"at": start + timedelta(minutes=offset), "id": telegram_id, + "payload": json.dumps({"id": telegram_id, "keep": f"payload-{offset}"})}, + ) + first = _insert_meter(connection, "meterone", start - timedelta(hours=1), start + timedelta(minutes=20)) + second = _insert_meter(connection, "metertwo", start + timedelta(minutes=20), start + timedelta(minutes=40)) + third = _insert_meter(connection, "meterthree", start + timedelta(minutes=40), None) + _insert_meter(connection, "nodata", start + timedelta(days=1), None) + contract_version = _insert_contract_version(connection, start - timedelta(days=1)) + # One normal period per epoch plus a period ending exactly at each + # replacement boundary. Meter/binding intervals are half-open, so + # the latter must remain unbound/degraded. + _insert_cost(connection, start, first, contract_version, 0) + _insert_cost(connection, start + timedelta(minutes=5), first, contract_version, 1) + _insert_cost(connection, start + timedelta(minutes=21), second, contract_version, 2) + _insert_cost(connection, start + timedelta(minutes=25), second, contract_version, 3) + _insert_cost(connection, start + timedelta(minutes=41), third, contract_version, 4) + cost_before = connection.execute( + text( + "SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, " + "export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, " + "degraded, computed_at FROM energy_cost_period ORDER BY period_start" + ) + ).mappings().all() + finally: + engine.dispose() + + command.upgrade(config, "20260822_16_dsmr_source_adoption") + engine = _engine(database_url) + try: + with engine.connect() as connection: + assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == ( + "20260822_16_dsmr_source_adoption" + ) + assert connection.execute(text("SELECT COUNT(*) FROM dsmr_reading")).scalar_one() == 3 + source = connection.execute( + text("SELECT id, enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'") + ).one() + assert source.enabled == 1 + assert json.loads(source.config) == { + "broker_host": "mqtt.example.invalid", "broker_port": 1884, + "username": "historic-user", "password": "historic-password", "tls_enabled": True, + "topic": "historic/dsmr", "tariff_topic": "historic/tariff", "sample_interval_s": 15, + } + assert connection.execute(text("SELECT value FROM app_config WHERE key = 'DSMR_MQTT_TOPIC'")).scalar_one() == "historic/dsmr" + assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == config_before + assert connection.execute(text("SELECT group_concat(telegram_id) FROM dsmr_reading")).scalar_one() == "77,78,77" + assert connection.execute(text("SELECT payload FROM dsmr_reading ORDER BY recorded_at")).scalars().all() == [ + json.dumps({"id": 77, "keep": "payload-0"}), + json.dumps({"id": 78, "keep": "payload-20"}), + json.dumps({"id": 77, "keep": "payload-40"}), + ] + assert connection.execute(text("SELECT COUNT(*) FROM meter_source_binding")).scalar_one() == 3 + assert connection.execute( + text("SELECT COUNT(*) FROM meter_source_binding WHERE meter_id = (SELECT id FROM meter WHERE label = 'nodata')") + ).scalar_one() == 0 + periods = connection.execute( + text( + "SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, " + "export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, " + "degraded, computed_at, source_binding_id FROM energy_cost_period ORDER BY period_start" + ) + ).mappings().all() + assert [ + {key: value for key, value in period.items() if key not in {"degraded", "source_binding_id"}} + for period in periods + ] == [ + {key: value for key, value in period.items() if key != "degraded"} + for period in cost_before + ] + assert [(period["degraded"], period["source_binding_id"] is not None) for period in periods] == [ + (0, True), (1, False), (0, True), (1, False), (0, True) + ] + assert [period["period_start"] for period in periods if period["source_binding_id"] is None] == [ + (start + timedelta(minutes=5)).isoformat(sep=" "), + (start + timedelta(minutes=25)).isoformat(sep=" "), + ] + bound_meter_ids = connection.execute( + text( + "SELECT binding.meter_id FROM energy_cost_period AS period " + "LEFT JOIN meter_source_binding AS binding " + "ON binding.id = period.source_binding_id ORDER BY period.period_start" + ) + ).scalars().all() + assert bound_meter_ids == [first, None, second, None, third] + assert connection.execute(text("PRAGMA foreign_key_check")).all() == [] + finally: + engine.dispose() + + +def test_dsmr_source_timestamp_uniqueness_allows_two_sources(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'two_sources.db'}" + config = _config(database_url) + command.upgrade(config, "20260822_16_dsmr_source_adoption") + engine = _engine(database_url) + timestamp = datetime(2026, 8, 1, tzinfo=timezone.utc) + try: + with engine.begin() as connection: + first_source = connection.execute(text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt'")).scalar_one() + assert connection.execute( + text("SELECT enabled FROM meter_source WHERE id = :id"), {"id": first_source} + ).scalar_one() == 0 + connection.execute( + text( + "INSERT INTO meter_source (uuid, name, kind, enabled, config, status, created_at, updated_at) " + "VALUES ('22222222-2222-4222-8222-222222222222', 'Second DSMR', 'dsmr_mqtt', 0, '{}', " + "'unknown', :at, :at)" + ), {"at": timestamp}, + ) + second_source = connection.execute(text("SELECT last_insert_rowid()")).scalar_one() + for source_id in (first_source, second_source): + connection.execute( + text( + "INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) " + "VALUES (:at, 9, :source, '{}')" + ), {"at": timestamp, "source": source_id}, + ) + with pytest.raises(sqlalchemy.exc.IntegrityError): + connection.execute( + text( + "INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) " + "VALUES (:at, 10, :source, '{}')" + ), {"at": timestamp, "source": first_source}, + ) + finally: + engine.dispose() + inspector = inspect(create_engine(database_url)) + assert ("meter_source_id", "recorded_at") in { + tuple(item["column_names"]) for item in inspector.get_unique_constraints("dsmr_reading") + } + + +def test_revision_15_without_legacy_dsmr_config_creates_unconfigured_source(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'revision_15_no_dsmr_config.db'}" + config = _config(database_url) + command.upgrade(config, "20260822_15_meter_sources") + engine = _engine(database_url) + try: + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO app_config (key, value, updated_at) " + "VALUES ('UNRELATED_CONFIG', 'untouched', :at)" + ), + {"at": datetime(2026, 8, 1, tzinfo=timezone.utc)}, + ) + finally: + engine.dispose() + + command.upgrade(config, "20260822_16_dsmr_source_adoption") + engine = _engine(database_url) + try: + with engine.connect() as connection: + source = connection.execute( + text("SELECT enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'") + ).one() + assert source.enabled == 0 + assert json.loads(source.config) == {} + assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == { + "UNRELATED_CONFIG": "untouched" + } + finally: + engine.dispose() diff --git a/tests/test_energy_models.py b/tests/test_energy_models.py index e67f801..25a0a16 100644 --- a/tests/test_energy_models.py +++ b/tests/test_energy_models.py @@ -21,7 +21,7 @@ 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 import UniqueConstraint, create_engine, event as sa_event, inspect, text from sqlalchemy.orm import Session from app.db import Base @@ -114,31 +114,42 @@ def test_energy_tables_exist_after_upgrade(energy_db): def test_dsmr_reading_columns(energy_db): - """dsmr_reading must have id, recorded_at (NOT NULL), source_id (nullable), payload (NOT NULL).""" + """dsmr_reading stores its telegram id separately from its source identity.""" 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 "telegram_id" in columns and columns["telegram_id"]["nullable"] + assert "meter_source_id" in columns and not columns["meter_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).""" +def test_dsmr_reading_source_timestamp_unique(energy_db): + """DSMR de-duplication is unique per configured source and timestamp.""" 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" + assert ("meter_source_id", "recorded_at") in { + tuple(uc["column_names"]) for uc in unique_constraints + } + foreign_keys = { + tuple(foreign_key["constrained_columns"]): foreign_key + for foreign_key in inspector.get_foreign_keys("dsmr_reading") + } + assert foreign_keys[("meter_source_id",)]["referred_table"] == "meter_source" + assert foreign_keys[("meter_source_id",)]["options"]["ondelete"] == "RESTRICT" + assert "ix_dsmr_reading_meter_source_id" in { + index["name"] for index in inspector.get_indexes("dsmr_reading") + } -def test_dsmr_reading_source_id_not_unique(energy_db): - """dsmr_reading.source_id (telegram id) must NOT be unique — it overflows/resets, +def test_dsmr_reading_telegram_id_not_unique(energy_db): + """dsmr_reading.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" + assert "telegram_id" not in unique_cols, "telegram_id must NOT have a unique constraint" def test_energy_contract_columns(energy_db): @@ -410,12 +421,15 @@ def test_energy_cost_period_meter_id_fk_ondelete_restrict(): ) -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).""" +def test_dsmr_reading_source_timestamp_unique_in_metadata(): + """DsmrReading de-duplicates by source/timestamp, never 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" + assert not table.columns["telegram_id"].unique, "telegram_id must NOT be unique" + assert any( + tuple(constraint.columns.keys()) == ("meter_source_id", "recorded_at") + for constraint in table.constraints + if isinstance(constraint, UniqueConstraint) + ) def test_tibber_price_starts_at_unique_in_metadata(): diff --git a/tests/test_meter_sources.py b/tests/test_meter_sources.py index 4b2aac1..0f74eee 100644 --- a/tests/test_meter_sources.py +++ b/tests/test_meter_sources.py @@ -139,7 +139,9 @@ def test_populated_revision_14_upgrades_to_meter_source_head_with_audit(tmp_path finally: engine.dispose() - command.upgrade(config, "head") + # This T01 fixture intentionally audits the schema-only revision 15. + # Revision 16 has its own DSMR-history adoption fixture. + command.upgrade(config, "20260822_15_meter_sources") engine = _engine_with_foreign_keys(database_url) try: @@ -187,7 +189,7 @@ def test_populated_revision_14_upgrades_to_meter_source_head_with_audit(tmp_path assert cost_fks["contract_version_id"]["referred_table"] == "energy_contract_version" assert cost_fks["source_binding_id"]["referred_table"] == "meter_source_binding" - command.upgrade(config, "head") + command.upgrade(config, "20260822_15_meter_sources") assert { table_name: engine.connect().execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar_one() for table_name in before_counts