"""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"}