tliu93/home-automation
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 12 Wiki Activity
Files
3e04b15656fe95d8df90530d6534e962530065e0
home-automation/tests/test_expose_catalog.py
T

738 lines
27 KiB
Python
Raw Blame History

"""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.
value_getter now accepts a Session argument (updated in T11-rework-1).
"""
from app.integrations.expose import DeviceInfo, ExposableEntity
from sqlalchemy import create_engine
from sqlalchemy.orm import Session
device = DeviceInfo(identifiers=("modbus", "dev1"), name="D1")
# The value_getter receives a Session; use a lambda that ignores it.
getter = lambda _sess: 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
eng = create_engine("sqlite:///:memory:")
with Session(eng) as session:
assert entity.value_getter(session) == 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 modbus devices must contain no modbus entities.
FUE-T05: the energy_cost provider now requires an active electricity meter.
Without one, it returns [] and the catalog contains no energy entities.
With one, it produces 6 entities.
This test verifies both cases:
1. No modbus devices → no modbus entities.
2. No active meter → no energy entities (provider returns []).
3. After inserting an active meter → 6 energy entities present.
"""
from app.integrations.expose import build_catalog
# Case: no modbus devices, no active meter → catalog is empty.
with Session(expose_db) as session:
catalog = build_catalog(session)
modbus_entities = [e for e in catalog if e.entity.key.startswith("modbus.")]
assert modbus_entities == [], (
"Expected no modbus entities when no modbus devices are enabled"
)
# No active electricity meter → energy-cost provider returns [] → no energy entities.
energy_keys_no_meter = {e.entity.key for e in catalog if e.entity.key.startswith("energy.")}
assert len(energy_keys_no_meter) == 0, (
f"Expected 0 energy_cost entities (no active meter), got {energy_keys_no_meter!r}"
)
# Insert an active electricity meter → provider should now produce 6 entities.
from datetime import datetime, timezone
now = datetime.now(tz=timezone.utc)
with Session(expose_db) as session:
from app.models.energy import Meter
m = Meter(
label="Test Meter for catalog",
commodity="electricity",
started_at=now,
ended_at=None,
reason="initial",
note=None,
created_at=now,
)
session.add(m)
session.commit()
with Session(expose_db) as session:
catalog_with_meter = build_catalog(session)
energy_keys_with_meter = {
e.entity.key for e in catalog_with_meter if e.entity.key.startswith("energy.")
}
assert len(energy_keys_with_meter) == 6, (
f"Expected exactly 6 energy_cost entities (4 original + 2 daily) with active meter, "
f"got {energy_keys_with_meter!r}"
)
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):
"""Downgrading to the revision before exposed_entity_toggle must drop that table cleanly.
We downgrade to the explicit down_revision of the expose-toggle migration
(``20260622_09_modbus_tables``) rather than using ``-1`` from head, so the test
remains correct as new revisions are added on top. This is the same pattern
used in test_modbus_models.py for the modbus downgrade test.
"""
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 to the explicit down_revision of the exposed_entity_toggle migration,
# consistent with how test_modbus_models.py handles its own downgrade test.
command.downgrade(alembic_cfg, "20260622_09_modbus_tables")
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 to 20260622_09_modbus_tables"
)
# 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 revision."""
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}"
)
Reference in New Issue View Git Blame Copy Permalink