tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity

M5-T02: add modbus_device and modbus_reading tables and models

Browse Source
This commit is contained in:
tliu93
2026-06-22 12:45:50 +02:00
parent eac7860b51
commit d9c82e39fb
7 changed files with 598 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_modbus_models.py
+396
View File
@@ -0,0 +1,396 @@
"""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):
"""downgrade -1 from head must drop both modbus tables cleanly."""
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 one step removes the modbus migration.
command.downgrade(alembic_cfg, "-1")
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()
tests/test_totp_models.py
+4 -2
View File
@@ -262,8 +262,10 @@ def test_downgrade_removes_totp_columns_and_table(tmp_path: Path):
assert "totp_enabled" in auth_user_cols
engine.dispose()
# Downgrade one step (removes TOTP migration).
command.downgrade(alembic_cfg, "-1")
# Downgrade to the revision before TOTP (removes TOTP migration).
# We use an explicit target rather than "-1" so the test stays correct even
# as new revisions are added on top of the TOTP revision.
command.downgrade(alembic_cfg, "20260621_07_auth_login_throttle")
engine = create_engine(db_url, connect_args={"check_same_thread": False})
inspector = inspect(engine)