M6-T01: add 5 energy tables, models, and migration
- 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.
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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")
|
||||
@@ -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.
|
||||
@@ -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**:
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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()
|
||||
@@ -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}"
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user