465 lines
17 KiB
Python
465 lines
17 KiB
Python
"""Tests for M5-T02: modbus_device + modbus_reading tables and ORM models.
|
|
|
|
Covers:
|
|
1. Migration shape: upgrade to head creates both tables with correct columns,
|
|
constraints, and indexes; downgrade -1 cleanly removes them.
|
|
2. ORM metadata: Base.metadata.tables contains both tables; FK is RESTRICT;
|
|
uuid column is unique and auto-generated.
|
|
3. Baseline constant: APP_BASELINE_REVISION matches the actual Alembic head.
|
|
4. Basic ORM round-trip: insert + retrieve, uuid auto-population, nullable fields.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from alembic import command
|
|
from alembic.config import Config
|
|
from sqlalchemy import create_engine, inspect, text
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.db import Base
|
|
from app.models.modbus import ModbusDevice, ModbusReading
|
|
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
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def modbus_db(tmp_path: Path):
|
|
"""Temporary SQLite DB upgraded to the current Alembic head."""
|
|
db_path = tmp_path / "modbus_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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Migration shape tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_modbus_tables_exist_after_upgrade(modbus_db):
|
|
"""Both modbus_device and modbus_reading tables must exist after upgrade to head."""
|
|
inspector = inspect(modbus_db)
|
|
table_names = inspector.get_table_names()
|
|
assert "modbus_device" in table_names, "modbus_device table missing after upgrade"
|
|
assert "modbus_reading" in table_names, "modbus_reading table missing after upgrade"
|
|
|
|
|
|
def test_modbus_device_columns(modbus_db):
|
|
"""modbus_device must have all required columns with correct nullability."""
|
|
inspector = inspect(modbus_db)
|
|
columns = {col["name"]: col for col in inspector.get_columns("modbus_device")}
|
|
|
|
expected_non_nullable = {
|
|
"id", "uuid", "friendly_name", "transport", "host", "port",
|
|
"unit_id", "profile", "poll_interval_s", "enabled",
|
|
"created_at", "updated_at",
|
|
}
|
|
expected_nullable = {"last_poll_at", "last_poll_ok"}
|
|
|
|
for col_name in expected_non_nullable:
|
|
assert col_name in columns, f"Missing column: {col_name}"
|
|
assert not columns[col_name]["nullable"], f"Column {col_name} should be NOT NULL"
|
|
|
|
for col_name in expected_nullable:
|
|
assert col_name in columns, f"Missing column: {col_name}"
|
|
assert columns[col_name]["nullable"], f"Column {col_name} should be nullable"
|
|
|
|
|
|
def test_modbus_reading_columns(modbus_db):
|
|
"""modbus_reading must have id, device_id, recorded_at, payload with correct nullability."""
|
|
inspector = inspect(modbus_db)
|
|
columns = {col["name"]: col for col in inspector.get_columns("modbus_reading")}
|
|
|
|
expected_non_nullable = {"id", "device_id", "recorded_at", "payload"}
|
|
for col_name in expected_non_nullable:
|
|
assert col_name in columns, f"Missing column: {col_name}"
|
|
assert not columns[col_name]["nullable"], f"Column {col_name} should be NOT NULL"
|
|
|
|
|
|
def test_modbus_device_uuid_is_unique(modbus_db):
|
|
"""modbus_device.uuid must have a unique constraint."""
|
|
inspector = inspect(modbus_db)
|
|
unique_constraints = inspector.get_unique_constraints("modbus_device")
|
|
unique_cols = [
|
|
col
|
|
for uc in unique_constraints
|
|
for col in uc["column_names"]
|
|
]
|
|
assert "uuid" in unique_cols, "uuid column must have a unique constraint"
|
|
|
|
|
|
def test_modbus_reading_fk_to_device(modbus_db):
|
|
"""modbus_reading.device_id must have a FK referencing modbus_device.id."""
|
|
inspector = inspect(modbus_db)
|
|
fks = inspector.get_foreign_keys("modbus_reading")
|
|
assert len(fks) == 1, f"Expected 1 FK on modbus_reading, got {len(fks)}"
|
|
fk = fks[0]
|
|
assert fk["referred_table"] == "modbus_device"
|
|
assert "device_id" in fk["constrained_columns"]
|
|
assert "id" in fk["referred_columns"]
|
|
|
|
|
|
def test_modbus_reading_composite_index_exists(modbus_db):
|
|
"""Composite index (device_id, recorded_at) must exist on modbus_reading."""
|
|
inspector = inspect(modbus_db)
|
|
indexes = {idx["name"]: idx for idx in inspector.get_indexes("modbus_reading")}
|
|
assert "ix_modbus_reading_device_recorded" in indexes, (
|
|
"Composite index ix_modbus_reading_device_recorded missing"
|
|
)
|
|
composite_idx = indexes["ix_modbus_reading_device_recorded"]
|
|
assert "device_id" in composite_idx["column_names"]
|
|
assert "recorded_at" in composite_idx["column_names"]
|
|
|
|
|
|
def test_modbus_reading_recorded_at_index_exists(modbus_db):
|
|
"""Individual index on recorded_at must exist on modbus_reading."""
|
|
inspector = inspect(modbus_db)
|
|
indexes = {idx["name"]: idx for idx in inspector.get_indexes("modbus_reading")}
|
|
# The ORM-level index=True creates ix_modbus_reading_recorded_at
|
|
assert "ix_modbus_reading_recorded_at" in indexes, (
|
|
"Index ix_modbus_reading_recorded_at missing from modbus_reading"
|
|
)
|
|
|
|
|
|
def test_downgrade_removes_modbus_tables(tmp_path: Path):
|
|
"""Downgrading past the modbus migration must drop both modbus tables cleanly.
|
|
|
|
We downgrade to the explicit revision ``20260621_08_totp`` (the down_revision
|
|
of the modbus tables migration), so the test stays correct as new revisions
|
|
are added on top. Using an explicit target rather than ``-2`` avoids drift
|
|
with chain growth (same fix pattern as T02 applied to the TOTP downgrade test).
|
|
"""
|
|
db_path = tmp_path / "downgrade_modbus_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)
|
|
assert "modbus_device" in inspector.get_table_names()
|
|
assert "modbus_reading" in inspector.get_table_names()
|
|
engine.dispose()
|
|
|
|
# Downgrade to the explicit down_revision of the modbus tables migration
|
|
# ("20260621_08_totp"), consistent with how T02 fixed the TOTP downgrade test.
|
|
# Using an explicit target instead of "-2" ensures the test stays correct even
|
|
# as new revisions are added on top of this one (no drift with chain growth).
|
|
command.downgrade(alembic_cfg, "20260621_08_totp")
|
|
|
|
engine = create_engine(db_url, connect_args={"check_same_thread": False})
|
|
inspector = inspect(engine)
|
|
table_names = inspector.get_table_names()
|
|
assert "modbus_device" not in table_names, "modbus_device should be gone after downgrade"
|
|
assert "modbus_reading" not in table_names, "modbus_reading should be gone after downgrade"
|
|
engine.dispose()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. ORM metadata checks (Base.metadata)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_base_metadata_contains_modbus_tables():
|
|
"""Base.metadata.tables must include both new modbus tables."""
|
|
assert "modbus_device" in Base.metadata.tables, "modbus_device not in Base.metadata"
|
|
assert "modbus_reading" in Base.metadata.tables, "modbus_reading not in Base.metadata"
|
|
|
|
|
|
def test_modbus_reading_fk_ondelete_restrict():
|
|
"""The FK from modbus_reading.device_id to modbus_device.id must be ON DELETE RESTRICT."""
|
|
reading_table = Base.metadata.tables["modbus_reading"]
|
|
fk_columns = {
|
|
col.name: col
|
|
for col in reading_table.columns
|
|
}
|
|
device_id_col = fk_columns["device_id"]
|
|
assert device_id_col.foreign_keys, "device_id must have a foreign key"
|
|
fk = next(iter(device_id_col.foreign_keys))
|
|
assert fk.ondelete == "RESTRICT", (
|
|
f"FK ondelete must be RESTRICT, got: {fk.ondelete!r}"
|
|
)
|
|
|
|
|
|
def test_modbus_device_uuid_unique_constraint_in_metadata():
|
|
"""ModbusDevice.uuid column must be declared unique in ORM metadata."""
|
|
device_table = Base.metadata.tables["modbus_device"]
|
|
uuid_col = device_table.columns["uuid"]
|
|
assert uuid_col.unique, "ModbusDevice.uuid 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_modbus_device_uuid_auto_generated(modbus_db):
|
|
"""uuid must be auto-populated when not explicitly provided."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(modbus_db) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="Test Meter",
|
|
host="192.168.1.100",
|
|
profile="sdm120",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.commit()
|
|
device_id = device.id
|
|
|
|
with Session(modbus_db) as session:
|
|
fetched = session.get(ModbusDevice, device_id)
|
|
assert fetched is not None
|
|
assert fetched.uuid is not None, "uuid should be auto-populated"
|
|
assert len(fetched.uuid) == 36, "uuid should be a standard UUID4 string (36 chars)"
|
|
# Verify it's a valid UUID4 format
|
|
import uuid as _uuid_mod
|
|
parsed = _uuid_mod.UUID(fetched.uuid)
|
|
assert parsed.version == 4, "uuid should be version 4"
|
|
|
|
|
|
def test_modbus_device_defaults(modbus_db):
|
|
"""Default values for port, unit_id, poll_interval_s, enabled, transport must apply."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(modbus_db) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="Default Test",
|
|
host="10.0.0.1",
|
|
profile="sdm120",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.commit()
|
|
device_id = device.id
|
|
|
|
with Session(modbus_db) as session:
|
|
fetched = session.get(ModbusDevice, device_id)
|
|
assert fetched is not None
|
|
assert fetched.port == 502
|
|
assert fetched.unit_id == 1
|
|
assert fetched.poll_interval_s == 5
|
|
assert fetched.enabled is True
|
|
assert fetched.transport == "tcp"
|
|
assert fetched.last_poll_at is None
|
|
assert fetched.last_poll_ok is None
|
|
|
|
|
|
def test_modbus_device_two_unique_uuids(modbus_db):
|
|
"""Two independently created devices must have different auto-generated UUIDs."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(modbus_db) as session:
|
|
d1 = ModbusDevice(
|
|
friendly_name="Device 1", host="10.0.0.1", profile="sdm120",
|
|
created_at=now, updated_at=now,
|
|
)
|
|
d2 = ModbusDevice(
|
|
friendly_name="Device 2", host="10.0.0.2", profile="sdm120",
|
|
created_at=now, updated_at=now,
|
|
)
|
|
session.add_all([d1, d2])
|
|
session.commit()
|
|
assert d1.uuid != d2.uuid, "Two devices must receive distinct UUIDs"
|
|
|
|
|
|
def test_modbus_reading_insert_and_retrieve(modbus_db):
|
|
"""A ModbusReading can be inserted with a JSON payload and retrieved correctly."""
|
|
now = datetime.now(tz=timezone.utc)
|
|
sample_payload = {
|
|
"voltage": 230.2,
|
|
"current": 1.3,
|
|
"active_power": 295.0,
|
|
"power_factor": 0.98,
|
|
"frequency": 50.0,
|
|
"import_energy": 123.4,
|
|
"export_energy": 0.0,
|
|
"total_energy": 123.4,
|
|
}
|
|
|
|
with Session(modbus_db) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="SDM120 AC",
|
|
host="192.168.1.100",
|
|
profile="sdm120",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.flush()
|
|
|
|
reading = ModbusReading(
|
|
device_id=device.id,
|
|
recorded_at=now,
|
|
payload=sample_payload,
|
|
)
|
|
session.add(reading)
|
|
session.commit()
|
|
reading_id = reading.id
|
|
|
|
with Session(modbus_db) as session:
|
|
fetched = session.get(ModbusReading, reading_id)
|
|
assert fetched is not None
|
|
assert fetched.recorded_at is not None
|
|
assert fetched.payload == sample_payload
|
|
assert fetched.payload["voltage"] == 230.2
|
|
assert fetched.payload["frequency"] == 50.0
|
|
|
|
|
|
def test_modbus_reading_restrict_prevents_device_deletion(tmp_path: Path):
|
|
"""Deleting a device with readings must fail due to ON DELETE RESTRICT.
|
|
|
|
SQLite only enforces FK constraints when ``PRAGMA foreign_keys = ON`` is set
|
|
on the connection, so we build a dedicated engine that enables this pragma.
|
|
"""
|
|
import sqlalchemy.exc
|
|
from sqlalchemy import event as sa_event
|
|
|
|
db_path = tmp_path / "restrict_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})
|
|
|
|
# Enable FK enforcement for every connection on this engine.
|
|
@sa_event.listens_for(engine, "connect")
|
|
def _enable_fk(dbapi_conn, _rec):
|
|
cursor = dbapi_conn.cursor()
|
|
cursor.execute("PRAGMA foreign_keys = ON")
|
|
cursor.close()
|
|
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(engine) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="Restricted Device",
|
|
host="10.0.0.1",
|
|
profile="sdm120",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.flush()
|
|
|
|
reading = ModbusReading(
|
|
device_id=device.id,
|
|
recorded_at=now,
|
|
payload={"voltage": 230.0},
|
|
)
|
|
session.add(reading)
|
|
session.commit()
|
|
device_id = device.id
|
|
|
|
# Attempting to delete the device should raise an IntegrityError.
|
|
with pytest.raises(sqlalchemy.exc.IntegrityError):
|
|
with Session(engine) as session:
|
|
session.execute(
|
|
text("DELETE FROM modbus_device WHERE id = :did"),
|
|
{"did": device_id},
|
|
)
|
|
session.commit()
|
|
|
|
engine.dispose()
|
|
|
|
|
|
def test_modbus_reading_restrict_enforced_via_app_engine(tmp_path: Path):
|
|
"""FK RESTRICT is enforced by the *default* app engine (db.py) without any manual pragma.
|
|
|
|
After enabling ``PRAGMA foreign_keys=ON`` in the db.py connect event, the app
|
|
engine must enforce ON DELETE RESTRICT on modbus_reading.device_id without
|
|
callers having to set the pragma themselves. This test uses the real app engine
|
|
obtained via ``app.db.get_engine`` (routed through the lru_cache, just like the
|
|
running app) so it exercises the exact production code path.
|
|
"""
|
|
import sqlalchemy.exc
|
|
|
|
from app.db import _get_engine, reset_db_caches
|
|
|
|
db_path = tmp_path / "app_engine_restrict_test.db"
|
|
db_url = f"sqlite:///{db_path}"
|
|
|
|
alembic_cfg = _make_app_alembic_config(db_url)
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
# Obtain engine through the same lru_cache path that production uses.
|
|
# Clear caches first so we get a fresh engine pointing at our tmp DB.
|
|
reset_db_caches()
|
|
engine = _get_engine(db_url)
|
|
|
|
try:
|
|
now = datetime.now(tz=timezone.utc)
|
|
with Session(engine) as session:
|
|
device = ModbusDevice(
|
|
friendly_name="App-Engine RESTRICT Test",
|
|
host="10.0.0.2",
|
|
profile="sdm120",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
session.add(device)
|
|
session.flush()
|
|
|
|
reading = ModbusReading(
|
|
device_id=device.id,
|
|
recorded_at=now,
|
|
payload={"voltage": 230.0},
|
|
)
|
|
session.add(reading)
|
|
session.commit()
|
|
device_id = device.id
|
|
|
|
# Without manual pragma, the app engine (db.py) must still enforce RESTRICT.
|
|
with pytest.raises(sqlalchemy.exc.IntegrityError):
|
|
with Session(engine) as session:
|
|
session.execute(
|
|
text("DELETE FROM modbus_device WHERE id = :did"),
|
|
{"did": device_id},
|
|
)
|
|
session.commit()
|
|
finally:
|
|
reset_db_caches()
|
|
engine.dispose()
|