From 009856a50d80716cd13e64086a7dc90d012a5e66 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sat, 22 Aug 2026 22:31:30 +0200 Subject: [PATCH] M8-T01: add meter source identity schema --- alembic_app/env.py | 1 + .../versions/20260822_15_meter_sources.py | 107 +++++ app/models/__init__.py | 4 + app/models/energy.py | 16 +- app/models/meter_source.py | 127 ++++++ docs/design/m8-warmtelink-energy.md | 2 +- scripts/app_db_adopt.py | 2 +- tests/test_energy_models.py | 4 + tests/test_meter_sources.py | 384 ++++++++++++++++++ 9 files changed, 644 insertions(+), 3 deletions(-) create mode 100644 alembic_app/versions/20260822_15_meter_sources.py create mode 100644 app/models/meter_source.py create mode 100644 tests/test_meter_sources.py diff --git a/alembic_app/env.py b/alembic_app/env.py index 8ba084c..2c36639 100644 --- a/alembic_app/env.py +++ b/alembic_app/env.py @@ -20,6 +20,7 @@ from app.models.energy import ( # noqa: F401 TibberPrice, EnergyCostPeriod, ) +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel # noqa: F401 config = context.config diff --git a/alembic_app/versions/20260822_15_meter_sources.py b/alembic_app/versions/20260822_15_meter_sources.py new file mode 100644 index 0000000..9d46485 --- /dev/null +++ b/alembic_app/versions/20260822_15_meter_sources.py @@ -0,0 +1,107 @@ +"""add protocol-agnostic meter source, channel, and binding tables + +Revision ID: 20260822_15_meter_sources +Revises: 20260625_14_meter_uuid +Create Date: 2026-08-22 00:00:00.000000 + +This revision is additive on upgrade. It deliberately does not backfill +existing DSMR data; that adoption is a later, separately audited migration. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "20260822_15_meter_sources" +down_revision: Union[str, None] = "20260625_14_meter_uuid" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.create_table( + "meter_source", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("uuid", sa.String(length=36), nullable=False), + sa.Column("name", sa.String(length=255), nullable=False), + sa.Column("kind", sa.String(length=64), nullable=False), + sa.Column("enabled", sa.Boolean(), nullable=False), + sa.Column("config", sa.JSON(), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("last_seen_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_error", sa.String(length=1024), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("uuid", name="uq_meter_source_uuid"), + ) + op.create_index("ix_meter_source_kind_enabled", "meter_source", ["kind", "enabled"]) + + op.create_table( + "meter_source_channel", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("uuid", sa.String(length=36), nullable=False), + sa.Column("source_id", sa.Integer(), nullable=False), + sa.Column("channel_key", sa.String(length=128), nullable=False), + sa.Column("label", sa.String(length=255), nullable=False), + sa.Column("suggested_commodity", sa.String(length=32), nullable=True), + sa.Column("unit", sa.String(length=32), nullable=False), + sa.Column("device_type", sa.String(length=64), nullable=True), + sa.Column("fingerprint", sa.String(length=64), nullable=True), + sa.Column("latest_value", sa.Numeric(precision=20, scale=6), nullable=True), + sa.Column("latest_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("latest_quality", sa.String(length=32), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["source_id"], ["meter_source.id"], ondelete="RESTRICT"), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("uuid", name="uq_meter_source_channel_uuid"), + sa.UniqueConstraint("source_id", "channel_key", name="uq_meter_source_channel_source_key"), + ) + op.create_index("ix_meter_source_channel_source_id", "meter_source_channel", ["source_id"]) + + op.create_table( + "meter_source_binding", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("uuid", sa.String(length=36), nullable=False), + sa.Column("meter_id", sa.Integer(), nullable=False), + sa.Column("channel_id", sa.Integer(), nullable=False), + sa.Column("started_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("ended_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["meter_id"], ["meter.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint( + ["channel_id"], ["meter_source_channel.id"], ondelete="RESTRICT" + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("uuid", name="uq_meter_source_binding_uuid"), + ) + op.create_index("ix_meter_source_binding_meter_id", "meter_source_binding", ["meter_id"]) + op.create_index("ix_meter_source_binding_channel_id", "meter_source_binding", ["channel_id"]) + + with op.batch_alter_table("energy_cost_period", schema=None) as batch_op: + batch_op.add_column(sa.Column("source_binding_id", sa.Integer(), nullable=True)) + batch_op.create_foreign_key( + "fk_energy_cost_period_source_binding_id", + "meter_source_binding", + ["source_binding_id"], + ["id"], + ondelete="RESTRICT", + ) + + +def downgrade() -> None: + with op.batch_alter_table("energy_cost_period", schema=None) as batch_op: + batch_op.drop_constraint("fk_energy_cost_period_source_binding_id", type_="foreignkey") + batch_op.drop_column("source_binding_id") + + op.drop_index("ix_meter_source_binding_channel_id", table_name="meter_source_binding") + op.drop_index("ix_meter_source_binding_meter_id", table_name="meter_source_binding") + op.drop_table("meter_source_binding") + op.drop_index("ix_meter_source_channel_source_id", table_name="meter_source_channel") + op.drop_table("meter_source_channel") + op.drop_index("ix_meter_source_kind_enabled", table_name="meter_source") + op.drop_table("meter_source") diff --git a/app/models/__init__.py b/app/models/__init__.py index 93e8e82..52e13e0 100644 --- a/app/models/__init__.py +++ b/app/models/__init__.py @@ -5,12 +5,16 @@ from app.models.config import AppConfigEntry from app.models.location import Location from app.models.poo import PooRecord from app.models.public_ip import PublicIPHistory, PublicIPState +from app.models.meter_source import MeterSource, MeterSourceBinding, MeterSourceChannel __all__ = [ "AppConfigEntry", "AuthSession", "AuthUser", "Location", + "MeterSource", + "MeterSourceBinding", + "MeterSourceChannel", "PooRecord", "PublicIPHistory", "PublicIPState", diff --git a/app/models/energy.py b/app/models/energy.py index 2813e1d..2dcf195 100644 --- a/app/models/energy.py +++ b/app/models/energy.py @@ -13,12 +13,12 @@ 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.types import JSON from app.db import Base +from app.models.meter_source import MeterSourceBinding def _uuid4_str() -> str: @@ -85,6 +85,10 @@ class Meter(Base): back_populates="meter", cascade="save-update, merge" ) + source_bindings: Mapped[list["MeterSourceBinding"]] = relationship( + back_populates="meter", cascade="save-update, merge" + ) + class DsmrReading(Base): """One down-sampled DSMR telegram stored as a full JSON blob. @@ -292,6 +296,12 @@ class EnergyCostPeriod(Base): ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True ) + # Nullable while M8 adopts historical DSMR rows. Future normal periods + # will point at the binding that supplied both cumulative endpoints. + source_binding_id: Mapped[int | None] = mapped_column( + ForeignKey("meter_source_binding.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) @@ -307,6 +317,10 @@ class EnergyCostPeriod(Base): # Relationship back to the meter epoch. meter: Mapped["Meter | None"] = relationship(back_populates="cost_periods") + source_binding: Mapped["MeterSourceBinding | 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/app/models/meter_source.py b/app/models/meter_source.py new file mode 100644 index 0000000..bd51997 --- /dev/null +++ b/app/models/meter_source.py @@ -0,0 +1,127 @@ +"""Protocol-agnostic source, channel, and meter-binding identity models.""" + +from __future__ import annotations + +import uuid as _uuid +from datetime import datetime +from typing import TYPE_CHECKING + +from sqlalchemy import ( + Boolean, + DateTime, + ForeignKey, + Index, + Integer, + Numeric, + String, + UniqueConstraint, +) +from sqlalchemy.orm import Mapped, mapped_column, relationship +from sqlalchemy.types import JSON + +from app.db import Base + +if TYPE_CHECKING: + from app.models.energy import EnergyCostPeriod, Meter + + +def _uuid4_str() -> str: + return str(_uuid.uuid4()) + + +def half_open_intervals_overlap( + started_at: datetime, + ended_at: datetime | None, + other_started_at: datetime, + other_ended_at: datetime | None, +) -> bool: + """Return whether two ``[started_at, ended_at)`` intervals overlap. + + ``None`` denotes an open-ended interval. Equal boundaries do not overlap, + which lets a source binding hand off at one exact timestamp. + """ + return (other_ended_at is None or started_at < other_ended_at) and ( + ended_at is None or other_started_at < ended_at + ) + + +class MeterSource(Base): + """A configured protocol connection that discovers one or more channels.""" + + __tablename__ = "meter_source" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + uuid: Mapped[str] = mapped_column(String(36), unique=True, nullable=False, default=_uuid4_str) + name: Mapped[str] = mapped_column(String(255), nullable=False) + kind: Mapped[str] = mapped_column(String(64), nullable=False) + enabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=True) + config: Mapped[dict] = mapped_column(JSON, nullable=False, default=dict) + status: Mapped[str] = mapped_column(String(32), nullable=False, default="unknown") + last_seen_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_error: Mapped[str | None] = mapped_column(String(1024), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + channels: Mapped[list["MeterSourceChannel"]] = relationship( + back_populates="source", cascade="save-update, merge" + ) + + +class MeterSourceChannel(Base): + """A stable cumulative measurement identity discovered from a source.""" + + __tablename__ = "meter_source_channel" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + uuid: Mapped[str] = mapped_column(String(36), unique=True, nullable=False, default=_uuid4_str) + source_id: Mapped[int] = mapped_column( + ForeignKey("meter_source.id", ondelete="RESTRICT"), nullable=False, index=True + ) + channel_key: Mapped[str] = mapped_column(String(128), nullable=False) + label: Mapped[str] = mapped_column(String(255), nullable=False) + suggested_commodity: Mapped[str | None] = mapped_column(String(32), nullable=True) + unit: Mapped[str] = mapped_column(String(32), nullable=False) + device_type: Mapped[str | None] = mapped_column(String(64), nullable=True) + fingerprint: Mapped[str | None] = mapped_column(String(64), nullable=True) + latest_value: Mapped[float | None] = mapped_column(Numeric(20, 6), nullable=True) + latest_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + latest_quality: Mapped[str | None] = mapped_column(String(32), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + source: Mapped["MeterSource"] = relationship(back_populates="channels") + bindings: Mapped[list["MeterSourceBinding"]] = relationship( + back_populates="channel", cascade="save-update, merge" + ) + + __table_args__ = ( + UniqueConstraint("source_id", "channel_key", name="uq_meter_source_channel_source_key"), + ) + + +class MeterSourceBinding(Base): + """Connect one source channel to one physical meter for a half-open window.""" + + __tablename__ = "meter_source_binding" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + uuid: Mapped[str] = mapped_column(String(36), unique=True, nullable=False, default=_uuid4_str) + meter_id: Mapped[int] = mapped_column( + ForeignKey("meter.id", ondelete="RESTRICT"), nullable=False, index=True + ) + channel_id: Mapped[int] = mapped_column( + ForeignKey("meter_source_channel.id", ondelete="RESTRICT"), nullable=False, index=True + ) + started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + ended_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + meter: Mapped["Meter"] = relationship(back_populates="source_bindings") + channel: Mapped["MeterSourceChannel"] = relationship(back_populates="bindings") + cost_periods: Mapped[list["EnergyCostPeriod"]] = relationship( + back_populates="source_binding", cascade="save-update, merge", passive_deletes="all" + ) + + +Index("ix_meter_source_kind_enabled", MeterSource.kind, MeterSource.enabled) diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 6102eef..71cc67c 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -283,7 +283,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T01 — 建立 Source / Channel / Binding 基础模型 [structural] -- **Status**: `todo` +- **Status**: `done` - **Depends**: none - **Context**: 先建立协议无关的身份链和时间约束；本卡只改 schema/model，不接运行时或 HTTP。 diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index abc785b..b8afe86 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 = "20260625_14_meter_uuid" +APP_BASELINE_REVISION = "20260822_15_meter_sources" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_energy_models.py b/tests/test_energy_models.py index 9bffaaf..e67f801 100644 --- a/tests/test_energy_models.py +++ b/tests/test_energy_models.py @@ -105,6 +105,9 @@ def test_energy_tables_exist_after_upgrade(energy_db): "energy_contract_version", "tibber_price", "energy_cost_period", + "meter_source", + "meter_source_channel", + "meter_source_binding", } for table in expected_tables: assert table in table_names, f"{table!r} missing after upgrade to head" @@ -920,6 +923,7 @@ def test_energy_cost_period_meter_id_nullable(energy_db): fetched = session.get(EnergyCostPeriod, period_id) assert fetched is not None assert fetched.meter_id is None + assert fetched.source_binding_id is None def test_energy_cost_period_meter_id_fk_enforced(tmp_path: Path): diff --git a/tests/test_meter_sources.py b/tests/test_meter_sources.py new file mode 100644 index 0000000..4b2aac1 --- /dev/null +++ b/tests/test_meter_sources.py @@ -0,0 +1,384 @@ +"""Schema and model tests for the M8 source/channel/binding identity chain.""" + +from __future__ import annotations + +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 UniqueConstraint, create_engine, event as sa_event, inspect, text +from sqlalchemy.orm import Session + +from app.db import Base +from app.models.energy import EnergyCostPeriod, Meter +from app.models.meter_source import ( + MeterSource, + MeterSourceBinding, + MeterSourceChannel, + half_open_intervals_overlap, +) + + +def _alembic_config(database_url: str) -> Config: + config = Config("alembic_app.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +def _engine_with_foreign_keys(database_url: str): + engine = create_engine(database_url, connect_args={"check_same_thread": False}) + + @sa_event.listens_for(engine, "connect") + def _enable_foreign_keys(dbapi_connection, _connection_record): + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + + return engine + + +@pytest.fixture() +def source_db(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'meter_sources.db'}" + command.upgrade(_alembic_config(database_url), "head") + engine = _engine_with_foreign_keys(database_url) + yield engine + engine.dispose() + + +def _add_source_chain(session: Session, now: datetime) -> tuple[Meter, MeterSource, MeterSourceChannel]: + meter = Meter( + label="Heating meter", + commodity="heating", + started_at=now, + ended_at=None, + reason="initial", + note=None, + created_at=now, + ) + source = MeterSource( + name="WarmteLink", + kind="warmtelink_serial", + created_at=now, + updated_at=now, + ) + session.add_all([meter, source]) + session.flush() + channel = MeterSourceChannel( + source_id=source.id, + channel_key="heating-total", + label="District heating total", + suggested_commodity="heating", + unit="GJ", + created_at=now, + updated_at=now, + ) + session.add(channel) + session.flush() + return meter, source, channel + + +def test_populated_revision_14_upgrades_to_meter_source_head_with_audit(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'revision_14.db'}" + config = _alembic_config(database_url) + command.upgrade(config, "20260625_14_meter_uuid") + + historical_at = datetime(2026, 8, 1, tzinfo=timezone.utc) + engine = _engine_with_foreign_keys(database_url) + try: + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO dsmr_reading (recorded_at, source_id, payload) " + "VALUES (:recorded_at, :source_id, :payload)" + ), + { + "recorded_at": historical_at, + "source_id": 17, + "payload": '{"electricity_delivered_1": "100.000"}', + }, + ) + connection.execute( + text( + "INSERT INTO meter " + "(uuid, label, commodity, started_at, ended_at, reason, note, created_at) " + "VALUES (:uuid, :label, :commodity, :started_at, NULL, :reason, NULL, :created_at)" + ), + { + "uuid": "11111111-1111-4111-8111-111111111111", + "label": "Historical meter", + "commodity": "electricity", + "started_at": historical_at, + "reason": "initial", + "created_at": historical_at, + }, + ) + meter_id = connection.execute(text("SELECT id FROM meter")).scalar_one() + 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 (:period_start, 1, 2, 0, 0, 0.5, 0, 0.5, 'EUR', '{}', NULL, " + ":meter_id, 0, :computed_at)" + ), + { + "period_start": historical_at, + "meter_id": meter_id, + "computed_at": historical_at, + }, + ) + before_counts = { + table_name: engine.connect().execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar_one() + for table_name in ("dsmr_reading", "meter", "energy_cost_period") + } + finally: + engine.dispose() + + command.upgrade(config, "head") + + engine = _engine_with_foreign_keys(database_url) + try: + inspector = inspect(engine) + assert {"meter_source", "meter_source_channel", "meter_source_binding"} <= set( + inspector.get_table_names() + ) + cost_columns = {column["name"] for column in inspector.get_columns("energy_cost_period")} + assert "source_binding_id" in cost_columns + after_counts = { + table_name: engine.connect().execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar_one() + for table_name in ("dsmr_reading", "meter", "energy_cost_period") + } + assert after_counts == before_counts + assert engine.connect().execute(text("SELECT version_num FROM alembic_version")).scalar_one() == ( + "20260822_15_meter_sources" + ) + assert engine.connect().execute(text("PRAGMA foreign_key_check")).all() == [] + + channel_constraints = inspector.get_unique_constraints("meter_source_channel") + assert {tuple(item["column_names"]) for item in channel_constraints} >= { + ("uuid",), + ("source_id", "channel_key"), + } + for table_name in ("meter_source", "meter_source_channel", "meter_source_binding"): + assert ("uuid",) in { + tuple(item["column_names"]) + for item in inspector.get_unique_constraints(table_name) + } + assert { + "ix_meter_source_kind_enabled", + "ix_meter_source_channel_source_id", + "ix_meter_source_binding_meter_id", + "ix_meter_source_binding_channel_id", + } <= { + index["name"] + for table_name in ("meter_source", "meter_source_channel", "meter_source_binding") + for index in inspector.get_indexes(table_name) + } + cost_fks = { + foreign_key["constrained_columns"][0]: foreign_key + for foreign_key in inspector.get_foreign_keys("energy_cost_period") + } + assert cost_fks["meter_id"]["referred_table"] == "meter" + 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") + assert { + table_name: engine.connect().execute(text(f"SELECT COUNT(*) FROM {table_name}")).scalar_one() + for table_name in before_counts + } == before_counts + finally: + engine.dispose() + + +def test_empty_database_upgrade_and_repeat_are_idempotent(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'empty_then_repeat.db'}" + config = _alembic_config(database_url) + + command.upgrade(config, "head") + command.upgrade(config, "head") + + engine = _engine_with_foreign_keys(database_url) + try: + inspector = inspect(engine) + source_columns = {column["name"]: column for column in inspector.get_columns("meter_source")} + channel_unique = { + tuple(constraint["column_names"]) + for constraint in inspector.get_unique_constraints("meter_source_channel") + } + binding_fks = { + tuple(foreign_key["constrained_columns"]): foreign_key + for foreign_key in inspector.get_foreign_keys("meter_source_binding") + } + cost_columns = { + column["name"]: column for column in inspector.get_columns("energy_cost_period") + } + cost_fks = { + tuple(foreign_key["constrained_columns"]): foreign_key + for foreign_key in inspector.get_foreign_keys("energy_cost_period") + } + + assert source_columns["uuid"]["nullable"] is False + assert source_columns["config"]["nullable"] is False + assert ("source_id", "channel_key") in channel_unique + assert binding_fks[("meter_id",)]["options"]["ondelete"] == "RESTRICT" + assert binding_fks[("channel_id",)]["options"]["ondelete"] == "RESTRICT" + assert cost_columns["source_binding_id"]["nullable"] is True + assert cost_fks[("source_binding_id",)]["options"]["ondelete"] == "RESTRICT" + finally: + engine.dispose() + + +def test_meter_source_migration_downgrade_is_schema_reversible(tmp_path: Path): + database_url = f"sqlite:///{tmp_path / 'meter_source_downgrade.db'}" + config = _alembic_config(database_url) + command.upgrade(config, "head") + command.downgrade(config, "20260625_14_meter_uuid") + + engine = _engine_with_foreign_keys(database_url) + try: + inspector = inspect(engine) + tables = set(inspector.get_table_names()) + assert not {"meter_source", "meter_source_channel", "meter_source_binding"} & tables + assert "source_binding_id" not in { + column["name"] for column in inspector.get_columns("energy_cost_period") + } + finally: + engine.dispose() + + +def test_source_model_defaults_and_channel_unique_constraint(source_db): + now = datetime.now(timezone.utc) + with Session(source_db) as session: + _, source, channel = _add_source_chain(session, now) + session.commit() + assert source.uuid + assert source.enabled is True + assert source.config == {} + assert source.status == "unknown" + assert channel.uuid + + with pytest.raises(sqlalchemy.exc.IntegrityError): + with Session(source_db) as session: + duplicate = MeterSourceChannel( + source_id=source.id, + channel_key="heating-total", + label="Duplicate", + unit="GJ", + created_at=now, + updated_at=now, + ) + session.add(duplicate) + session.commit() + + +def test_foreign_keys_restrict_history_and_no_relationship_delete_cascade(source_db): + now = datetime.now(timezone.utc) + with Session(source_db) as session: + meter, source, channel = _add_source_chain(session, now) + binding = MeterSourceBinding( + meter_id=meter.id, + channel_id=channel.id, + started_at=now, + ended_at=None, + created_at=now, + updated_at=now, + ) + session.add(binding) + session.flush() + 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=meter.id, + source_binding_id=binding.id, + degraded=False, + computed_at=now, + ) + session.add(period) + session.commit() + source_id, channel_id, binding_id = source.id, channel.id, binding.id + + for table_name, row_id in ( + ("meter_source", source_id), + ("meter_source_channel", channel_id), + ("meter_source_binding", binding_id), + ): + with pytest.raises(sqlalchemy.exc.IntegrityError): + with Session(source_db) as session: + session.execute(text(f"DELETE FROM {table_name} WHERE id = :row_id"), {"row_id": row_id}) + session.commit() + + with pytest.raises(sqlalchemy.exc.IntegrityError): + with Session(source_db) as session: + binding = session.get(MeterSourceBinding, binding_id) + assert binding is not None + assert len(binding.cost_periods) == 1 + session.delete(binding) + session.commit() + + with Session(source_db) as session: + binding = session.get(MeterSourceBinding, binding_id) + period = session.execute( + text("SELECT source_binding_id FROM energy_cost_period WHERE id = :period_id"), + {"period_id": 1}, + ).scalar_one() + assert binding is not None + assert period == binding_id + + assert "delete" not in MeterSource.channels.property.cascade + assert "delete" not in MeterSourceChannel.bindings.property.cascade + assert "delete" not in MeterSourceBinding.cost_periods.property.cascade + assert "delete" not in Meter.source_bindings.property.cascade + + +def test_channel_unique_constraint_metadata_matches_migrated_schema(source_db): + metadata_table = Base.metadata.tables["meter_source_channel"] + metadata_unique_constraints = { + tuple(column.name for column in constraint.columns) + for constraint in metadata_table.constraints + if isinstance(constraint, UniqueConstraint) + } + inspector = inspect(source_db) + schema_unique_constraints = { + tuple(constraint["column_names"]) + for constraint in inspector.get_unique_constraints("meter_source_channel") + } + schema_unique_indexes = { + tuple(index["column_names"]) + for index in inspector.get_indexes("meter_source_channel") + if index["unique"] + } + + assert ("source_id", "channel_key") in metadata_unique_constraints + assert ("source_id", "channel_key") in schema_unique_constraints + assert ("source_id", "channel_key") not in schema_unique_indexes + + +def test_half_open_binding_interval_helper(): + start = datetime(2026, 1, 1, tzinfo=timezone.utc) + boundary = start + timedelta(hours=1) + later = boundary + timedelta(hours=1) + + assert half_open_intervals_overlap(start, boundary, boundary, later) is False + assert half_open_intervals_overlap(start, later, boundary, None) is True + assert half_open_intervals_overlap(start, None, boundary, later) is True + + +def test_model_tables_and_foreign_keys_are_registered(): + assert {"meter_source", "meter_source_channel", "meter_source_binding"} <= set(Base.metadata.tables) + binding_fks = Base.metadata.tables["meter_source_binding"].foreign_keys + assert {foreign_key.ondelete for foreign_key in binding_fks} == {"RESTRICT"}