From 567ddb9779bd6237f6a857528bd514bf4c9b66bb Mon Sep 17 00:00:00 2001 From: Tianyu Liu Date: Sun, 23 Aug 2026 09:45:28 +0200 Subject: [PATCH] M8-T14: add generic meter cost periods --- .../20260822_19_meter_cost_periods.py | 95 ++++ app/models/energy.py | 196 +++++++- docs/design/m8-warmtelink-energy.md | 3 +- scripts/app_db_adopt.py | 2 +- tests/test_energy_models.py | 6 +- tests/test_meter_cost_models.py | 465 ++++++++++++++++++ 6 files changed, 759 insertions(+), 8 deletions(-) create mode 100644 alembic_app/versions/20260822_19_meter_cost_periods.py create mode 100644 tests/test_meter_cost_models.py diff --git a/alembic_app/versions/20260822_19_meter_cost_periods.py b/alembic_app/versions/20260822_19_meter_cost_periods.py new file mode 100644 index 0000000..5f1fb7d --- /dev/null +++ b/alembic_app/versions/20260822_19_meter_cost_periods.py @@ -0,0 +1,95 @@ +"""add generic commodity-scoped meter cost periods + +Revision ID: 20260822_19_meter_cost_periods +Revises: 20260822_18_contract_scopes +Create Date: 2026-08-22 00:00:00.000000 + +This additive migration creates a separate audit ledger for non-electricity +meter costs. It deliberately does not alter, migrate, or delete rows from the +existing electricity-only energy_cost_period table. +""" + +from typing import Sequence, Union + +import sqlalchemy as sa +from alembic import op + + +revision: str = "20260822_19_meter_cost_periods" +down_revision: Union[str, None] = "20260822_18_contract_scopes" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +class ExactDecimal(sa.TypeDecorator): + """Use SQLite text storage while retaining Numeric semantics elsewhere.""" + + impl = sa.Numeric + cache_ok = True + + def __init__(self, precision: int, scale: int) -> None: + self.precision = precision + self.scale = scale + super().__init__(precision=precision, scale=scale) + + def load_dialect_impl(self, dialect): + if dialect.name == "sqlite": + return dialect.type_descriptor(sa.String(self.precision + 2)) + return dialect.type_descriptor(sa.Numeric(self.precision, self.scale, asdecimal=True)) + + +def upgrade() -> None: + op.create_table( + "meter_cost_period", + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("commodity", sa.String(length=32), nullable=False), + sa.Column("period_start", sa.DateTime(timezone=True), nullable=False), + sa.Column("period_end", sa.DateTime(timezone=True), nullable=False), + sa.Column("meter_id", sa.Integer(), nullable=True), + sa.Column("source_binding_id", sa.Integer(), nullable=True), + sa.Column("contract_version_id", sa.Integer(), nullable=True), + # SQLite NUMERIC coercion binds Decimal values as binary floats. Store + # fixed-width decimal text there, while retaining Numeric semantics on + # other supported dialects. + sa.Column("quantity", ExactDecimal(15, 6), nullable=False), + sa.Column("cost", ExactDecimal(15, 9), nullable=False), + sa.Column("currency", sa.String(length=8), nullable=False), + sa.Column("cost_breakdown", sa.JSON(), nullable=False), + sa.Column("pricing_snapshot", sa.JSON(), nullable=False), + sa.Column("quality", sa.String(length=32), nullable=False), + sa.Column("degraded", sa.Boolean(), nullable=False, server_default=sa.false()), + sa.Column("degraded_reason", sa.String(length=255), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.ForeignKeyConstraint(["meter_id"], ["meter.id"], ondelete="RESTRICT"), + sa.ForeignKeyConstraint( + ["source_binding_id"], ["meter_source_binding.id"], ondelete="RESTRICT" + ), + sa.ForeignKeyConstraint( + ["contract_version_id"], ["energy_contract_version.id"], ondelete="RESTRICT" + ), + sa.CheckConstraint( + "degraded OR (meter_id IS NOT NULL AND source_binding_id IS NOT NULL " + "AND contract_version_id IS NOT NULL)", + name="ck_meter_cost_period_normal_audit_links", + ), + sa.CheckConstraint("period_end > period_start", name="ck_meter_cost_period_positive_interval"), + sa.CheckConstraint( + "NOT degraded OR (degraded_reason IS NOT NULL AND length(trim(degraded_reason)) > 0)", + name="ck_meter_cost_period_degraded_reason", + ), + sa.PrimaryKeyConstraint("id"), + sa.UniqueConstraint("commodity", "period_start", name="uq_meter_cost_period_commodity_start"), + ) + op.create_index( + "ix_meter_cost_period_commodity_start", "meter_cost_period", ["commodity", "period_start"] + ) + op.create_index( + "ix_meter_cost_period_source_binding_id", "meter_cost_period", ["source_binding_id"] + ) + + +def downgrade() -> None: + op.drop_index("ix_meter_cost_period_source_binding_id", table_name="meter_cost_period") + op.drop_index("ix_meter_cost_period_commodity_start", table_name="meter_cost_period") + op.drop_table("meter_cost_period") diff --git a/app/models/energy.py b/app/models/energy.py index 97fcc52..22dcc23 100644 --- a/app/models/energy.py +++ b/app/models/energy.py @@ -12,20 +12,25 @@ Six tables: from __future__ import annotations import uuid as _uuid -from datetime import datetime +from datetime import datetime, timezone +from decimal import Decimal +from typing import Any from sqlalchemy import ( Boolean, + CheckConstraint, DateTime, Float, ForeignKey, + Index, Integer, + Numeric, String, UniqueConstraint, event, text, ) -from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym -from sqlalchemy.types import JSON +from sqlalchemy.orm import Mapped, mapped_column, relationship, synonym, validates +from sqlalchemy.types import JSON, TypeDecorator from app.db import Base from app.models.meter_source import MeterSourceBinding @@ -35,6 +40,106 @@ def _uuid4_str() -> str: return str(_uuid.uuid4()) +def _decimal_json(value: Any) -> Any: + """Make auditable JSON portable without admitting binary numeric values.""" + if isinstance(value, Decimal): + return format(value, "f") + if isinstance(value, float) or (isinstance(value, int) and not isinstance(value, bool)): + raise ValueError("JSON amounts and quantities must be decimal strings, not numeric JSON values") + if isinstance(value, dict): + return {key: _decimal_json(child) for key, child in value.items()} + if isinstance(value, list): + return [_decimal_json(child) for child in value] + return value + + +def _validate_fixed_decimal(value: Decimal, precision: int, scale: int, field: str) -> Decimal: + if not isinstance(value, Decimal): + raise ValueError(f"{field} must be a Decimal, not a binary float or other numeric type") + if not value.is_finite(): + raise ValueError(f"{field} must be finite") + if -value.as_tuple().exponent > scale: + raise ValueError(f"{field} exceeds scale {scale}") + integer_digits = 0 if value.is_zero() else max(value.copy_abs().adjusted() + 1, 0) + if integer_digits > precision - scale: + raise ValueError(f"{field} exceeds precision {precision},{scale}") + return value + + +class ExactDecimal(TypeDecorator[Decimal]): + """Fixed-point Decimal which uses SQLite TEXT, never a binary float.""" + + impl = Numeric + cache_ok = True + + def __init__(self, precision: int, scale: int) -> None: + self.precision = precision + self.scale = scale + super().__init__(precision=precision, scale=scale) + + def load_dialect_impl(self, dialect): + if dialect.name == "sqlite": + return dialect.type_descriptor(String(self.precision + 2)) + return dialect.type_descriptor(Numeric(self.precision, self.scale, asdecimal=True)) + + def process_bind_param(self, value: Decimal | None, dialect) -> Decimal | str | None: + if value is None: + return None + value = _validate_fixed_decimal(value, self.precision, self.scale, "decimal value") + if dialect.name == "sqlite": + return format(value, f".{self.scale}f") + return value + + def process_result_value(self, value: Decimal | str | None, _dialect) -> Decimal | None: + return None if value is None else Decimal(value) + + +class DecimalJSON(TypeDecorator[dict]): + """JSON which serializes Decimal values as strings on every write path.""" + + impl = JSON + cache_ok = True + + def process_bind_param(self, value: Any, _dialect) -> Any: + return None if value is None else _decimal_json(value) + + +class UTCDateTime(TypeDecorator[datetime]): + """UTC timestamps that preserve instant identity on SQLite and other dialects.""" + + impl = DateTime(timezone=True) + cache_ok = True + + def __init__(self, field: str) -> None: + self.field = field + super().__init__() + + def process_bind_param(self, value: datetime | None, _dialect) -> datetime | None: + return None if value is None else _normalise_utc_period(value, self.field) + + def process_result_value(self, value: datetime | None, _dialect) -> datetime | None: + if value is None: + return None + if value.tzinfo is None or value.utcoffset() is None: + return value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc) + + +def _require_aware_period(start: datetime, end: datetime) -> None: + if start.tzinfo is None or start.utcoffset() is None: + raise ValueError("period_start must be timezone-aware") + if end.tzinfo is None or end.utcoffset() is None: + raise ValueError("period_end must be timezone-aware") + if end <= start: + raise ValueError("period_end must be after period_start") + + +def _normalise_utc_period(value: datetime, field: str) -> datetime: + if value.tzinfo is None or value.utcoffset() is None: + raise ValueError(f"{field} must be timezone-aware") + return value.astimezone(timezone.utc) + + class Meter(Base): """One physical electricity meter's installation epoch. @@ -358,6 +463,91 @@ class EnergyCostPeriod(Base): ) +class MeterCostPeriod(Base): + """Auditable commodity-scoped ledger row for one half-open metering period. + + A degraded row intentionally permits missing audit links; services must + still require them before writing a normal row. + """ + + __tablename__ = "meter_cost_period" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + commodity: Mapped[str] = mapped_column(String(32), nullable=False) + period_start: Mapped[datetime] = mapped_column(UTCDateTime("period_start"), nullable=False) + period_end: Mapped[datetime] = mapped_column(UTCDateTime("period_end"), nullable=False) + meter_id: Mapped[int | None] = mapped_column( + ForeignKey("meter.id", ondelete="RESTRICT"), nullable=True + ) + source_binding_id: Mapped[int | None] = mapped_column( + ForeignKey("meter_source_binding.id", ondelete="RESTRICT"), nullable=True + ) + contract_version_id: Mapped[int | None] = mapped_column( + ForeignKey("energy_contract_version.id", ondelete="RESTRICT"), nullable=True + ) + + # SQLite reliably round-trips at most fifteen significant decimal digits. + # WarmteLink itself reports 0.001 units, so nine cost fractional digits + # retain a six-place tariff times that source precision without float loss. + quantity: Mapped[Decimal] = mapped_column(ExactDecimal(15, 6), nullable=False) + cost: Mapped[Decimal] = mapped_column(ExactDecimal(15, 9), nullable=False) + currency: Mapped[str] = mapped_column(String(8), nullable=False) + cost_breakdown: Mapped[dict] = mapped_column(DecimalJSON(), nullable=False, default=dict) + pricing_snapshot: Mapped[dict] = mapped_column(DecimalJSON(), nullable=False, default=dict) + quality: Mapped[str] = mapped_column(String(32), nullable=False, default="valid") + degraded: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False) + degraded_reason: Mapped[str | None] = mapped_column(String(255), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False) + + meter: Mapped["Meter | None"] = relationship() + source_binding: Mapped["MeterSourceBinding | None"] = relationship() + contract_version: Mapped["EnergyContractVersion | None"] = relationship() + + __table_args__ = ( + CheckConstraint( + "degraded OR (meter_id IS NOT NULL AND source_binding_id IS NOT NULL " + "AND contract_version_id IS NOT NULL)", + name="ck_meter_cost_period_normal_audit_links", + ), + CheckConstraint("period_end > period_start", name="ck_meter_cost_period_positive_interval"), + CheckConstraint( + "NOT degraded OR (degraded_reason IS NOT NULL AND length(trim(degraded_reason)) > 0)", + name="ck_meter_cost_period_degraded_reason", + ), + UniqueConstraint("commodity", "period_start", name="uq_meter_cost_period_commodity_start"), + Index("ix_meter_cost_period_commodity_start", "commodity", "period_start"), + Index("ix_meter_cost_period_source_binding_id", "source_binding_id"), + ) + + @validates("cost_breakdown", "pricing_snapshot") + def _validate_decimal_json(self, _key: str, value: dict) -> dict: + return _decimal_json(value) + + @validates("quantity", "cost") + def _validate_fixed_decimal(self, key: str, value: Decimal) -> Decimal: + precision, scale = (15, 6) if key == "quantity" else (15, 9) + return _validate_fixed_decimal(value, precision, scale, key) + + @validates("period_start", "period_end") + def _normalise_period(self, key: str, value: datetime) -> datetime: + return _normalise_utc_period(value, key) + + +@event.listens_for(MeterCostPeriod, "before_insert") +@event.listens_for(MeterCostPeriod, "before_update") +def _validate_meter_cost_period(_mapper, _connection, target: MeterCostPeriod) -> None: + _require_aware_period(target.period_start, target.period_end) + if not target.degraded and ( + target.meter_id is None + or target.source_binding_id is None + or target.contract_version_id is None + ): + raise ValueError("normal meter cost periods require meter, binding, and contract version") + if target.degraded and not target.degraded_reason: + raise ValueError("degraded meter cost periods require a degraded_reason") + + # 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.) diff --git a/docs/design/m8-warmtelink-energy.md b/docs/design/m8-warmtelink-energy.md index 5d4eb9d..7daa72e 100644 --- a/docs/design/m8-warmtelink-energy.md +++ b/docs/design/m8-warmtelink-energy.md @@ -810,7 +810,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 ### M8-T14 — 建立通用 Meter Cost Period 账本 -- **Status**: `todo` +- **Status**: `done` - **Depends**: M8-T13 - **Context**: 热力成本不能硬塞进 electricity 专用账本；先建立按 commodity 审计的独立表。 @@ -819,6 +819,7 @@ T01～T06 先把现有 DSMR 安全迁到统一 source/binding；T07～T11 再接 - `create alembic_app/versions/20260822_19_meter_cost_periods.py` - `modify scripts/app_db_adopt.py` - `create tests/test_meter_cost_models.py` +- `modify tests/test_energy_models.py` **Steps** 1. 新建 `MeterCostPeriod`，字段按 §5：commodity、period start/end、nullable Meter/binding、nullable diff --git a/scripts/app_db_adopt.py b/scripts/app_db_adopt.py index 5a3d218..27fb537 100644 --- a/scripts/app_db_adopt.py +++ b/scripts/app_db_adopt.py @@ -15,7 +15,7 @@ if str(PROJECT_ROOT) not in sys.path: from app.config import get_settings -APP_BASELINE_REVISION = "20260822_18_contract_scopes" +APP_BASELINE_REVISION = "20260822_19_meter_cost_periods" class AppDatabaseAdoptionError(RuntimeError): diff --git a/tests/test_energy_models.py b/tests/test_energy_models.py index 6db0890..6a89f0d 100644 --- a/tests/test_energy_models.py +++ b/tests/test_energy_models.py @@ -1251,8 +1251,8 @@ def test_contract_scope_migration_preserves_historical_contract_audit(tmp_path: {"now": now, "pricing": pricing, "version_id": version_id}, ) - command.upgrade(cfg, "head") - command.upgrade(cfg, "head") + command.upgrade(cfg, "20260822_18_contract_scopes") + command.upgrade(cfg, "20260822_18_contract_scopes") with engine.connect() as connection: assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == ( "20260822_18_contract_scopes" @@ -1359,7 +1359,7 @@ def test_contract_scope_migration_audit_failure_restores_revision_17(tmp_path: P engine.dispose() del cfg.attributes["m8_t12_post_ddl_audit_failure"] - command.upgrade(cfg, "head") + command.upgrade(cfg, "20260822_18_contract_scopes") engine = _engine_with_fk(db_url) with engine.connect() as connection: assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == ( diff --git a/tests/test_meter_cost_models.py b/tests/test_meter_cost_models.py new file mode 100644 index 0000000..12542d4 --- /dev/null +++ b/tests/test_meter_cost_models.py @@ -0,0 +1,465 @@ +"""Migration and ORM contracts for the commodity-scoped meter cost ledger.""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone +from decimal import Decimal +from pathlib import Path + +import pytest +from alembic import command +from alembic.config import Config +from sqlalchemy import create_engine, event as sa_event, insert, inspect, select, text +from sqlalchemy.exc import IntegrityError, StatementError +from sqlalchemy.orm import Session + +from app.models.energy import MeterCostPeriod +from scripts.app_db_adopt import APP_BASELINE_REVISION + + +REVISION_18 = "20260822_18_contract_scopes" +REVISION_19 = "20260822_19_meter_cost_periods" +UTC = timezone.utc + + +def _config(database_url: str) -> Config: + config = Config("alembic_app.ini") + config.set_main_option("sqlalchemy.url", database_url) + return config + + +def _fk_engine(database_url: str): + engine = create_engine(database_url) + + @sa_event.listens_for(engine, "connect") + def _enable_foreign_keys(connection, _record) -> None: + connection.execute("PRAGMA foreign_keys = ON") + + return engine + + +def _period(**overrides) -> MeterCostPeriod: + start = datetime(2026, 8, 22, 12, tzinfo=UTC) + values = { + "commodity": "heating", + "period_start": start, + "period_end": start + timedelta(minutes=15), + "quantity": Decimal("999999999.123456"), + "cost": Decimal("999999.123456789"), + "currency": "EUR", + "cost_breakdown": {"heating": "0.012345678"}, + "pricing_snapshot": {"heating": "10.000000", "kind": "district_heating"}, + "quality": "valid", + "degraded": True, + "degraded_reason": "test_fixture_without_audit_links", + "created_at": start, + "updated_at": start, + } + values.update(overrides) + return MeterCostPeriod(**values) + + +def test_meter_cost_period_empty_db_upgrade_shape_and_baseline(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'empty.db'}" + config = _config(database_url) + command.upgrade(config, "head") + command.upgrade(config, "head") + + engine = create_engine(database_url) + inspector = inspect(engine) + columns = {column["name"]: column for column in inspector.get_columns("meter_cost_period")} + assert APP_BASELINE_REVISION == REVISION_19 + assert {"commodity", "period_start", "period_end", "quantity", "cost"} <= columns.keys() + assert columns["meter_id"]["nullable"] + assert columns["source_binding_id"]["nullable"] + assert columns["contract_version_id"]["nullable"] + assert {index["name"] for index in inspector.get_indexes("meter_cost_period")} >= { + "ix_meter_cost_period_commodity_start", + "ix_meter_cost_period_source_binding_id", + } + assert ("commodity", "period_start") in { + tuple(constraint["column_names"]) + for constraint in inspector.get_unique_constraints("meter_cost_period") + } + foreign_keys = { + foreign_key["constrained_columns"][0]: foreign_key + for foreign_key in inspector.get_foreign_keys("meter_cost_period") + } + for column, table in { + "meter_id": "meter", + "source_binding_id": "meter_source_binding", + "contract_version_id": "energy_contract_version", + }.items(): + assert foreign_keys[column]["referred_table"] == table + assert foreign_keys[column]["options"]["ondelete"] == "RESTRICT" + engine.dispose() + + +def test_meter_cost_period_revision_18_upgrade_preserves_electricity_rows(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'revision18.db'}" + config = _config(database_url) + command.upgrade(config, REVISION_18) + engine = create_engine(database_url) + stamp = datetime(2026, 8, 22, 12) + with engine.begin() as connection: + connection.execute( + text( + "INSERT INTO energy_cost_period " + "(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, " + "export_revenue, net_cost, currency, pricing, contract_version_id, " + "meter_id, source_binding_id, degraded, computed_at) " + "VALUES (:stamp, 1, 2, 3, 4, 5, 6, 7, 'EUR', :pricing, NULL, NULL, NULL, 0, :stamp)" + ), + {"stamp": stamp, "pricing": '{"historic":"unchanged"}'}, + ) + command.upgrade(config, "head") + command.upgrade(config, "head") + with engine.connect() as connection: + assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == REVISION_19 + assert connection.execute(text("SELECT COUNT(*) FROM energy_cost_period")).scalar_one() == 1 + assert connection.execute(text("SELECT pricing FROM energy_cost_period")).scalar_one() == ( + '{"historic":"unchanged"}' + ) + assert connection.execute(text("SELECT COUNT(*) FROM meter_cost_period")).scalar_one() == 0 + command.downgrade(config, REVISION_18) + assert "meter_cost_period" not in inspect(engine).get_table_names() + engine.dispose() + + +def test_meter_cost_period_decimal_json_and_degraded_round_trip(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'roundtrip.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + with Session(engine) as session: + normal = _period() + degraded = _period( + commodity="hot_water", + meter_id=None, + source_binding_id=None, + contract_version_id=None, + quality="unverifiable", + degraded=True, + degraded_reason="missing_binding", + quantity=Decimal("0.000000"), + cost=Decimal("0.000000000"), + cost_breakdown={"hot_water": "0.000000000"}, + pricing_snapshot={"reason": "missing_binding"}, + ) + session.add_all((normal, degraded)) + session.commit() + session.expire_all() + stored = session.get(MeterCostPeriod, normal.id) + assert stored is not None + assert stored.quantity == Decimal("999999999.123456") + assert stored.cost == Decimal("999999.123456789") + assert stored.cost_breakdown == {"heating": "0.012345678"} + assert all(not isinstance(value, float) for value in stored.pricing_snapshot.values()) + assert session.get(MeterCostPeriod, degraded.id).degraded_reason == "missing_binding" + engine.dispose() + + +def test_meter_cost_period_uses_decimal_text_and_normalises_json_snapshots(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'exact.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + with Session(engine) as session: + row = _period( + cost_breakdown={"nested": [Decimal("2.000000000")]}, + pricing_snapshot={"rate": Decimal("2.000000")}, + ) + session.add(row) + session.commit() + session.expire_all() + stored = session.get(MeterCostPeriod, row.id) + assert stored is not None + assert stored.cost_breakdown == {"nested": ["2.000000000"]} + assert stored.pricing_snapshot == {"rate": "2.000000"} + raw = session.execute(text("SELECT typeof(quantity), typeof(cost) FROM meter_cost_period")).one() + assert raw == ("text", "text") + with pytest.raises(ValueError, match="numeric JSON"): + _period(cost_breakdown={"nested": [1]}) + with pytest.raises(ValueError, match="numeric JSON"): + _period(pricing_snapshot={"nested": [1.25]}) + engine.dispose() + + +def test_meter_cost_period_core_json_bind_normalises_decimal_and_rejects_numeric(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'core-json.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + values = { + "commodity": "heating", + "period_start": datetime(2026, 8, 22, 12, tzinfo=UTC), + "period_end": datetime(2026, 8, 22, 12, 15, tzinfo=UTC), + "quantity": Decimal("0.000000"), + "cost": Decimal("0.000000000"), + "currency": "EUR", + "cost_breakdown": {"nested": [Decimal("2.000000000")]}, + "pricing_snapshot": {"rate": Decimal("2.000000")}, + "quality": "valid", + "degraded": True, + "degraded_reason": "core fixture", + "created_at": datetime(2026, 8, 22, 12, tzinfo=UTC), + "updated_at": datetime(2026, 8, 22, 12, tzinfo=UTC), + } + with engine.begin() as connection: + connection.execute(insert(MeterCostPeriod.__table__).values(values)) + with Session(engine) as session: + stored = session.scalar(select(MeterCostPeriod)) + assert stored is not None + assert stored.cost_breakdown == {"nested": ["2.000000000"]} + assert stored.pricing_snapshot == {"rate": "2.000000"} + for field, numeric_value in (("cost_breakdown", {"nested": [1]}), ("pricing_snapshot", {"rate": 1.25})): + with engine.begin() as connection, pytest.raises(StatementError, match="numeric JSON"): + connection.execute( + insert(MeterCostPeriod.__table__).values( + {**values, "commodity": f"invalid-{field}", field: numeric_value} + ) + ) + engine.dispose() + + +@pytest.mark.parametrize( + ("field", "value", "message"), + [ + ("quantity", Decimal("1000000000.000000"), "precision"), + ("quantity", Decimal("1.1234567"), "scale"), + ("cost", Decimal("1000000.000000000"), "precision"), + ("cost", Decimal("1.1234567899"), "scale"), + ], +) +def test_meter_cost_period_rejects_decimal_precision_and_scale_overflow(field, value, message) -> None: + with pytest.raises(ValueError, match=message): + _period(**{field: value}) + + +def test_meter_cost_period_accepts_signed_decimal_boundaries_without_float_bind(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'boundaries.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + with Session(engine) as session: + rows = ( + _period(quantity=Decimal("999999999.999999"), cost=Decimal("999999.999999999")), + _period( + commodity="hot_water", + quantity=Decimal("-999999999.999999"), + cost=Decimal("-999999.999999999"), + ), + ) + session.add_all(rows) + session.commit() + session.expire_all() + assert session.get(MeterCostPeriod, rows[0].id).quantity == Decimal("999999999.999999") + assert session.get(MeterCostPeriod, rows[1].id).cost == Decimal("-999999.999999999") + engine.dispose() + + +def test_meter_cost_period_rejects_duplicate_period_invalid_interval_and_json_float(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'constraints.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + with Session(engine) as session: + first = _period() + session.add(first) + session.commit() + session.add(_period(cost=Decimal("1.000000000"))) + with pytest.raises(IntegrityError): + session.commit() + session.rollback() + session.add(_period(commodity="hot_water", period_end=datetime(2026, 8, 22, 12, tzinfo=UTC))) + with pytest.raises(ValueError, match="after period_start"): + session.flush() + session.rollback() + with pytest.raises(ValueError, match="timezone-aware"): + _period( + commodity="hot_water", + period_start=datetime(2026, 8, 22, 12), + period_end=datetime(2026, 8, 22, 12, 15, tzinfo=UTC), + ) + with pytest.raises(ValueError, match="decimal strings"): + _period(pricing_snapshot={"heating": 1.25}) + with pytest.raises(ValueError, match="normal meter cost periods"): + session.add(_period(commodity="hot_water", degraded=False, degraded_reason=None)) + session.flush() + session.rollback() + engine.dispose() + + +def test_meter_cost_period_uses_utc_instant_idempotency_and_db_constraints(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'utc.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + utc_start = datetime(2026, 8, 22, 12, tzinfo=UTC) + offset_start = datetime(2026, 8, 22, 14, tzinfo=timezone(timedelta(hours=2))) + with Session(engine) as session: + first = _period(period_start=utc_start, period_end=utc_start + timedelta(minutes=15)) + assert first.period_start == utc_start + session.add(first) + session.commit() + session.add( + _period( + period_start=offset_start, + period_end=offset_start + timedelta(minutes=15), + cost=Decimal("1.000000000"), + ) + ) + with pytest.raises(IntegrityError): + session.commit() + session.rollback() + with engine.begin() as connection: + values = { + "start": "2026-08-23 00:00:00.000000", + "quantity": "0.000000", + "cost": "0.000000000", + "now": "2026-08-23 00:00:00.000000", + } + for commodity, end in (("equal", values["start"]), ("reversed", "2026-08-22 23:59:59.000000")): + with pytest.raises(IntegrityError): + connection.execute( + text( + "INSERT INTO meter_cost_period " + "(commodity, period_start, period_end, quantity, cost, currency, cost_breakdown, " + "pricing_snapshot, quality, degraded, degraded_reason, created_at, updated_at) " + "VALUES (:commodity, :start, :end, :quantity, :cost, 'EUR', '{}', '{}', 'valid', " + "1, 'core test', :now, :now)" + ), + {**values, "commodity": commodity, "end": end}, + ) + with pytest.raises(IntegrityError): + connection.execute( + text( + "INSERT INTO meter_cost_period " + "(commodity, period_start, period_end, quantity, cost, currency, cost_breakdown, " + "pricing_snapshot, quality, degraded, degraded_reason, created_at, updated_at) " + "VALUES ('missing-reason', :start, '2026-08-23 00:15:00.000000', :quantity, :cost, " + "'EUR', '{}', '{}', 'valid', 1, NULL, :now, :now)" + ), + values, + ) + engine.dispose() + + +def test_meter_cost_period_core_uses_utc_instant_idempotency_and_reloads_aware(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'core-utc.db'}" + command.upgrade(_config(database_url), "head") + engine = create_engine(database_url) + utc_start = datetime(2026, 8, 22, 12, tzinfo=UTC) + values = { + "commodity": "heating", + "period_start": utc_start, + "period_end": utc_start + timedelta(minutes=15), + "quantity": Decimal("0.000000"), + "cost": Decimal("0.000000000"), + "currency": "EUR", + "cost_breakdown": {}, + "pricing_snapshot": {}, + "quality": "valid", + "degraded": True, + "degraded_reason": "core fixture", + "created_at": utc_start, + "updated_at": utc_start, + } + with engine.begin() as connection: + row_id = connection.execute(insert(MeterCostPeriod.__table__).values(values)).inserted_primary_key[0] + with engine.begin() as connection, pytest.raises(IntegrityError): + connection.execute( + insert(MeterCostPeriod.__table__).values( + { + **values, + "period_start": datetime(2026, 8, 22, 14, tzinfo=timezone(timedelta(hours=2))), + "period_end": datetime(2026, 8, 22, 14, 15, tzinfo=timezone(timedelta(hours=2))), + } + ) + ) + with engine.connect() as connection, pytest.raises(StatementError, match="timezone-aware"): + connection.execute( + insert(MeterCostPeriod.__table__).values( + {**values, "commodity": "naive", "period_start": datetime(2026, 8, 23, 12)} + ) + ) + with Session(engine) as session: + stored = session.get(MeterCostPeriod, row_id) + assert stored is not None + assert stored.period_start == utc_start + assert stored.period_start.tzinfo is not None + assert stored.period_start.utcoffset() == timedelta(0) + stored.quality = "unverifiable" + session.commit() + engine.dispose() + + +def test_meter_cost_period_foreign_keys_restrict_deletion(tmp_path: Path) -> None: + database_url = f"sqlite:///{tmp_path / 'foreign_keys.db'}" + command.upgrade(_config(database_url), "head") + engine = _fk_engine(database_url) + timestamp = datetime(2026, 8, 22, 12) + with engine.begin() as connection: + meter_id = connection.execute( + text( + "INSERT INTO meter (uuid, label, commodity, started_at, ended_at, reason, note, created_at) " + "VALUES ('meter-cost-test', 'Test', 'heating', :timestamp, NULL, 'initial', NULL, :timestamp)" + ), + {"timestamp": timestamp}, + ).lastrowid + source_id = connection.execute( + text( + "INSERT INTO meter_source (uuid, name, kind, enabled, config, status, created_at, updated_at) " + "VALUES ('source-cost-test', 'Test', 'warmtelink_serial', 1, '{}', 'online', :timestamp, :timestamp)" + ), + {"timestamp": timestamp}, + ).lastrowid + channel_id = connection.execute( + text( + "INSERT INTO meter_source_channel (uuid, source_id, channel_key, label, unit, created_at, updated_at) " + "VALUES ('channel-cost-test', :source_id, 'heating', 'Heating', 'GJ', :timestamp, :timestamp)" + ), + {"source_id": source_id, "timestamp": timestamp}, + ).lastrowid + binding_id = connection.execute( + text( + "INSERT INTO meter_source_binding " + "(uuid, meter_id, channel_id, started_at, ended_at, created_at, updated_at) " + "VALUES ('binding-cost-test', :meter_id, :channel_id, :timestamp, NULL, :timestamp, :timestamp)" + ), + {"meter_id": meter_id, "channel_id": channel_id, "timestamp": timestamp}, + ).lastrowid + contract_id = connection.execute( + text( + "INSERT INTO energy_contract (name, kind, scope, active, currency, created_at, updated_at) " + "VALUES ('Test', 'district_heating', 'thermal', 1, 'EUR', :timestamp, :timestamp)" + ), + {"timestamp": timestamp}, + ).lastrowid + version_id = connection.execute( + text( + "INSERT INTO energy_contract_version " + "(contract_id, effective_from, effective_to, \"values\", created_at) " + "VALUES (:contract_id, :timestamp, NULL, '{}', :timestamp)" + ), + {"contract_id": contract_id, "timestamp": timestamp}, + ).lastrowid + connection.execute( + text( + "INSERT INTO meter_cost_period " + "(commodity, period_start, period_end, meter_id, source_binding_id, contract_version_id, " + "quantity, cost, currency, cost_breakdown, pricing_snapshot, quality, degraded, degraded_reason, " + "created_at, updated_at) VALUES " + "('heating', :timestamp, :period_end, :meter_id, :binding_id, :version_id, " + "'0.001000', '0.010000000', 'EUR', '{}', '{}', 'valid', 0, NULL, :timestamp, :timestamp)" + ), + { + "timestamp": timestamp, + "period_end": timestamp + timedelta(minutes=15), + "meter_id": meter_id, + "binding_id": binding_id, + "version_id": version_id, + }, + ) + for statement, values in ( + ("DELETE FROM meter WHERE id = :id", {"id": meter_id}), + ("DELETE FROM meter_source_binding WHERE id = :id", {"id": binding_id}), + ("DELETE FROM energy_contract_version WHERE id = :id", {"id": version_id}), + ): + with pytest.raises(IntegrityError): + connection.execute(text(statement), values) + engine.dispose()