478 lines
17 KiB
Python
478 lines
17 KiB
Python
"""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
|