M5-T12: add expose toggle API and Home Assistant Expose settings panel
This commit is contained in:
@@ -0,0 +1,477 @@
|
||||
"""Tests for M5-T12: Expose API (app/api/routes/api/expose.py).
|
||||
|
||||
Coverage:
|
||||
- GET /api/expose — catalog + toggle state + MQTT/Discovery status
|
||||
- PUT /api/expose — upsert toggles; triggers discovery re-publish
|
||||
- POST /api/expose/republish — manual re-publish
|
||||
|
||||
Auth/CSRF matrix:
|
||||
- Unauthenticated GET → 401
|
||||
- Authenticated GET (no CSRF needed) → 200
|
||||
- Authenticated PUT without CSRF → 403
|
||||
- Authenticated PUT + CSRF → 200; toggles persisted; publish_discovery called
|
||||
- Authenticated POST /republish without CSRF → 403
|
||||
- Authenticated POST /republish + CSRF + MQTT off → 200 ok=False
|
||||
- Authenticated POST /republish + CSRF + MQTT on → 200 ok=True; publish_discovery called
|
||||
|
||||
Value-getter non-serialisation:
|
||||
- value_getter field is NOT present in the API response payload.
|
||||
|
||||
Isolation strategy:
|
||||
- MqttManager and publish_discovery are always mocked so no real broker is needed.
|
||||
- The Modbus _modbus_provider is patched away to a simple fake provider for catalog
|
||||
tests, avoiding the need for DB-registered modbus devices in every test.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Shared helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_CSRF = "test-csrf-token"
|
||||
|
||||
|
||||
def _login(client: TestClient) -> None:
|
||||
resp = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "admin", "password": "test-password"},
|
||||
)
|
||||
assert resp.status_code == 200, f"Login failed: {resp.text}"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# A minimal fake provider that returns a fixed entity (no real DB device needed)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _make_fake_provider(entity_key: str = "modbus.abc.voltage"):
|
||||
"""Return a provider callable that yields one hard-coded ExposableEntity."""
|
||||
|
||||
def _provider(session): # noqa: ANN001
|
||||
from app.integrations.expose import DeviceInfo, ExposableEntity
|
||||
|
||||
return [
|
||||
ExposableEntity(
|
||||
key=entity_key,
|
||||
component="sensor",
|
||||
device=DeviceInfo(identifiers=("modbus", "abc"), name="Test Meter"),
|
||||
device_class="voltage",
|
||||
unit="V",
|
||||
name="Test Meter Voltage",
|
||||
value_getter=lambda sess: 230.2, # non-serialisable callable
|
||||
state_class="measurement",
|
||||
)
|
||||
]
|
||||
|
||||
return _provider
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def expose_client(auth_database):
|
||||
"""TestClient + engine for Expose API tests, with real DB but no real broker."""
|
||||
from app.main import create_app
|
||||
|
||||
app_url = auth_database["app_url"]
|
||||
engine = create_engine(app_url, connect_args={"check_same_thread": False})
|
||||
|
||||
fastapi_app = create_app()
|
||||
with TestClient(fastapi_app) as test_client:
|
||||
yield test_client, engine
|
||||
|
||||
engine.dispose()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers to insert/read toggle rows directly
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _count_toggles(engine) -> int:
|
||||
with Session(engine) as session:
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
return session.query(ExposedEntityToggle).count()
|
||||
|
||||
|
||||
def _get_toggle(engine, key: str) -> bool | None:
|
||||
with Session(engine) as session:
|
||||
from app.models.expose import ExposedEntityToggle
|
||||
|
||||
row = session.query(ExposedEntityToggle).filter(ExposedEntityToggle.key == key).first()
|
||||
return None if row is None else row.enabled
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: GET /api/expose
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestGetExpose:
|
||||
def test_unauthenticated_returns_401(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
resp = client.get("/api/expose")
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_authenticated_returns_catalog_and_status(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
fake_key = "modbus.abc.voltage"
|
||||
with patch(
|
||||
"app.integrations.expose._REGISTRY",
|
||||
[_make_fake_provider(fake_key)],
|
||||
):
|
||||
resp = client.get("/api/expose")
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
|
||||
# Must have 'catalog' list and 'mqtt_status' dict
|
||||
assert "catalog" in data
|
||||
assert "mqtt_status" in data
|
||||
|
||||
# Default toggle is False (no row in DB)
|
||||
assert len(data["catalog"]) == 1
|
||||
entry = data["catalog"][0]
|
||||
assert entry["entity"]["key"] == fake_key
|
||||
assert entry["enabled"] is False
|
||||
|
||||
# value_getter must NOT appear in the serialised entity
|
||||
assert "value_getter" not in entry["entity"]
|
||||
|
||||
def test_mqtt_status_fields_present(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", []):
|
||||
resp = client.get("/api/expose")
|
||||
|
||||
assert resp.status_code == 200
|
||||
status_data = resp.json()["mqtt_status"]
|
||||
assert "mqtt_configured" in status_data
|
||||
assert "mqtt_connected" in status_data
|
||||
assert "discovery_enabled" in status_data
|
||||
|
||||
def test_entity_fields_correct(self, expose_client):
|
||||
"""Verify the entity schema shape matches ExposableEntitySchema."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
with patch(
|
||||
"app.integrations.expose._REGISTRY",
|
||||
[_make_fake_provider("modbus.abc.voltage")],
|
||||
):
|
||||
resp = client.get("/api/expose")
|
||||
|
||||
entity = resp.json()["catalog"][0]["entity"]
|
||||
for field in ("key", "component", "device", "device_class", "unit", "name"):
|
||||
assert field in entity, f"Missing field: {field}"
|
||||
# device sub-fields
|
||||
assert "identifiers" in entity["device"]
|
||||
assert "name" in entity["device"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: PUT /api/expose
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPutExpose:
|
||||
def test_unauthenticated_returns_401(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
resp = client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_missing_csrf_returns_403(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
resp = client.put("/api/expose", json={"toggles": {}})
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_toggles_persisted_in_db(self, expose_client):
|
||||
client, engine = expose_client
|
||||
_login(client)
|
||||
|
||||
fake_key = "modbus.abc.voltage"
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", [_make_fake_provider(fake_key)]):
|
||||
with patch("app.api.routes.api.expose._trigger_republish"):
|
||||
resp = client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {fake_key: True}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
# Row inserted
|
||||
assert _get_toggle(engine, fake_key) is True
|
||||
|
||||
def test_toggle_upsert(self, expose_client):
|
||||
"""Second PUT on same key updates the existing row rather than inserting another."""
|
||||
client, engine = expose_client
|
||||
_login(client)
|
||||
|
||||
fake_key = "modbus.abc.voltage"
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", [_make_fake_provider(fake_key)]):
|
||||
with patch("app.api.routes.api.expose._trigger_republish"):
|
||||
# Enable
|
||||
client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {fake_key: True}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
# Disable (upsert)
|
||||
resp = client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {fake_key: False}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
# Still exactly one row
|
||||
assert _count_toggles(engine) == 1
|
||||
assert _get_toggle(engine, fake_key) is False
|
||||
|
||||
def test_toggle_change_triggers_discovery(self, expose_client):
|
||||
"""PUT on toggles calls _trigger_republish (which calls publish_discovery)."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
fake_key = "modbus.abc.voltage"
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", [_make_fake_provider(fake_key)]):
|
||||
with patch(
|
||||
"app.api.routes.api.expose._trigger_republish"
|
||||
) as mock_republish:
|
||||
resp = client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {fake_key: True}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
mock_republish.assert_called_once()
|
||||
|
||||
def test_response_contains_updated_catalog(self, expose_client):
|
||||
"""PUT response reflects the newly set toggle state."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
fake_key = "modbus.abc.voltage"
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", [_make_fake_provider(fake_key)]):
|
||||
with patch("app.api.routes.api.expose._trigger_republish"):
|
||||
resp = client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {fake_key: True}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
data = resp.json()
|
||||
assert "catalog" in data
|
||||
# The entry for fake_key should now be enabled=True
|
||||
entries = {e["entity"]["key"]: e["enabled"] for e in data["catalog"]}
|
||||
assert entries[fake_key] is True
|
||||
|
||||
def test_value_getter_not_in_put_response(self, expose_client):
|
||||
"""value_getter callable must not appear in the PUT response."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
fake_key = "modbus.abc.voltage"
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", [_make_fake_provider(fake_key)]):
|
||||
with patch("app.api.routes.api.expose._trigger_republish"):
|
||||
resp = client.put(
|
||||
"/api/expose",
|
||||
json={"toggles": {fake_key: True}},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
entity = resp.json()["catalog"][0]["entity"]
|
||||
assert "value_getter" not in entity
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: POST /api/expose/republish
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPostRepublish:
|
||||
def test_unauthenticated_returns_401(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
resp = client.post(
|
||||
"/api/expose/republish",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
def test_missing_csrf_returns_403(self, expose_client):
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
resp = client.post("/api/expose/republish")
|
||||
assert resp.status_code == 403
|
||||
|
||||
def test_republish_when_mqtt_not_configured(self, expose_client):
|
||||
"""When MQTT is not enabled/connected, republish returns ok=False."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
# Ensure MQTT is not configured (default in test env)
|
||||
with patch.object(
|
||||
type(
|
||||
__import__(
|
||||
"app.integrations.mqtt", fromlist=["mqtt_manager"]
|
||||
).mqtt_manager
|
||||
),
|
||||
"is_connected",
|
||||
new_callable=lambda: property(lambda self: False),
|
||||
):
|
||||
resp = client.post(
|
||||
"/api/expose/republish",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["ok"] is False
|
||||
assert "message" in data
|
||||
|
||||
def test_republish_calls_publish_discovery_when_mqtt_on(self, expose_client):
|
||||
"""When MQTT is active, republish calls publish_discovery."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
# Patch settings to enable MQTT + discovery
|
||||
mock_settings: Any = MagicMock()
|
||||
mock_settings.mqtt_enabled = True
|
||||
mock_settings.ha_discovery_enabled = True
|
||||
mock_settings.ha_discovery_prefix = "homeassistant"
|
||||
|
||||
with patch("app.api.routes.api.expose.get_settings", return_value=mock_settings):
|
||||
with patch("app.api.routes.api.expose.mqtt_manager") as mock_mqtt:
|
||||
mock_mqtt.is_configured.return_value = True
|
||||
mock_mqtt.is_connected = True
|
||||
with patch(
|
||||
"app.api.routes.api.expose._trigger_republish"
|
||||
) as mock_republish:
|
||||
resp = client.post(
|
||||
"/api/expose/republish",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
data = resp.json()
|
||||
assert data["ok"] is True
|
||||
mock_republish.assert_called_once()
|
||||
|
||||
def test_trigger_republish_calls_publish_discovery(self):
|
||||
"""Unit test: _trigger_republish calls publish_discovery with the session."""
|
||||
from app.api.routes.api.expose import _trigger_republish
|
||||
|
||||
mock_session: Any = MagicMock()
|
||||
|
||||
# publish_discovery is imported inside _trigger_republish; patch it
|
||||
# at the source module so all code paths see the mock.
|
||||
with patch("app.services.ha_discovery.publish_discovery") as mock_pd:
|
||||
_trigger_republish(mock_session)
|
||||
mock_pd.assert_called_once_with(mock_session)
|
||||
|
||||
def test_republish_mqtt_not_connected_returns_false(self, expose_client):
|
||||
"""Republish when MQTT not connected returns ok=False (not an exception)."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
# Ensure settings say MQTT is enabled but mqtt_manager says not connected
|
||||
mock_settings: Any = MagicMock()
|
||||
mock_settings.mqtt_enabled = True
|
||||
mock_settings.ha_discovery_enabled = True
|
||||
|
||||
with patch("app.api.routes.api.expose.get_settings", return_value=mock_settings):
|
||||
with patch("app.api.routes.api.expose.mqtt_manager") as mock_mqtt:
|
||||
mock_mqtt.is_connected = False
|
||||
resp = client.post(
|
||||
"/api/expose/republish",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["ok"] is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tests: device grouping in catalog
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCatalogGrouping:
|
||||
def test_catalog_groups_entities_by_device(self, expose_client):
|
||||
"""Multiple entities from the same device share the same device info."""
|
||||
client, _engine = expose_client
|
||||
_login(client)
|
||||
|
||||
def _multi_entity_provider(session):
|
||||
from app.integrations.expose import DeviceInfo, ExposableEntity
|
||||
|
||||
device_info = DeviceInfo(identifiers=("modbus", "uuid-1"), name="My Meter")
|
||||
return [
|
||||
ExposableEntity(
|
||||
key="modbus.uuid-1.voltage",
|
||||
component="sensor",
|
||||
device=device_info,
|
||||
device_class="voltage",
|
||||
unit="V",
|
||||
name="My Meter Voltage",
|
||||
),
|
||||
ExposableEntity(
|
||||
key="modbus.uuid-1.current",
|
||||
component="sensor",
|
||||
device=device_info,
|
||||
device_class="current",
|
||||
unit="A",
|
||||
name="My Meter Current",
|
||||
),
|
||||
ExposableEntity(
|
||||
key="modbus.uuid-1.online",
|
||||
component="binary_sensor",
|
||||
device=device_info,
|
||||
device_class="connectivity",
|
||||
unit="",
|
||||
name="My Meter Online",
|
||||
),
|
||||
]
|
||||
|
||||
with patch("app.integrations.expose._REGISTRY", [_multi_entity_provider]):
|
||||
resp = client.get("/api/expose")
|
||||
|
||||
assert resp.status_code == 200
|
||||
catalog = resp.json()["catalog"]
|
||||
assert len(catalog) == 3
|
||||
|
||||
# All entities share the same device name
|
||||
device_names = {e["entity"]["device"]["name"] for e in catalog}
|
||||
assert device_names == {"My Meter"}
|
||||
|
||||
# binary_sensor is present (not-sensor component)
|
||||
components = {e["entity"]["component"] for e in catalog}
|
||||
assert "binary_sensor" in components
|
||||
assert "sensor" in components
|
||||
Reference in New Issue
Block a user