M5-T09: add exposed_entities table and ExposableEntity provider framework
This commit is contained in:
@@ -0,0 +1,681 @@
|
||||
"""Tests for M5-T09: exposed_entities table + ExposableEntity/provider framework.
|
||||
|
||||
Covers:
|
||||
1. ExposableEntity dataclass: fields, key stability.
|
||||
2. Provider registry: register_provider, get_providers.
|
||||
3. build_catalog: entity enumeration, device grouping, metadata from profile,
|
||||
toggle defaults (enabled=False), binary_sensor presence.
|
||||
4. Migration shape: upgrade creates the table + unique index; downgrade drops it.
|
||||
5. ORM round-trip: toggle insert/read/update; default enabled=False.
|
||||
6. APP_BASELINE_REVISION matches the new Alembic head.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
import pytest
|
||||
from alembic import command
|
||||
from alembic.config import Config
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
if TYPE_CHECKING:
|
||||
pass
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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 _make_modbus_device(session: Session, *, friendly_name: str, uuid: str) -> object:
|
||||
"""Insert a ModbusDevice row into a test DB session and return it."""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.models.modbus import ModbusDevice
|
||||
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
device = ModbusDevice(
|
||||
uuid=uuid,
|
||||
friendly_name=friendly_name,
|
||||
host="192.168.1.1",
|
||||
port=502,
|
||||
unit_id=1,
|
||||
profile="sdm120",
|
||||
poll_interval_s=5,
|
||||
enabled=True,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
session.add(device)
|
||||
session.flush()
|
||||
return device
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def expose_db(tmp_path: Path):
|
||||
"""Temporary SQLite DB upgraded to the current Alembic head (includes new toggle table)."""
|
||||
db_path = tmp_path / "expose_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. ExposableEntity / DeviceInfo dataclass tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_exposable_entity_fields():
|
||||
"""ExposableEntity must have all required fields with correct defaults."""
|
||||
from app.integrations.expose import DeviceInfo, ExposableEntity
|
||||
|
||||
device = DeviceInfo(identifiers=("modbus", "abc-uuid"), name="Test Device")
|
||||
entity = ExposableEntity(
|
||||
key="modbus.abc-uuid.voltage",
|
||||
component="sensor",
|
||||
device=device,
|
||||
device_class="voltage",
|
||||
unit="V",
|
||||
name="Test Device Voltage",
|
||||
)
|
||||
|
||||
assert entity.key == "modbus.abc-uuid.voltage"
|
||||
assert entity.component == "sensor"
|
||||
assert entity.device.identifiers == ("modbus", "abc-uuid")
|
||||
assert entity.device.name == "Test Device"
|
||||
assert entity.device_class == "voltage"
|
||||
assert entity.unit == "V"
|
||||
assert entity.name == "Test Device Voltage"
|
||||
assert entity.value_getter is None # default
|
||||
assert entity.state_class is None # default
|
||||
|
||||
|
||||
def test_exposable_entity_with_value_getter():
|
||||
"""ExposableEntity.value_getter stores and calls the provided callable."""
|
||||
from app.integrations.expose import DeviceInfo, ExposableEntity
|
||||
|
||||
device = DeviceInfo(identifiers=("modbus", "dev1"), name="D1")
|
||||
getter = lambda: 230.5 # noqa: E731
|
||||
entity = ExposableEntity(
|
||||
key="modbus.dev1.voltage",
|
||||
component="sensor",
|
||||
device=device,
|
||||
device_class="voltage",
|
||||
unit="V",
|
||||
name="D1 Voltage",
|
||||
value_getter=getter,
|
||||
)
|
||||
assert entity.value_getter is not None
|
||||
assert entity.value_getter() == 230.5
|
||||
|
||||
|
||||
def test_entity_key_stability_uses_uuid_not_id():
|
||||
"""Keys must be derived from uuid (stable), never from auto-increment DB ids."""
|
||||
from app.integrations.expose import DeviceInfo, ExposableEntity
|
||||
|
||||
uuid_val = "550e8400-e29b-41d4-a716-446655440000"
|
||||
key = f"modbus.{uuid_val}.voltage"
|
||||
device = DeviceInfo(identifiers=("modbus", uuid_val), name="Stable Device")
|
||||
entity = ExposableEntity(
|
||||
key=key,
|
||||
component="sensor",
|
||||
device=device,
|
||||
device_class="voltage",
|
||||
unit="V",
|
||||
name="Stable Device Voltage",
|
||||
)
|
||||
# Key is uuid-based: it must not contain any integer-only segment that
|
||||
# looks like an auto-increment id (i.e., it contains the full uuid string).
|
||||
assert uuid_val in entity.key
|
||||
assert "modbus" in entity.key
|
||||
assert "voltage" in entity.key
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 2. Provider registry tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_register_provider_as_decorator():
|
||||
"""register_provider used as a decorator must add the callable to the registry."""
|
||||
from app.integrations.expose import get_providers, register_provider
|
||||
|
||||
before = len(get_providers())
|
||||
|
||||
@register_provider
|
||||
def _dummy_provider(session):
|
||||
return []
|
||||
|
||||
after_providers = get_providers()
|
||||
assert len(after_providers) == before + 1
|
||||
assert _dummy_provider in after_providers
|
||||
|
||||
# Clean up: remove the dummy from the module-level registry.
|
||||
from app.integrations import expose as _expose_mod
|
||||
_expose_mod._REGISTRY.remove(_dummy_provider)
|
||||
|
||||
|
||||
def test_register_provider_direct_call():
|
||||
"""register_provider called directly must also register the callable."""
|
||||
from app.integrations.expose import get_providers, register_provider
|
||||
from app.integrations import expose as _expose_mod
|
||||
|
||||
def _another_dummy(session):
|
||||
return []
|
||||
|
||||
before = len(get_providers())
|
||||
returned = register_provider(_another_dummy)
|
||||
assert returned is _another_dummy # returns the provider unchanged
|
||||
assert len(get_providers()) == before + 1
|
||||
|
||||
# Clean up.
|
||||
_expose_mod._REGISTRY.remove(_another_dummy)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 3. build_catalog tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_build_catalog_empty_with_no_devices(expose_db):
|
||||
"""build_catalog with no enabled devices must return an empty list."""
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
# The modbus provider will find no enabled devices; other providers may
|
||||
# add entries only if registered. Since we only register the modbus
|
||||
# provider at module load, the result should be empty.
|
||||
assert catalog == []
|
||||
|
||||
|
||||
def test_build_catalog_produces_entities_for_enabled_device(expose_db):
|
||||
"""build_catalog must yield entities (including binary_sensor) for each enabled device."""
|
||||
from app.integrations.expose import build_catalog, CatalogEntry
|
||||
|
||||
# Insert a test device.
|
||||
test_uuid = "aaaaaaaa-0000-0000-0000-000000000001"
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="SDM120 Test", uuid=test_uuid)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
assert len(catalog) > 0, "Expected at least one entity for the enabled device"
|
||||
|
||||
keys = [entry.entity.key for entry in catalog]
|
||||
|
||||
# Must contain a 'modbus.<uuid>.online' binary_sensor.
|
||||
online_key = f"modbus.{test_uuid}.online"
|
||||
assert online_key in keys, f"Expected online entity key {online_key!r} in catalog"
|
||||
|
||||
# Must contain at least one sensor entity (voltage is always in sdm120 profile).
|
||||
voltage_key = f"modbus.{test_uuid}.voltage"
|
||||
assert voltage_key in keys, f"Expected voltage entity key {voltage_key!r} in catalog"
|
||||
|
||||
# Verify CatalogEntry type.
|
||||
assert all(isinstance(e, CatalogEntry) for e in catalog)
|
||||
|
||||
|
||||
def test_build_catalog_default_enabled_false(expose_db):
|
||||
"""Entities with no toggle row must default to enabled=False."""
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
test_uuid = "aaaaaaaa-0000-0000-0000-000000000002"
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="SDM120 B", uuid=test_uuid)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
# No toggle rows inserted → all must be disabled.
|
||||
assert all(not entry.enabled for entry in catalog), (
|
||||
"All catalog entries must default to enabled=False when no toggle row exists"
|
||||
)
|
||||
|
||||
|
||||
def test_build_catalog_respects_toggle_enabled(expose_db):
|
||||
"""An entity with a toggle row enabled=True must appear as enabled in the catalog."""
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from app.integrations.expose import build_catalog
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
test_uuid = "aaaaaaaa-0000-0000-0000-000000000003"
|
||||
voltage_key = f"modbus.{test_uuid}.voltage"
|
||||
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="SDM120 C", uuid=test_uuid)
|
||||
# Enable the voltage entity explicitly.
|
||||
toggle = ExposedEntityToggle(
|
||||
key=voltage_key,
|
||||
enabled=True,
|
||||
updated_at=datetime.now(tz=timezone.utc),
|
||||
)
|
||||
session.add(toggle)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
enabled_keys = {entry.entity.key for entry in catalog if entry.enabled}
|
||||
assert voltage_key in enabled_keys, (
|
||||
f"Expected {voltage_key!r} to be enabled after toggling"
|
||||
)
|
||||
|
||||
# Other entities (no toggle row) must still default to disabled.
|
||||
online_key = f"modbus.{test_uuid}.online"
|
||||
disabled_keys = {entry.entity.key for entry in catalog if not entry.enabled}
|
||||
assert online_key in disabled_keys, f"Expected {online_key!r} to remain disabled"
|
||||
|
||||
|
||||
def test_build_catalog_device_grouping(expose_db):
|
||||
"""All entities for a device must share the same DeviceInfo (device grouping)."""
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
test_uuid = "aaaaaaaa-0000-0000-0000-000000000004"
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="SDM120 D", uuid=test_uuid)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
device_entities = [e for e in catalog if test_uuid in e.entity.key]
|
||||
assert len(device_entities) > 0
|
||||
|
||||
# All entities for this device must use the same device identifiers tuple.
|
||||
identifiers_set = {e.entity.device.identifiers for e in device_entities}
|
||||
assert len(identifiers_set) == 1, (
|
||||
"All entities for one device must share the same DeviceInfo identifiers"
|
||||
)
|
||||
assert identifiers_set.pop() == ("modbus", test_uuid)
|
||||
|
||||
|
||||
def test_build_catalog_metric_metadata_from_profile(expose_db):
|
||||
"""Entity device_class and unit must match the sdm120 profile's metrics."""
|
||||
from app.integrations.expose import build_catalog
|
||||
from app.integrations.modbus.profiles import load_profile
|
||||
|
||||
test_uuid = "aaaaaaaa-0000-0000-0000-000000000005"
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="SDM120 E", uuid=test_uuid)
|
||||
session.commit()
|
||||
|
||||
profile = load_profile("sdm120")
|
||||
metric_map = {m.key: m for m in profile.metrics}
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
sensor_entities = [
|
||||
e for e in catalog
|
||||
if e.entity.component == "sensor" and test_uuid in e.entity.key
|
||||
]
|
||||
|
||||
for entry in sensor_entities:
|
||||
# Extract metric key from entity key: "modbus.<uuid>.<metric_key>"
|
||||
_, _, metric_key = entry.entity.key.split(".", 2)
|
||||
if metric_key in metric_map:
|
||||
metric_spec = metric_map[metric_key]
|
||||
assert entry.entity.device_class == (metric_spec.device_class or None), (
|
||||
f"device_class mismatch for {metric_key!r}: "
|
||||
f"expected {metric_spec.device_class!r}, "
|
||||
f"got {entry.entity.device_class!r}"
|
||||
)
|
||||
assert entry.entity.unit == metric_spec.unit, (
|
||||
f"unit mismatch for {metric_key!r}: "
|
||||
f"expected {metric_spec.unit!r}, got {entry.entity.unit!r}"
|
||||
)
|
||||
assert entry.entity.component == metric_spec.ha_component, (
|
||||
f"component mismatch for {metric_key!r}: "
|
||||
f"expected {metric_spec.ha_component!r}, "
|
||||
f"got {entry.entity.component!r}"
|
||||
)
|
||||
|
||||
|
||||
def test_build_catalog_online_binary_sensor_present(expose_db):
|
||||
"""Each enabled device must produce exactly one binary_sensor 'online' entity."""
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
uuid1 = "aaaaaaaa-0000-0000-0000-000000000006"
|
||||
uuid2 = "aaaaaaaa-0000-0000-0000-000000000007"
|
||||
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="Device F1", uuid=uuid1)
|
||||
_make_modbus_device(session, friendly_name="Device F2", uuid=uuid2)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
binary_sensors = [e for e in catalog if e.entity.component == "binary_sensor"]
|
||||
binary_sensor_keys = {e.entity.key for e in binary_sensors}
|
||||
|
||||
assert f"modbus.{uuid1}.online" in binary_sensor_keys
|
||||
assert f"modbus.{uuid2}.online" in binary_sensor_keys
|
||||
# No sensor must masquerade as binary_sensor and vice versa.
|
||||
assert all("online" in e.entity.key for e in binary_sensors), (
|
||||
"All binary_sensor entities should be 'online' entities"
|
||||
)
|
||||
|
||||
|
||||
def test_build_catalog_has_non_sensor_component(expose_db):
|
||||
"""The catalog must contain at least one entity with component != 'sensor'."""
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
test_uuid = "aaaaaaaa-0000-0000-0000-000000000008"
|
||||
with Session(expose_db) as session:
|
||||
_make_modbus_device(session, friendly_name="SDM120 G", uuid=test_uuid)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
non_sensor = [e for e in catalog if e.entity.component != "sensor"]
|
||||
assert len(non_sensor) >= 1, (
|
||||
"Expected at least one non-sensor component (binary_sensor 'online')"
|
||||
)
|
||||
|
||||
|
||||
def test_build_catalog_only_includes_enabled_devices(expose_db):
|
||||
"""Disabled devices must not contribute entities to the catalog."""
|
||||
from app.models.modbus import ModbusDevice
|
||||
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
enabled_uuid = "aaaaaaaa-0000-0000-0000-000000000009"
|
||||
disabled_uuid = "aaaaaaaa-0000-0000-0000-00000000000a"
|
||||
|
||||
with Session(expose_db) as session:
|
||||
enabled_device = ModbusDevice(
|
||||
uuid=enabled_uuid,
|
||||
friendly_name="Enabled Device",
|
||||
host="10.0.0.1",
|
||||
port=502,
|
||||
unit_id=1,
|
||||
profile="sdm120",
|
||||
poll_interval_s=5,
|
||||
enabled=True,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
disabled_device = ModbusDevice(
|
||||
uuid=disabled_uuid,
|
||||
friendly_name="Disabled Device",
|
||||
host="10.0.0.2",
|
||||
port=502,
|
||||
unit_id=2,
|
||||
profile="sdm120",
|
||||
poll_interval_s=5,
|
||||
enabled=False,
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
session.add_all([enabled_device, disabled_device])
|
||||
session.commit()
|
||||
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
all_keys = {e.entity.key for e in catalog}
|
||||
assert any(enabled_uuid in k for k in all_keys), "Enabled device must be in catalog"
|
||||
assert not any(disabled_uuid in k for k in all_keys), (
|
||||
"Disabled device must NOT be in catalog"
|
||||
)
|
||||
|
||||
|
||||
def test_build_catalog_key_uses_uuid_not_db_id(expose_db):
|
||||
"""Entity keys must contain the device uuid, not the integer DB primary key."""
|
||||
from app.integrations.expose import build_catalog
|
||||
|
||||
test_uuid = "bbbbbbbb-1111-0000-0000-000000000001"
|
||||
with Session(expose_db) as session:
|
||||
device = _make_modbus_device(session, friendly_name="UUID Key Test", uuid=test_uuid)
|
||||
session.commit()
|
||||
db_id = device.id # type: ignore[union-attr]
|
||||
|
||||
with Session(expose_db) as session:
|
||||
catalog = build_catalog(session)
|
||||
|
||||
entity_keys = [e.entity.key for e in catalog if test_uuid in e.entity.key]
|
||||
assert len(entity_keys) > 0, "Expected entities for the test device"
|
||||
|
||||
# Keys must contain the uuid string and not be keyed by integer id only.
|
||||
for key in entity_keys:
|
||||
assert test_uuid in key, f"Key {key!r} must contain the uuid {test_uuid!r}"
|
||||
# Ensure the integer DB id does not appear where the uuid should be.
|
||||
parts = key.split(".")
|
||||
assert str(db_id) not in parts[1:2], (
|
||||
f"Key {key!r} must not use the integer DB id {db_id} in the uuid slot"
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 4. Migration shape tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_exposed_entity_toggle_table_exists_after_upgrade(expose_db):
|
||||
"""exposed_entity_toggle table must exist after upgrade to head."""
|
||||
inspector = inspect(expose_db)
|
||||
assert "exposed_entity_toggle" in inspector.get_table_names(), (
|
||||
"exposed_entity_toggle table missing after upgrade to head"
|
||||
)
|
||||
|
||||
|
||||
def test_exposed_entity_toggle_columns(expose_db):
|
||||
"""exposed_entity_toggle must have id, key, enabled, updated_at columns."""
|
||||
inspector = inspect(expose_db)
|
||||
columns = {col["name"]: col for col in inspector.get_columns("exposed_entity_toggle")}
|
||||
|
||||
assert "id" in columns
|
||||
assert "key" in columns
|
||||
assert "enabled" in columns
|
||||
assert "updated_at" in columns
|
||||
|
||||
# id must be non-nullable.
|
||||
assert not columns["id"]["nullable"]
|
||||
# key must be non-nullable.
|
||||
assert not columns["key"]["nullable"]
|
||||
# enabled must be non-nullable.
|
||||
assert not columns["enabled"]["nullable"]
|
||||
|
||||
|
||||
def test_exposed_entity_toggle_key_unique_constraint(expose_db):
|
||||
"""exposed_entity_toggle.key must have a unique constraint."""
|
||||
inspector = inspect(expose_db)
|
||||
unique_constraints = inspector.get_unique_constraints("exposed_entity_toggle")
|
||||
unique_cols = [col for uc in unique_constraints for col in uc["column_names"]]
|
||||
assert "key" in unique_cols, "key column must have a unique constraint"
|
||||
|
||||
|
||||
def test_exposed_entity_toggle_key_unique_index(expose_db):
|
||||
"""exposed_entity_toggle must have a unique index on key."""
|
||||
inspector = inspect(expose_db)
|
||||
indexes = {idx["name"]: idx for idx in inspector.get_indexes("exposed_entity_toggle")}
|
||||
assert "ix_exposed_entity_toggle_key" in indexes, (
|
||||
"Index ix_exposed_entity_toggle_key missing"
|
||||
)
|
||||
assert indexes["ix_exposed_entity_toggle_key"]["unique"]
|
||||
|
||||
|
||||
def test_downgrade_removes_toggle_table(tmp_path: Path):
|
||||
"""downgrade -1 from head must drop the exposed_entity_toggle table cleanly."""
|
||||
db_path = tmp_path / "downgrade_expose_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 "exposed_entity_toggle" in inspector.get_table_names()
|
||||
engine.dispose()
|
||||
|
||||
# Downgrade one step removes the exposed_entity_toggle revision.
|
||||
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 "exposed_entity_toggle" not in table_names, (
|
||||
"exposed_entity_toggle should be gone after downgrade -1"
|
||||
)
|
||||
# Previous tables must still exist.
|
||||
assert "modbus_device" in table_names
|
||||
assert "modbus_reading" in table_names
|
||||
engine.dispose()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 5. ORM round-trip tests
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_expose_toggle_in_base_metadata():
|
||||
"""Base.metadata.tables must include exposed_entity_toggle."""
|
||||
from app.db import Base
|
||||
|
||||
# Importing the model registers it with Base.metadata.
|
||||
from app.models.expose import ExposedEntityToggle # noqa: F401
|
||||
|
||||
assert "exposed_entity_toggle" in Base.metadata.tables, (
|
||||
"exposed_entity_toggle not in Base.metadata after model import"
|
||||
)
|
||||
|
||||
|
||||
def test_expose_toggle_insert_and_read(expose_db):
|
||||
"""ExposedEntityToggle can be inserted and retrieved correctly."""
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
key = "modbus.test-uuid.voltage"
|
||||
|
||||
with Session(expose_db) as session:
|
||||
toggle = ExposedEntityToggle(key=key, enabled=True, updated_at=now)
|
||||
session.add(toggle)
|
||||
session.commit()
|
||||
toggle_id = toggle.id
|
||||
|
||||
with Session(expose_db) as session:
|
||||
fetched = session.get(ExposedEntityToggle, toggle_id)
|
||||
assert fetched is not None
|
||||
assert fetched.key == key
|
||||
assert fetched.enabled is True
|
||||
|
||||
|
||||
def test_expose_toggle_default_enabled_is_false(expose_db):
|
||||
"""ExposedEntityToggle.enabled must default to False after INSERT (column default).
|
||||
|
||||
SQLAlchemy applies ``default=False`` at INSERT time, not at Python object
|
||||
construction time. We verify the column default by inserting a row without
|
||||
providing ``enabled`` and reading it back.
|
||||
"""
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
|
||||
with Session(expose_db) as session:
|
||||
toggle = ExposedEntityToggle(key="default.enabled.test", updated_at=now)
|
||||
session.add(toggle)
|
||||
session.commit()
|
||||
toggle_id = toggle.id
|
||||
|
||||
with Session(expose_db) as session:
|
||||
fetched = session.get(ExposedEntityToggle, toggle_id)
|
||||
assert fetched is not None
|
||||
assert fetched.enabled is False, (
|
||||
"ExposedEntityToggle.enabled must default to False (not True, not None)"
|
||||
)
|
||||
|
||||
|
||||
def test_expose_toggle_key_uniqueness_enforced(expose_db):
|
||||
"""Inserting two rows with the same key must raise an IntegrityError."""
|
||||
import sqlalchemy.exc
|
||||
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
key = "modbus.dup-uuid.power"
|
||||
|
||||
with Session(expose_db) as session:
|
||||
session.add(ExposedEntityToggle(key=key, enabled=False, updated_at=now))
|
||||
session.commit()
|
||||
|
||||
with pytest.raises(sqlalchemy.exc.IntegrityError):
|
||||
with Session(expose_db) as session:
|
||||
session.add(ExposedEntityToggle(key=key, enabled=True, updated_at=now))
|
||||
session.commit()
|
||||
|
||||
|
||||
def test_expose_toggle_update(expose_db):
|
||||
"""Updating an existing toggle row must persist the new enabled value."""
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
now = datetime.now(tz=timezone.utc)
|
||||
key = "modbus.update-uuid.current"
|
||||
|
||||
with Session(expose_db) as session:
|
||||
toggle = ExposedEntityToggle(key=key, enabled=False, updated_at=now)
|
||||
session.add(toggle)
|
||||
session.commit()
|
||||
toggle_id = toggle.id
|
||||
|
||||
with Session(expose_db) as session:
|
||||
toggle = session.get(ExposedEntityToggle, toggle_id)
|
||||
toggle.enabled = True
|
||||
toggle.updated_at = datetime.now(tz=timezone.utc)
|
||||
session.commit()
|
||||
|
||||
with Session(expose_db) as session:
|
||||
fetched = session.get(ExposedEntityToggle, toggle_id)
|
||||
assert fetched.enabled is True
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 6. APP_BASELINE_REVISION
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_app_baseline_revision_matches_new_head(tmp_path: Path):
|
||||
"""APP_BASELINE_REVISION must equal the Alembic head after the new migration."""
|
||||
from alembic.script import ScriptDirectory
|
||||
|
||||
from scripts.app_db_adopt import APP_BASELINE_REVISION
|
||||
|
||||
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}"
|
||||
)
|
||||
# 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}"
|
||||
)
|
||||
@@ -144,7 +144,13 @@ def test_modbus_reading_recorded_at_index_exists(modbus_db):
|
||||
|
||||
|
||||
def test_downgrade_removes_modbus_tables(tmp_path: Path):
|
||||
"""downgrade -1 from head must drop both modbus tables cleanly."""
|
||||
"""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)
|
||||
@@ -157,8 +163,11 @@ def test_downgrade_removes_modbus_tables(tmp_path: Path):
|
||||
assert "modbus_reading" in inspector.get_table_names()
|
||||
engine.dispose()
|
||||
|
||||
# Downgrade one step removes the modbus migration.
|
||||
command.downgrade(alembic_cfg, "-1")
|
||||
# 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)
|
||||
|
||||
Reference in New Issue
Block a user