291 lines
14 KiB
Python
291 lines
14 KiB
Python
"""Isolated revision-14/15 fixtures for the DSMR source-adoption migration."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import sqlalchemy.exc
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, event, inspect, text
|
|
|
|
from app.integrations.meter_sources import sanitize_source_config, validate_source_config
|
|
|
|
|
|
def _config(database_url: str) -> Config:
|
|
config = Config("alembic_app.ini")
|
|
config.set_main_option("sqlalchemy.url", database_url)
|
|
return config
|
|
|
|
|
|
def _engine(database_url: str):
|
|
engine = create_engine(database_url, connect_args={"check_same_thread": False})
|
|
|
|
@event.listens_for(engine, "connect")
|
|
def _foreign_keys(connection, _record) -> None:
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
|
|
return engine
|
|
|
|
|
|
def _insert_meter(connection, label: str, started: datetime, ended: datetime | None) -> int:
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO meter (uuid, label, commodity, started_at, ended_at, reason, note, created_at) "
|
|
"VALUES (:uuid, :label, 'electricity', :started, :ended, 'initial', NULL, :started)"
|
|
),
|
|
{"uuid": f"{label:0<8}-0000-4000-8000-000000000000", "label": label,
|
|
"started": started, "ended": ended},
|
|
)
|
|
return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
|
|
|
|
|
|
def _insert_contract_version(connection, start: datetime) -> int:
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO energy_contract (name, kind, active, currency, created_at, updated_at) "
|
|
"VALUES ('Historic contract', 'manual', 1, 'EUR', :at, :at)"
|
|
),
|
|
{"at": start},
|
|
)
|
|
contract_id = int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO energy_contract_version "
|
|
"(contract_id, effective_from, effective_to, \"values\", created_at) "
|
|
"VALUES (:contract_id, :at, NULL, :values, :at)"
|
|
),
|
|
{"contract_id": contract_id, "at": start, "values": json.dumps({"historic": True})},
|
|
)
|
|
return int(connection.execute(text("SELECT last_insert_rowid()")).scalar_one())
|
|
|
|
|
|
def _insert_cost(
|
|
connection,
|
|
period_start: datetime,
|
|
meter_id: int,
|
|
contract_version_id: int,
|
|
sequence: int,
|
|
) -> None:
|
|
amount = 2.5 + sequence
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO energy_cost_period "
|
|
"(period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, export_revenue, net_cost, "
|
|
"currency, pricing, contract_version_id, meter_id, degraded, computed_at) "
|
|
"VALUES (:start, :d1, :d2, :r1, :r2, :import_cost, :export_revenue, :net_cost, "
|
|
"'EUR', :pricing, :contract_version_id, :meter, 0, :computed_at)"
|
|
),
|
|
{
|
|
"start": period_start,
|
|
"d1": 1.0 + sequence,
|
|
"d2": 2.0 + sequence,
|
|
"r1": 3.0 + sequence,
|
|
"r2": 4.0 + sequence,
|
|
"import_cost": amount,
|
|
"export_revenue": 0.25 + sequence,
|
|
"net_cost": amount - (0.25 + sequence),
|
|
"pricing": json.dumps({"historic": True, "sequence": sequence}),
|
|
"contract_version_id": contract_version_id,
|
|
"meter": meter_id,
|
|
"computed_at": period_start + timedelta(seconds=sequence),
|
|
},
|
|
)
|
|
|
|
|
|
def test_populated_revision_14_adopts_dsmr_history_at_revision_16(tmp_path: Path):
|
|
database_url = f"sqlite:///{tmp_path / 'revision_14_history.db'}"
|
|
config = _config(database_url)
|
|
command.upgrade(config, "20260625_14_meter_uuid")
|
|
start = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
|
engine = _engine(database_url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(
|
|
text("INSERT INTO app_config (key, value, updated_at) VALUES (:key, :value, :at)"),
|
|
[
|
|
{"key": "DSMR_INGEST_ENABLED", "value": "true", "at": start},
|
|
{"key": "DSMR_MQTT_TOPIC", "value": "historic/dsmr", "at": start},
|
|
{"key": "DSMR_TARIFF_TOPIC", "value": "historic/tariff", "at": start},
|
|
{"key": "DSMR_SAMPLE_INTERVAL_S", "value": "0", "at": start},
|
|
{"key": "MQTT_BROKER_HOST", "value": "mqtt.example.invalid", "at": start},
|
|
{"key": "MQTT_BROKER_PORT", "value": "1884", "at": start},
|
|
{"key": "MQTT_USERNAME", "value": "historic-user", "at": start},
|
|
{"key": "MQTT_PASSWORD", "value": "historic-password", "at": start},
|
|
{"key": "MQTT_TLS_ENABLED", "value": "true", "at": start},
|
|
{"key": "UNRELATED_CONFIG", "value": "untouched", "at": start},
|
|
],
|
|
)
|
|
config_before = dict(connection.execute(text("SELECT key, value FROM app_config")).all())
|
|
for offset, telegram_id in ((0, 77), (20, 78), (40, 77)):
|
|
connection.execute(
|
|
text("INSERT INTO dsmr_reading (recorded_at, source_id, payload) VALUES (:at, :id, :payload)"),
|
|
{"at": start + timedelta(minutes=offset), "id": telegram_id,
|
|
"payload": json.dumps({"id": telegram_id, "keep": f"payload-{offset}"})},
|
|
)
|
|
first = _insert_meter(connection, "meterone", start - timedelta(hours=1), start + timedelta(minutes=20))
|
|
second = _insert_meter(connection, "metertwo", start + timedelta(minutes=20), start + timedelta(minutes=40))
|
|
third = _insert_meter(connection, "meterthree", start + timedelta(minutes=40), None)
|
|
_insert_meter(connection, "nodata", start + timedelta(days=1), None)
|
|
contract_version = _insert_contract_version(connection, start - timedelta(days=1))
|
|
# One normal period per epoch plus a period ending exactly at each
|
|
# replacement boundary. Meter/binding intervals are half-open, so
|
|
# the latter must remain unbound/degraded.
|
|
_insert_cost(connection, start, first, contract_version, 0)
|
|
_insert_cost(connection, start + timedelta(minutes=5), first, contract_version, 1)
|
|
_insert_cost(connection, start + timedelta(minutes=21), second, contract_version, 2)
|
|
_insert_cost(connection, start + timedelta(minutes=25), second, contract_version, 3)
|
|
_insert_cost(connection, start + timedelta(minutes=41), third, contract_version, 4)
|
|
cost_before = connection.execute(
|
|
text(
|
|
"SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
|
|
"export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, "
|
|
"degraded, computed_at FROM energy_cost_period ORDER BY period_start"
|
|
)
|
|
).mappings().all()
|
|
finally:
|
|
engine.dispose()
|
|
|
|
command.upgrade(config, "20260822_16_dsmr_source_adoption")
|
|
engine = _engine(database_url)
|
|
try:
|
|
with engine.connect() as connection:
|
|
assert connection.execute(text("SELECT version_num FROM alembic_version")).scalar_one() == (
|
|
"20260822_16_dsmr_source_adoption"
|
|
)
|
|
assert connection.execute(text("SELECT COUNT(*) FROM dsmr_reading")).scalar_one() == 3
|
|
source = connection.execute(
|
|
text("SELECT id, enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
|
|
).one()
|
|
assert source.enabled == 1
|
|
source_config = json.loads(source.config)
|
|
assert source_config == {
|
|
"broker_host": "mqtt.example.invalid", "broker_port": 1884,
|
|
"username": "historic-user", "password": "historic-password", "tls_enabled": True,
|
|
"topic": "historic/dsmr", "tariff_topic": "historic/tariff", "sample_interval_s": 0,
|
|
}
|
|
assert validate_source_config("dsmr_mqtt", source_config) == source_config
|
|
assert sanitize_source_config("dsmr_mqtt", source_config)["sample_interval_s"] == 0
|
|
assert connection.execute(text("SELECT value FROM app_config WHERE key = 'DSMR_MQTT_TOPIC'")).scalar_one() == "historic/dsmr"
|
|
assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == config_before
|
|
assert connection.execute(text("SELECT group_concat(telegram_id) FROM dsmr_reading")).scalar_one() == "77,78,77"
|
|
assert connection.execute(text("SELECT payload FROM dsmr_reading ORDER BY recorded_at")).scalars().all() == [
|
|
json.dumps({"id": 77, "keep": "payload-0"}),
|
|
json.dumps({"id": 78, "keep": "payload-20"}),
|
|
json.dumps({"id": 77, "keep": "payload-40"}),
|
|
]
|
|
assert connection.execute(text("SELECT COUNT(*) FROM meter_source_binding")).scalar_one() == 3
|
|
assert connection.execute(
|
|
text("SELECT COUNT(*) FROM meter_source_binding WHERE meter_id = (SELECT id FROM meter WHERE label = 'nodata')")
|
|
).scalar_one() == 0
|
|
periods = connection.execute(
|
|
text(
|
|
"SELECT id, period_start, d1_kwh, d2_kwh, r1_kwh, r2_kwh, import_cost, "
|
|
"export_revenue, net_cost, currency, pricing, contract_version_id, meter_id, "
|
|
"degraded, computed_at, source_binding_id FROM energy_cost_period ORDER BY period_start"
|
|
)
|
|
).mappings().all()
|
|
assert [
|
|
{key: value for key, value in period.items() if key not in {"degraded", "source_binding_id"}}
|
|
for period in periods
|
|
] == [
|
|
{key: value for key, value in period.items() if key != "degraded"}
|
|
for period in cost_before
|
|
]
|
|
assert [(period["degraded"], period["source_binding_id"] is not None) for period in periods] == [
|
|
(0, True), (1, False), (0, True), (1, False), (0, True)
|
|
]
|
|
assert [period["period_start"] for period in periods if period["source_binding_id"] is None] == [
|
|
(start + timedelta(minutes=5)).isoformat(sep=" "),
|
|
(start + timedelta(minutes=25)).isoformat(sep=" "),
|
|
]
|
|
bound_meter_ids = connection.execute(
|
|
text(
|
|
"SELECT binding.meter_id FROM energy_cost_period AS period "
|
|
"LEFT JOIN meter_source_binding AS binding "
|
|
"ON binding.id = period.source_binding_id ORDER BY period.period_start"
|
|
)
|
|
).scalars().all()
|
|
assert bound_meter_ids == [first, None, second, None, third]
|
|
assert connection.execute(text("PRAGMA foreign_key_check")).all() == []
|
|
finally:
|
|
engine.dispose()
|
|
|
|
|
|
def test_dsmr_source_timestamp_uniqueness_allows_two_sources(tmp_path: Path):
|
|
database_url = f"sqlite:///{tmp_path / 'two_sources.db'}"
|
|
config = _config(database_url)
|
|
command.upgrade(config, "20260822_16_dsmr_source_adoption")
|
|
engine = _engine(database_url)
|
|
timestamp = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
|
try:
|
|
with engine.begin() as connection:
|
|
first_source = connection.execute(text("SELECT id FROM meter_source WHERE kind = 'dsmr_mqtt'")).scalar_one()
|
|
assert connection.execute(
|
|
text("SELECT enabled FROM meter_source WHERE id = :id"), {"id": first_source}
|
|
).scalar_one() == 0
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO meter_source (uuid, name, kind, enabled, config, status, created_at, updated_at) "
|
|
"VALUES ('22222222-2222-4222-8222-222222222222', 'Second DSMR', 'dsmr_mqtt', 0, '{}', "
|
|
"'unknown', :at, :at)"
|
|
), {"at": timestamp},
|
|
)
|
|
second_source = connection.execute(text("SELECT last_insert_rowid()")).scalar_one()
|
|
for source_id in (first_source, second_source):
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) "
|
|
"VALUES (:at, 9, :source, '{}')"
|
|
), {"at": timestamp, "source": source_id},
|
|
)
|
|
with pytest.raises(sqlalchemy.exc.IntegrityError):
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO dsmr_reading (recorded_at, telegram_id, meter_source_id, payload) "
|
|
"VALUES (:at, 10, :source, '{}')"
|
|
), {"at": timestamp, "source": first_source},
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
inspector = inspect(create_engine(database_url))
|
|
assert ("meter_source_id", "recorded_at") in {
|
|
tuple(item["column_names"]) for item in inspector.get_unique_constraints("dsmr_reading")
|
|
}
|
|
|
|
|
|
def test_revision_15_without_legacy_dsmr_config_creates_unconfigured_source(tmp_path: Path):
|
|
database_url = f"sqlite:///{tmp_path / 'revision_15_no_dsmr_config.db'}"
|
|
config = _config(database_url)
|
|
command.upgrade(config, "20260822_15_meter_sources")
|
|
engine = _engine(database_url)
|
|
try:
|
|
with engine.begin() as connection:
|
|
connection.execute(
|
|
text(
|
|
"INSERT INTO app_config (key, value, updated_at) "
|
|
"VALUES ('UNRELATED_CONFIG', 'untouched', :at)"
|
|
),
|
|
{"at": datetime(2026, 8, 1, tzinfo=timezone.utc)},
|
|
)
|
|
finally:
|
|
engine.dispose()
|
|
|
|
command.upgrade(config, "20260822_16_dsmr_source_adoption")
|
|
engine = _engine(database_url)
|
|
try:
|
|
with engine.connect() as connection:
|
|
source = connection.execute(
|
|
text("SELECT enabled, config FROM meter_source WHERE kind = 'dsmr_mqtt'")
|
|
).one()
|
|
assert source.enabled == 0
|
|
assert json.loads(source.config) == {}
|
|
assert dict(connection.execute(text("SELECT key, value FROM app_config")).all()) == {
|
|
"UNRELATED_CONFIG": "untouched"
|
|
}
|
|
finally:
|
|
engine.dispose()
|