From 18fe18281be5544b805d65f85738287e25b2dc30 Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Tue, 23 Jun 2026 20:33:35 +0200 Subject: [PATCH] M6-T01: add 5 energy tables, models, and migration MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - app/models/energy.py: DsmrReading, EnergyContract, EnergyContractVersion, TibberPrice, EnergyCostPeriod (per design §3.5; JSON blobs, RESTRICT FKs). - alembic_app migration 20260623_11_energy_tables (head), env.py imports. - app_db_adopt APP_BASELINE_REVISION bumped to new head. - tests/test_energy_models.py + expose_catalog test fixups for new head. --- alembic_app/env.py | 7 + .../versions/20260623_11_energy_tables.py | 132 ++++ app/models/energy.py | 234 ++++++ docs/design/m6-tibber-dynamic-energy.md | 2 +- scripts/app_db_adopt.py | 2 +- tests/test_energy_models.py | 706 ++++++++++++++++++ tests/test_expose_catalog.py | 21 +- 7 files changed, 1093 insertions(+), 11 deletions(-) create mode 100644 alembic_app/versions/20260623_11_energy_tables.py create mode 100644 app/models/energy.py create mode 100644 tests/test_energy_models.py diff --git a/alembic_app/env.py b/alembic_app/env.py index efa98a9..8ba084c 100644 --- a/alembic_app/env.py +++ b/alembic_app/env.py @@ -13,6 +13,13 @@ from app.models.location import Location # noqa: F401 from app.models.poo import PooRecord # noqa: F401 from app.models.modbus import ModbusDevice, ModbusReading # noqa: F401 from app.models.expose import ExposedEntityToggle # noqa: F401 +from app.models.energy import ( # noqa: F401 + DsmrReading, + EnergyContract, + EnergyContractVersion, + TibberPrice, + EnergyCostPeriod, +) config = context.config diff --git a/alembic_app/versions/20260623_11_energy_tables.py b/alembic_app/versions/20260623_11_energy_tables.py new file mode 100644 index 0000000..6eeb1db --- /dev/null +++ b/alembic_app/versions/20260623_11_energy_tables.py @@ -0,0 +1,132 @@ +"""add energy pricing and DSMR metering tables + +Revision ID: 20260623_11_energy_tables +Revises: 20260622_10_exposed_entities +Create Date: 2026-06-23 00:00:00.000000 +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "20260623_11_energy_tables" +down_revision: Union[str, None] = "20260622_10_exposed_entities" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + # dsmr_reading — raw DSMR telegram blobs (10-second down-sampled). + # No device table: single P1 smart meter; a ``source`` column can be added later + # if a second meter is introduced. + op.create_table( + "dsmr_reading", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("recorded_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("source_id", sa.Integer(), nullable=True), + sa.Column("payload", sa.JSON(), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("source_id", name="uq_dsmr_reading_source_id"), + ) + op.create_index( + "ix_dsmr_reading_recorded_at", + "dsmr_reading", + ["recorded_at"], + unique=False, + ) + + # energy_contract — contract head (manual or tibber, one active at a time). + op.create_table( + "energy_contract", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("name", sa.String(length=255), nullable=False), + sa.Column("kind", sa.String(length=32), nullable=False), + sa.Column("active", sa.Boolean(), nullable=False), + sa.Column("currency", sa.String(length=8), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + + # energy_contract_version — versioned pricing values; append-only for auditability. + # Must be created after energy_contract because of the FK dependency. + op.create_table( + "energy_contract_version", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("contract_id", sa.Integer(), nullable=False), + sa.Column("effective_from", sa.DateTime(timezone=True), nullable=False), + sa.Column("effective_to", sa.DateTime(timezone=True), nullable=True), + sa.Column("values", sa.JSON(), nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["contract_id"], + ["energy_contract.id"], + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id"), + ) + + # tibber_price — cached Tibber 15-minute spot prices (immutable once fetched). + op.create_table( + "tibber_price", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("starts_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("resolution", sa.String(length=32), nullable=False), + sa.Column("energy", sa.Float(), nullable=False), + sa.Column("tax", sa.Float(), nullable=False), + sa.Column("total", sa.Float(), nullable=False), + sa.Column("level", sa.String(length=32), nullable=True), + sa.Column("currency", sa.String(length=8), nullable=False), + sa.Column("fetched_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("starts_at", name="uq_tibber_price_starts_at"), + ) + + # energy_cost_period — computed 15-minute billing periods (immutable snapshot). + # Must be created after energy_contract_version because of the FK dependency. + op.create_table( + "energy_cost_period", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("period_start", sa.DateTime(timezone=True), nullable=False), + sa.Column("d1_kwh", sa.Float(), nullable=False), + sa.Column("d2_kwh", sa.Float(), nullable=False), + sa.Column("r1_kwh", sa.Float(), nullable=False), + sa.Column("r2_kwh", sa.Float(), nullable=False), + sa.Column("import_cost", sa.Float(), nullable=False), + sa.Column("export_revenue", sa.Float(), nullable=False), + sa.Column("net_cost", sa.Float(), nullable=False), + sa.Column("currency", sa.String(length=8), nullable=False), + sa.Column("pricing", sa.JSON(), nullable=False), + sa.Column("contract_version_id", sa.Integer(), nullable=True), + sa.Column("degraded", sa.Boolean(), nullable=False), + sa.Column("computed_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint( + ["contract_version_id"], + ["energy_contract_version.id"], + ondelete="RESTRICT", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("period_start", name="uq_energy_cost_period_period_start"), + ) + + +def downgrade() -> None: + # Drop tables in reverse dependency order: tables with FKs first. + + # energy_cost_period has a FK to energy_contract_version — drop it first. + op.drop_table("energy_cost_period") + + # tibber_price has no FK dependencies on our new tables. + op.drop_table("tibber_price") + + # energy_contract_version has a FK to energy_contract — drop it before energy_contract. + op.drop_table("energy_contract_version") + + # energy_contract has no FK dependencies on our new tables. + op.drop_table("energy_contract") + + # dsmr_reading has no FK dependencies. + op.drop_index("ix_dsmr_reading_recorded_at", table_name="dsmr_reading") + op.drop_table("dsmr_reading") diff --git a/app/models/energy.py b/app/models/energy.py new file mode 100644 index 0000000..55462c4 --- /dev/null +++ b/app/models/energy.py @@ -0,0 +1,234 @@ +"""SQLAlchemy models for the energy pricing and DSMR metering subsystem. + +Five tables: + - 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. + - tibber_price: cached Tibber 15-minute spot prices (immutable). + - energy_cost_period: computed 15-minute billing periods (immutable snapshot). +""" + +from __future__ import annotations + +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 + + +class DsmrReading(Base): + """One down-sampled DSMR telegram stored as a full JSON blob. + + ``recorded_at`` is a real indexed column (not inside the payload) so that + time-range queries are efficient. ``source_id`` holds the telegram's own + integer ``id`` and is used for idempotent upsert (skip if already present). + The entire telegram frame is stored verbatim in ``payload``; no field + allow-list is applied so future commodities (gas, heating, three-phase) + are accommodated without a schema change. + """ + + __tablename__ = "dsmr_reading" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # UTC timestamp of the sample — real column so it can be indexed efficiently. + recorded_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, index=True + ) + + # Telegram's own auto-increment id (DSMR Reader assigns it). Used for + # idempotent ingestion: if the same telegram arrives twice we skip it. + # Nullable because some DSMR sources may not emit an id. + source_id: Mapped[int | None] = mapped_column(Integer, unique=True, nullable=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) + + +class EnergyContract(Base): + """Contract head: a named energy contract with a chosen pricing strategy. + + ``kind`` determines which price strategy is used (``manual`` for fixed + dual-tariff rates entered by the user, ``tibber`` for dynamic API prices). + Only one contract may be ``active`` at a time; the service layer enforces + mutual exclusion. Specific pricing values live in ``EnergyContractVersion`` + so that price changes can be tracked without modifying historical records. + """ + + __tablename__ = "energy_contract" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # Human-readable label; freely editable by the user. + name: Mapped[str] = mapped_column(String(255), nullable=False) + + # Strategy selector: "manual" or "tibber". Application-layer validation + # enforces the allowed set; no DB CHECK constraint is added to keep the + # migration simple and the strategy registry extensible. + kind: Mapped[str] = mapped_column(String(32), nullable=False) + + # Whether this is the currently active contract (at most one should be True). + active: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + + # ISO 4217 currency code for all monetary values in this contract. + currency: Mapped[str] = mapped_column(String(8), nullable=False, default="EUR") + + # Audit timestamps. + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + # Relationship to versions (back-reference; not loaded eagerly). + versions: Mapped[list["EnergyContractVersion"]] = relationship( + back_populates="contract", cascade="save-update, merge" + ) + + +class EnergyContractVersion(Base): + """One time-bounded version of an energy contract's pricing values. + + Pricing changes are modelled as new versions (append-only); existing versions + are never modified so that historical ``EnergyCostPeriod`` records remain + fully auditable. ``effective_to`` is ``NULL`` for the currently open version. + """ + + __tablename__ = "energy_contract_version" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # FK to the parent contract. RESTRICT prevents deletion of a contract that + # still has versioned pricing rows attached to it. + contract_id: Mapped[int] = mapped_column( + ForeignKey("energy_contract.id", ondelete="RESTRICT"), nullable=False + ) + + # Start of this version's validity window (inclusive, UTC). + effective_from: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False + ) + + # End of this version's validity window (exclusive, UTC). NULL means open-ended + # (i.e. this is the most recent / current version). + effective_to: Mapped[datetime | None] = mapped_column( + DateTime(timezone=True), nullable=True + ) + + # Pricing values as a JSON object conforming to the profile structure for + # ``contract.kind`` (validated by the application layer against the YAML profile). + values: Mapped[dict] = mapped_column(JSON, nullable=False) + + # Creation timestamp. + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + # Relationship back to the parent contract. + contract: Mapped["EnergyContract"] = relationship(back_populates="versions") + + # Relationship to cost periods that reference this version. + cost_periods: Mapped[list["EnergyCostPeriod"]] = relationship( + back_populates="contract_version", cascade="save-update, merge" + ) + + +class TibberPrice(Base): + """Cached Tibber 15-minute spot price point (immutable once fetched). + + ``starts_at`` is unique so that upserts are idempotent. Past prices are + never overwritten; the fetch job only adds rows for future time slots. + """ + + __tablename__ = "tibber_price" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # UTC start of the 15-minute slot; unique so upsert is idempotent. + starts_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, unique=True + ) + + # Resolution label as returned by the Tibber API (e.g. "QUARTER_HOURLY"). + resolution: Mapped[str] = mapped_column(String(32), nullable=False) + + # Price components in the contract currency (all include VAT, user-facing). + energy: Mapped[float] = mapped_column(Float, nullable=False) + tax: Mapped[float] = mapped_column(Float, nullable=False) + total: Mapped[float] = mapped_column(Float, nullable=False) + + # Tibber price level (e.g. "NORMAL", "CHEAP", "EXPENSIVE"); may be absent. + level: Mapped[str | None] = mapped_column(String(32), nullable=True) + + # ISO 4217 currency code as returned by the API. + currency: Mapped[str] = mapped_column(String(8), nullable=False) + + # UTC timestamp of when this row was fetched/inserted. + fetched_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + +class EnergyCostPeriod(Base): + """Computed billing record for one 15-minute metering period (immutable snapshot). + + Each row captures the per-register kWh deltas, the resulting import cost and + export revenue, and a full snapshot of the pricing values used so that the + calculation is fully auditable and reproducible without re-querying the + contract version. Rows are written once and never modified; explicit + recomputation via the API is the only way to overwrite a period. + """ + + __tablename__ = "energy_cost_period" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + + # UTC start of the 15-minute period; unique so upsert is idempotent. + period_start: Mapped[datetime] = mapped_column( + DateTime(timezone=True), nullable=False, unique=True + ) + + # Per-register kWh deltas for the period (end minus start of cumulative registers). + # _1 = dal/low-tariff, _2 = normal/high-tariff (NL convention). + d1_kwh: Mapped[float] = mapped_column(Float, nullable=False) # delivered low + d2_kwh: Mapped[float] = mapped_column(Float, nullable=False) # delivered high + r1_kwh: Mapped[float] = mapped_column(Float, nullable=False) # returned low + r2_kwh: Mapped[float] = mapped_column(Float, nullable=False) # returned high + + # Computed monetary amounts for the period (in ``currency``). + import_cost: Mapped[float] = mapped_column(Float, nullable=False) + export_revenue: Mapped[float] = mapped_column(Float, nullable=False) + net_cost: Mapped[float] = mapped_column(Float, nullable=False) + + # ISO 4217 currency code matching the contract. + currency: Mapped[str] = mapped_column(String(8), nullable=False) + + # Full snapshot of the pricing inputs used during computation. This makes + # each row self-contained and auditable even if the contract is later changed. + pricing: Mapped[dict] = mapped_column(JSON, nullable=False) + + # FK to the exact contract version whose values were used. RESTRICT prevents + # deletion of a version that has cost records attached. Nullable to support + # periods computed in ``degraded`` mode (missing price data). + contract_version_id: Mapped[int | None] = mapped_column( + ForeignKey("energy_contract_version.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) + + # UTC timestamp of when this row was computed/inserted. + computed_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + # Relationship back to the contract version. + contract_version: Mapped["EnergyContractVersion | 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; +# no composite index is needed for single-meter deployments.) + +# Index on period_start is covered by the unique constraint (SQLite creates an +# implicit index for UNIQUE columns), so no additional index is required. + +# Index on starts_at for TibberPrice is covered by the unique constraint similarly. diff --git a/docs/design/m6-tibber-dynamic-energy.md b/docs/design/m6-tibber-dynamic-energy.md index d65a54b..44c94e2 100644 --- a/docs/design/m6-tibber-dynamic-energy.md +++ b/docs/design/m6-tibber-dynamic-energy.md @@ -289,7 +289,7 @@ Phase D（API + 前端） > 后端任务沿用校验闸门（`pytest` / `ruff` / 改路由或 schema 则 `export_openapi` 重导出入库）；前端闸门见 §8。**不新增依赖**。纯新增、不删/移文件；改 `app/main.py` lifespan 时不得破坏既有 job/连接。 ### M6-T01 — 5 张表与模型 `[schema]` -- **Status**: `todo` · **Depends**: none +- **Status**: `done` · **Depends**: none - **Context**: 单库 app 链新增 5 张表（§3.5）。只建 schema + 模型，不写采集/计算/接口。 - **Files**: `create app/models/energy.py`（`DsmrReading`/`EnergyContract`/`EnergyContractVersion`/`TibberPrice`/`EnergyCostPeriod`）；`create alembic_app/versions/20260623_11_energy_tables.py`；`modify alembic_app/env.py`、`scripts/app_db_adopt.py`（`APP_BASELINE_REVISION`→新 head）；`create tests/test_energy_models.py` - **Steps**: diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index 1155e29..68b9f6f 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 = "20260622_10_exposed_entities" +APP_BASELINE_REVISION = "20260623_11_energy_tables" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_energy_models.py b/tests/test_energy_models.py new file mode 100644 index 0000000..093faa8 --- /dev/null +++ b/tests/test_energy_models.py @@ -0,0 +1,706 @@ +"""Tests for M6-T01: energy pricing and DSMR metering tables and ORM models. + +Covers: + 1. Migration shape: upgrade to head creates all five 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; + 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. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from pathlib import Path + +import pytest +import sqlalchemy.exc +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, event as sa_event, inspect, text +from sqlalchemy.orm import Session + +from app.db import Base +from app.models.energy import ( + DsmrReading, + EnergyContract, + EnergyContractVersion, + TibberPrice, + EnergyCostPeriod, +) +from scripts.app_db_adopt import APP_BASELINE_REVISION + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_app_alembic_config(database_url: str) -> Config: + cfg = Config("alembic_app.ini") + cfg.set_main_option("sqlalchemy.url", database_url) + return cfg + + +def _engine_with_fk(db_url: str): + """Create a SQLAlchemy engine with SQLite FK enforcement enabled.""" + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + + @sa_event.listens_for(engine, "connect") + def _enable_fk(dbapi_conn, _rec): + cursor = dbapi_conn.cursor() + cursor.execute("PRAGMA foreign_keys = ON") + cursor.close() + + return engine + + +# --------------------------------------------------------------------------- +# Fixtures +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def energy_db(tmp_path: Path): + """Temporary SQLite DB upgraded to the current Alembic head.""" + db_path = tmp_path / "energy_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + yield engine + engine.dispose() + + +@pytest.fixture() +def energy_db_with_fk(tmp_path: Path): + """Temporary SQLite DB upgraded to head with FK enforcement enabled.""" + db_path = tmp_path / "energy_fk_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = _engine_with_fk(db_url) + yield engine + engine.dispose() + + +# --------------------------------------------------------------------------- +# 1. Migration shape tests +# --------------------------------------------------------------------------- + + +def test_energy_tables_exist_after_upgrade(energy_db): + """All five energy tables must exist after upgrade to head.""" + inspector = inspect(energy_db) + table_names = inspector.get_table_names() + expected_tables = { + "dsmr_reading", + "energy_contract", + "energy_contract_version", + "tibber_price", + "energy_cost_period", + } + for table in expected_tables: + assert table in table_names, f"{table!r} missing after upgrade to head" + + +def test_dsmr_reading_columns(energy_db): + """dsmr_reading must have id, recorded_at (NOT NULL), source_id (nullable), payload (NOT NULL).""" + inspector = inspect(energy_db) + columns = {col["name"]: col for col in inspector.get_columns("dsmr_reading")} + + assert "id" in columns and not columns["id"]["nullable"] + assert "recorded_at" in columns and not columns["recorded_at"]["nullable"] + assert "source_id" in columns and columns["source_id"]["nullable"] + assert "payload" in columns and not columns["payload"]["nullable"] + + +def test_dsmr_reading_source_id_unique(energy_db): + """dsmr_reading.source_id must have a unique constraint.""" + 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" in unique_cols, "source_id must have a unique constraint" + + +def test_dsmr_reading_recorded_at_index(energy_db): + """dsmr_reading.recorded_at must have an index.""" + inspector = inspect(energy_db) + indexes = {idx["name"]: idx for idx in inspector.get_indexes("dsmr_reading")} + assert "ix_dsmr_reading_recorded_at" in indexes, ( + "ix_dsmr_reading_recorded_at missing from dsmr_reading" + ) + + +def test_energy_contract_columns(energy_db): + """energy_contract must have all required columns.""" + inspector = inspect(energy_db) + columns = {col["name"]: col for col in inspector.get_columns("energy_contract")} + + required_non_nullable = {"id", "name", "kind", "active", "currency", "created_at", "updated_at"} + for col_name in required_non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" + + +def test_energy_contract_version_columns(energy_db): + """energy_contract_version must have all required columns with correct nullability.""" + inspector = inspect(energy_db) + columns = {col["name"]: col for col in inspector.get_columns("energy_contract_version")} + + non_nullable = {"id", "contract_id", "effective_from", "values", "created_at"} + nullable = {"effective_to"} + + for col_name in non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" + + for col_name in nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert columns[col_name]["nullable"], f"{col_name} should be nullable" + + +def test_tibber_price_columns(energy_db): + """tibber_price must have all required columns.""" + inspector = inspect(energy_db) + columns = {col["name"]: col for col in inspector.get_columns("tibber_price")} + + non_nullable = { + "id", "starts_at", "resolution", "energy", "tax", "total", + "currency", "fetched_at", + } + nullable = {"level"} + + for col_name in non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" + + for col_name in nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert columns[col_name]["nullable"], f"{col_name} should be nullable" + + +def test_tibber_price_starts_at_unique(energy_db): + """tibber_price.starts_at must have a unique constraint.""" + inspector = inspect(energy_db) + unique_constraints = inspector.get_unique_constraints("tibber_price") + unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] + assert "starts_at" in unique_cols, "starts_at must have a unique constraint" + + +def test_energy_cost_period_columns(energy_db): + """energy_cost_period must have all required columns with correct nullability.""" + inspector = inspect(energy_db) + columns = {col["name"]: col for col in inspector.get_columns("energy_cost_period")} + + non_nullable = { + "id", "period_start", "d1_kwh", "d2_kwh", "r1_kwh", "r2_kwh", + "import_cost", "export_revenue", "net_cost", "currency", + "pricing", "degraded", "computed_at", + } + nullable = {"contract_version_id"} + + for col_name in non_nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert not columns[col_name]["nullable"], f"{col_name} should be NOT NULL" + + for col_name in nullable: + assert col_name in columns, f"Missing column: {col_name}" + assert columns[col_name]["nullable"], f"{col_name} should be nullable" + + +def test_energy_cost_period_period_start_unique(energy_db): + """energy_cost_period.period_start must have a unique constraint.""" + inspector = inspect(energy_db) + unique_constraints = inspector.get_unique_constraints("energy_cost_period") + unique_cols = [col for uc in unique_constraints for col in uc["column_names"]] + assert "period_start" in unique_cols, "period_start must have a unique constraint" + + +def test_energy_contract_version_fk_to_contract(energy_db): + """energy_contract_version.contract_id must have a FK referencing energy_contract.id.""" + inspector = inspect(energy_db) + fks = inspector.get_foreign_keys("energy_contract_version") + assert len(fks) == 1, f"Expected 1 FK on energy_contract_version, got {len(fks)}" + fk = fks[0] + assert fk["referred_table"] == "energy_contract" + assert "contract_id" in fk["constrained_columns"] + assert "id" in fk["referred_columns"] + + +def test_energy_cost_period_fk_to_contract_version(energy_db): + """energy_cost_period.contract_version_id must have a FK referencing energy_contract_version.id.""" + inspector = inspect(energy_db) + fks = inspector.get_foreign_keys("energy_cost_period") + assert len(fks) == 1, f"Expected 1 FK on energy_cost_period, got {len(fks)}" + fk = fks[0] + assert fk["referred_table"] == "energy_contract_version" + assert "contract_version_id" in fk["constrained_columns"] + assert "id" in fk["referred_columns"] + + +def test_downgrade_removes_energy_tables(tmp_path: Path): + """Downgrading to 20260622_10_exposed_entities must cleanly remove all five energy tables.""" + db_path = tmp_path / "downgrade_energy_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + + command.upgrade(alembic_cfg, "head") + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + inspector = inspect(engine) + for table in ("dsmr_reading", "energy_contract", "energy_contract_version", + "tibber_price", "energy_cost_period"): + assert table in inspector.get_table_names(), f"{table} should exist after upgrade" + engine.dispose() + + # Downgrade to the revision just before ours — the explicit down_revision. + command.downgrade(alembic_cfg, "20260622_10_exposed_entities") + + engine = create_engine(db_url, connect_args={"check_same_thread": False}) + inspector = inspect(engine) + table_names = inspector.get_table_names() + for table in ("dsmr_reading", "energy_contract", "energy_contract_version", + "tibber_price", "energy_cost_period"): + assert table not in table_names, f"{table} should be gone after downgrade" + engine.dispose() + + +# --------------------------------------------------------------------------- +# 2. ORM metadata checks (Base.metadata) +# --------------------------------------------------------------------------- + + +def test_base_metadata_contains_energy_tables(): + """Base.metadata.tables must include all five new energy tables.""" + expected = { + "dsmr_reading", + "energy_contract", + "energy_contract_version", + "tibber_price", + "energy_cost_period", + } + for table in expected: + assert table in Base.metadata.tables, f"{table!r} not in Base.metadata" + + +def test_dsmr_reading_payload_is_json(): + """DsmrReading.payload must be mapped as a JSON column in ORM metadata.""" + from sqlalchemy.types import JSON as JSON_TYPE + table = Base.metadata.tables["dsmr_reading"] + col = table.columns["payload"] + assert isinstance(col.type, JSON_TYPE), "payload column must be JSON type" + + +def test_energy_contract_version_values_is_json(): + """EnergyContractVersion.values must be mapped as a JSON column in ORM metadata.""" + from sqlalchemy.types import JSON as JSON_TYPE + table = Base.metadata.tables["energy_contract_version"] + col = table.columns["values"] + assert isinstance(col.type, JSON_TYPE), "values column must be JSON type" + + +def test_energy_cost_period_pricing_is_json(): + """EnergyCostPeriod.pricing must be mapped as a JSON column in ORM metadata.""" + from sqlalchemy.types import JSON as JSON_TYPE + table = Base.metadata.tables["energy_cost_period"] + col = table.columns["pricing"] + assert isinstance(col.type, JSON_TYPE), "pricing column must be JSON type" + + +def test_energy_contract_version_fk_ondelete_restrict(): + """The FK from energy_contract_version.contract_id must be ON DELETE RESTRICT.""" + table = Base.metadata.tables["energy_contract_version"] + contract_id_col = table.columns["contract_id"] + assert contract_id_col.foreign_keys, "contract_id must have a foreign key" + fk = next(iter(contract_id_col.foreign_keys)) + assert fk.ondelete == "RESTRICT", ( + f"FK ondelete must be RESTRICT, got: {fk.ondelete!r}" + ) + + +def test_energy_cost_period_fk_ondelete_restrict(): + """The FK from energy_cost_period.contract_version_id must be ON DELETE RESTRICT.""" + table = Base.metadata.tables["energy_cost_period"] + col = table.columns["contract_version_id"] + assert col.foreign_keys, "contract_version_id must have a foreign key" + fk = next(iter(col.foreign_keys)) + assert fk.ondelete == "RESTRICT", ( + f"FK ondelete must be RESTRICT, got: {fk.ondelete!r}" + ) + + +def test_dsmr_reading_source_id_unique_in_metadata(): + """DsmrReading.source_id must be declared unique in ORM metadata.""" + table = Base.metadata.tables["dsmr_reading"] + col = table.columns["source_id"] + assert col.unique, "source_id must be declared unique" + + +def test_tibber_price_starts_at_unique_in_metadata(): + """TibberPrice.starts_at must be declared unique in ORM metadata.""" + table = Base.metadata.tables["tibber_price"] + col = table.columns["starts_at"] + assert col.unique, "starts_at must be declared unique" + + +# --------------------------------------------------------------------------- +# 3. Baseline constant +# --------------------------------------------------------------------------- + + +def test_app_baseline_revision_matches_head(tmp_path: Path): + """APP_BASELINE_REVISION must equal the Alembic head revision.""" + from alembic.script import ScriptDirectory + + db_url = f"sqlite:///{tmp_path / 'rev_check.db'}" + alembic_cfg = _make_app_alembic_config(db_url) + script = ScriptDirectory.from_config(alembic_cfg) + heads = script.get_heads() + assert len(heads) == 1, f"Expected exactly 1 Alembic head, got {heads}" + head = heads[0] + assert APP_BASELINE_REVISION == head, ( + f"APP_BASELINE_REVISION={APP_BASELINE_REVISION!r} does not match " + f"Alembic head={head!r}" + ) + + +# --------------------------------------------------------------------------- +# 4. ORM round-trip tests +# --------------------------------------------------------------------------- + + +def test_dsmr_reading_insert_and_retrieve(energy_db): + """A DsmrReading can be inserted with a JSON payload and retrieved correctly.""" + now = datetime.now(tz=timezone.utc) + sample_payload = { + "timestamp": "2026-06-23T10:00:00+00:00", + "electricity_delivered_1": "20915.154", + "electricity_delivered_2": "18372.099", + "electricity_returned_1": "1234.567", + "electricity_returned_2": "890.123", + "current_electricity_usage": "1.234", + "extra_device_timestamp": None, + } + + with Session(energy_db) as session: + reading = DsmrReading( + recorded_at=now, + source_id=42, + payload=sample_payload, + ) + session.add(reading) + session.commit() + reading_id = reading.id + + with Session(energy_db) as session: + fetched = session.get(DsmrReading, reading_id) + assert fetched is not None + assert fetched.source_id == 42 + assert fetched.payload == sample_payload + assert fetched.payload["electricity_delivered_1"] == "20915.154" + assert fetched.payload["extra_device_timestamp"] is None + + +def test_dsmr_reading_source_id_nullable(energy_db): + """DsmrReading can be inserted without a source_id (nullable).""" + now = datetime.now(tz=timezone.utc) + with Session(energy_db) as session: + reading = DsmrReading( + recorded_at=now, + source_id=None, + payload={"test": True}, + ) + session.add(reading) + session.commit() + reading_id = reading.id + + with Session(energy_db) as session: + fetched = session.get(DsmrReading, reading_id) + assert fetched is not None + assert fetched.source_id is None + + +def test_energy_contract_insert_and_retrieve(energy_db): + """An EnergyContract can be inserted and retrieved with correct defaults.""" + now = datetime.now(tz=timezone.utc) + with Session(energy_db) as session: + contract = EnergyContract( + name="My Manual Contract", + kind="manual", + active=True, + currency="EUR", + created_at=now, + updated_at=now, + ) + session.add(contract) + session.commit() + contract_id = contract.id + + with Session(energy_db) as session: + fetched = session.get(EnergyContract, contract_id) + assert fetched is not None + assert fetched.name == "My Manual Contract" + assert fetched.kind == "manual" + assert fetched.active is True + assert fetched.currency == "EUR" + + +def test_energy_contract_version_insert_and_retrieve(energy_db): + """An EnergyContractVersion can be inserted with a JSON values blob.""" + now = datetime.now(tz=timezone.utc) + pricing_values = { + "energy": { + "buy": {"normal": 0.133, "dal": 0.127}, + "sell": {"normal": 0.05, "dal": 0.05}, + "energy_tax": 0.1108, + "ode": 0.0, + }, + "standing": { + "network_fee": 9.87, + "management_fee": 9.87, + }, + "credits": { + "heffingskorting": 600.0, + }, + } + + with Session(energy_db) as session: + contract = EnergyContract( + name="Fixed Rate 2026", + kind="manual", + active=False, + currency="EUR", + created_at=now, + updated_at=now, + ) + session.add(contract) + session.flush() + + version = EnergyContractVersion( + contract_id=contract.id, + effective_from=now, + effective_to=None, + values=pricing_values, + created_at=now, + ) + session.add(version) + session.commit() + version_id = version.id + + with Session(energy_db) as session: + fetched = session.get(EnergyContractVersion, version_id) + assert fetched is not None + assert fetched.effective_to is None + assert fetched.values == pricing_values + assert fetched.values["energy"]["buy"]["normal"] == 0.133 + + +def test_tibber_price_insert_and_retrieve(energy_db): + """A TibberPrice can be inserted and retrieved correctly.""" + now = datetime.now(tz=timezone.utc) + with Session(energy_db) as session: + price = TibberPrice( + starts_at=now, + resolution="QUARTER_HOURLY", + energy=0.08, + tax=0.05, + total=0.13, + level="NORMAL", + currency="EUR", + fetched_at=now, + ) + session.add(price) + session.commit() + price_id = price.id + + with Session(energy_db) as session: + fetched = session.get(TibberPrice, price_id) + assert fetched is not None + assert fetched.total == 0.13 + assert fetched.level == "NORMAL" + assert fetched.currency == "EUR" + + +def test_tibber_price_level_nullable(energy_db): + """TibberPrice.level can be None.""" + now = datetime.now(tz=timezone.utc) + from datetime import timedelta + other_time = now + timedelta(minutes=15) + + with Session(energy_db) as session: + price = TibberPrice( + starts_at=other_time, + resolution="QUARTER_HOURLY", + energy=0.07, + tax=0.04, + total=0.11, + level=None, + currency="EUR", + fetched_at=now, + ) + session.add(price) + session.commit() + price_id = price.id + + with Session(energy_db) as session: + fetched = session.get(TibberPrice, price_id) + assert fetched is not None + assert fetched.level is None + + +def test_energy_cost_period_insert_and_retrieve(energy_db): + """An EnergyCostPeriod can be inserted with a JSON pricing snapshot.""" + now = datetime.now(tz=timezone.utc) + pricing_snapshot = { + "kind": "manual", + "buy_normal": 0.133, + "buy_dal": 0.127, + "sell_normal": 0.05, + "sell_dal": 0.05, + } + + with Session(energy_db) as session: + period = EnergyCostPeriod( + period_start=now, + d1_kwh=0.5, + d2_kwh=1.2, + r1_kwh=0.0, + r2_kwh=0.0, + import_cost=0.225, + export_revenue=0.0, + net_cost=0.225, + currency="EUR", + pricing=pricing_snapshot, + contract_version_id=None, + degraded=False, + computed_at=now, + ) + session.add(period) + session.commit() + period_id = period.id + + with Session(energy_db) as session: + fetched = session.get(EnergyCostPeriod, period_id) + assert fetched is not None + assert fetched.d1_kwh == 0.5 + assert fetched.d2_kwh == 1.2 + assert fetched.net_cost == 0.225 + assert fetched.degraded is False + assert fetched.contract_version_id is None + assert fetched.pricing == pricing_snapshot + + +# --------------------------------------------------------------------------- +# 5. FK RESTRICT enforcement tests +# --------------------------------------------------------------------------- + + +def test_contract_version_restrict_prevents_contract_deletion( + tmp_path: Path, +): + """Deleting an EnergyContract with versions must fail due to ON DELETE RESTRICT.""" + db_path = tmp_path / "restrict_contract_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = _engine_with_fk(db_url) + + try: + now = datetime.now(tz=timezone.utc) + with Session(engine) as session: + contract = EnergyContract( + name="RESTRICT Test Contract", + kind="manual", + active=False, + currency="EUR", + created_at=now, + updated_at=now, + ) + session.add(contract) + session.flush() + + version = EnergyContractVersion( + contract_id=contract.id, + effective_from=now, + effective_to=None, + values={"test": True}, + created_at=now, + ) + session.add(version) + session.commit() + contract_id = contract.id + + with pytest.raises(sqlalchemy.exc.IntegrityError): + with Session(engine) as session: + session.execute( + text("DELETE FROM energy_contract WHERE id = :cid"), + {"cid": contract_id}, + ) + session.commit() + finally: + engine.dispose() + + +def test_cost_period_restrict_prevents_version_deletion(tmp_path: Path): + """Deleting an EnergyContractVersion with cost periods must fail due to ON DELETE RESTRICT.""" + db_path = tmp_path / "restrict_version_test.db" + db_url = f"sqlite:///{db_path}" + alembic_cfg = _make_app_alembic_config(db_url) + command.upgrade(alembic_cfg, "head") + engine = _engine_with_fk(db_url) + + try: + now = datetime.now(tz=timezone.utc) + with Session(engine) as session: + contract = EnergyContract( + name="RESTRICT Version Test", + kind="manual", + active=False, + currency="EUR", + created_at=now, + updated_at=now, + ) + session.add(contract) + session.flush() + + version = EnergyContractVersion( + contract_id=contract.id, + effective_from=now, + effective_to=None, + values={"test": True}, + created_at=now, + ) + session.add(version) + session.flush() + + period = EnergyCostPeriod( + period_start=now, + d1_kwh=0.1, + d2_kwh=0.2, + r1_kwh=0.0, + r2_kwh=0.0, + import_cost=0.05, + export_revenue=0.0, + net_cost=0.05, + currency="EUR", + pricing={"kind": "manual"}, + contract_version_id=version.id, + degraded=False, + computed_at=now, + ) + session.add(period) + session.commit() + version_id = version.id + + with pytest.raises(sqlalchemy.exc.IntegrityError): + with Session(engine) as session: + session.execute( + text("DELETE FROM energy_contract_version WHERE id = :vid"), + {"vid": version_id}, + ) + session.commit() + finally: + engine.dispose() diff --git a/tests/test_expose_catalog.py b/tests/test_expose_catalog.py index d30ba26..0b886c0 100644 --- a/tests/test_expose_catalog.py +++ b/tests/test_expose_catalog.py @@ -530,7 +530,13 @@ def test_exposed_entity_toggle_key_unique_index(expose_db): def test_downgrade_removes_toggle_table(tmp_path: Path): - """downgrade -1 from head must drop the exposed_entity_toggle table cleanly.""" + """Downgrading to the revision before exposed_entity_toggle must drop that table cleanly. + + We downgrade to the explicit down_revision of the expose-toggle migration + (``20260622_09_modbus_tables``) rather than using ``-1`` from head, so the test + remains correct as new revisions are added on top. This is the same pattern + used in test_modbus_models.py for the modbus downgrade test. + """ db_path = tmp_path / "downgrade_expose_test.db" db_url = f"sqlite:///{db_path}" alembic_cfg = _make_app_alembic_config(db_url) @@ -542,14 +548,15 @@ def test_downgrade_removes_toggle_table(tmp_path: Path): assert "exposed_entity_toggle" in inspector.get_table_names() engine.dispose() - # Downgrade one step removes the exposed_entity_toggle revision. - command.downgrade(alembic_cfg, "-1") + # Downgrade to the explicit down_revision of the exposed_entity_toggle migration, + # consistent with how test_modbus_models.py handles its own downgrade test. + command.downgrade(alembic_cfg, "20260622_09_modbus_tables") engine = create_engine(db_url, connect_args={"check_same_thread": False}) inspector = inspect(engine) table_names = inspector.get_table_names() assert "exposed_entity_toggle" not in table_names, ( - "exposed_entity_toggle should be gone after downgrade -1" + "exposed_entity_toggle should be gone after downgrade to 20260622_09_modbus_tables" ) # Previous tables must still exist. assert "modbus_device" in table_names @@ -668,7 +675,7 @@ def test_expose_toggle_update(expose_db): def test_app_baseline_revision_matches_new_head(tmp_path: Path): - """APP_BASELINE_REVISION must equal the Alembic head after the new migration.""" + """APP_BASELINE_REVISION must equal the Alembic head revision.""" from alembic.script import ScriptDirectory from scripts.app_db_adopt import APP_BASELINE_REVISION @@ -683,7 +690,3 @@ def test_app_baseline_revision_matches_new_head(tmp_path: Path): f"APP_BASELINE_REVISION={APP_BASELINE_REVISION!r} does not match " f"Alembic head={head!r}" ) - # Confirm the new revision is the head. - assert head == "20260622_10_exposed_entities", ( - f"Expected new head to be '20260622_10_exposed_entities', got {head!r}" - )