M5-T05: add Modbus JSON API (device CRUD, readings, metrics, test)
This commit is contained in:
@@ -0,0 +1,778 @@
|
||||
"""Tests for M5-T05: Modbus JSON API (app/api/routes/api/modbus.py).
|
||||
|
||||
Coverage:
|
||||
- GET /api/modbus/profiles — list profiles
|
||||
- GET /api/modbus/devices — list devices
|
||||
- POST /api/modbus/devices — create device (profile validation, auth/CSRF)
|
||||
- GET /api/modbus/devices/{uuid} — get single device
|
||||
- PATCH /api/modbus/devices/{uuid} — update device (enable/disable, profile validation)
|
||||
- DELETE /api/modbus/devices/{uuid} — delete with 409 guard (app-layer, not FK)
|
||||
- GET /api/modbus/devices/{uuid}/latest — latest reading (found / not-found)
|
||||
- GET /api/modbus/devices/{uuid}/readings — time-range + limit cap
|
||||
- GET /api/modbus/devices/{uuid}/metrics — profile metric catalogue
|
||||
- POST /api/modbus/devices/{uuid}/test — test-read (mocked driver, not persisted)
|
||||
|
||||
Auth/CSRF matrix:
|
||||
- Unauthenticated read → 401
|
||||
- Authenticated write without CSRF → 403
|
||||
- Authenticated read → 200/201/204
|
||||
- Authenticated write + CSRF → 200/201/204
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime, timedelta
|
||||
from typing import Any
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
from sqlalchemy import create_engine
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from app.models.modbus import ModbusDevice, ModbusReading
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 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.status_code} {resp.text}"
|
||||
|
||||
|
||||
def _create_device_payload(**overrides) -> dict[str, Any]:
|
||||
base: dict[str, Any] = {
|
||||
"friendly_name": "Test Meter",
|
||||
"host": "10.0.0.1",
|
||||
"port": 502,
|
||||
"unit_id": 1,
|
||||
"profile": "sdm120",
|
||||
"poll_interval_s": 5,
|
||||
"enabled": True,
|
||||
}
|
||||
base.update(overrides)
|
||||
return base
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Fixtures
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def modbus_client(auth_database):
|
||||
"""TestClient + engine for Modbus API tests."""
|
||||
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()
|
||||
|
||||
|
||||
def _make_device(engine, **kwargs) -> ModbusDevice:
|
||||
"""Insert a ModbusDevice row directly and return it."""
|
||||
now = datetime.now(UTC)
|
||||
with Session(engine) as session:
|
||||
device = ModbusDevice(
|
||||
friendly_name=kwargs.get("friendly_name", "Test Meter"),
|
||||
host=kwargs.get("host", "10.0.0.1"),
|
||||
port=kwargs.get("port", 502),
|
||||
unit_id=kwargs.get("unit_id", 1),
|
||||
profile=kwargs.get("profile", "sdm120"),
|
||||
poll_interval_s=kwargs.get("poll_interval_s", 5),
|
||||
enabled=kwargs.get("enabled", True),
|
||||
created_at=now,
|
||||
updated_at=now,
|
||||
)
|
||||
session.add(device)
|
||||
session.commit()
|
||||
session.refresh(device)
|
||||
# Detach from this session so caller can inspect plain attributes.
|
||||
device_id = device.id
|
||||
device_uuid = device.uuid
|
||||
device_friendly_name = device.friendly_name
|
||||
device_host = device.host
|
||||
device_profile = device.profile
|
||||
|
||||
# Return a simple namespace-like object (uuid is what we mostly need).
|
||||
class _DeviceStub:
|
||||
id = device_id
|
||||
uuid = device_uuid
|
||||
friendly_name = device_friendly_name
|
||||
host = device_host
|
||||
profile = device_profile
|
||||
|
||||
return _DeviceStub()
|
||||
|
||||
|
||||
def _make_reading(engine, device_id: int, recorded_at: datetime, payload: dict) -> None:
|
||||
"""Insert a ModbusReading row directly."""
|
||||
with Session(engine) as session:
|
||||
reading = ModbusReading(
|
||||
device_id=device_id,
|
||||
recorded_at=recorded_at,
|
||||
payload=payload,
|
||||
)
|
||||
session.add(reading)
|
||||
session.commit()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/modbus/profiles
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_profiles_unauthenticated_returns_401(modbus_client):
|
||||
client, _ = modbus_client
|
||||
resp = client.get("/api/modbus/profiles")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_profiles_returns_list(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.get("/api/modbus/profiles")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert "profiles" in body
|
||||
assert isinstance(body["profiles"], list)
|
||||
# sdm120 profile must be present
|
||||
names = [p["name"] for p in body["profiles"]]
|
||||
assert "sdm120" in names
|
||||
# Each profile has name + description
|
||||
for p in body["profiles"]:
|
||||
assert "name" in p
|
||||
assert "description" in p
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/modbus/devices
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_list_devices_unauthenticated_returns_401(modbus_client):
|
||||
client, _ = modbus_client
|
||||
resp = client.get("/api/modbus/devices")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_list_devices_empty(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.get("/api/modbus/devices")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["items"] == []
|
||||
assert body["total"] == 0
|
||||
|
||||
|
||||
def test_list_devices_returns_created_device(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.get("/api/modbus/devices")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["total"] == 1
|
||||
assert body["items"][0]["uuid"] == device.uuid
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/modbus/devices
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_create_device_unauthenticated_returns_401(modbus_client):
|
||||
client, _ = modbus_client
|
||||
resp = client.post(
|
||||
"/api/modbus/devices",
|
||||
json=_create_device_payload(),
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_create_device_missing_csrf_returns_403(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.post("/api/modbus/devices", json=_create_device_payload())
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_create_device_success(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
resp = client.post(
|
||||
"/api/modbus/devices",
|
||||
json=_create_device_payload(),
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 201
|
||||
body = resp.json()
|
||||
assert body["friendly_name"] == "Test Meter"
|
||||
assert "uuid" in body
|
||||
assert len(body["uuid"]) == 36 # uuid4 format
|
||||
|
||||
# Verify row exists in DB
|
||||
with Session(engine) as session:
|
||||
devices = session.query(ModbusDevice).all()
|
||||
assert len(devices) == 1
|
||||
|
||||
|
||||
def test_create_device_invalid_profile_returns_422(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.post(
|
||||
"/api/modbus/devices",
|
||||
json=_create_device_payload(profile="nonexistent_profile_xyz"),
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
assert "nonexistent_profile_xyz" in resp.json()["detail"]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/modbus/devices/{uuid}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_get_device_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_get_device_not_found_returns_404(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.get("/api/modbus/devices/00000000-0000-0000-0000-000000000000")
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
def test_get_device_returns_correct_fields(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine, friendly_name="My Meter")
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["uuid"] == device.uuid
|
||||
assert body["friendly_name"] == "My Meter"
|
||||
assert body["profile"] == "sdm120"
|
||||
assert "last_poll_at" in body
|
||||
assert "last_poll_ok" in body
|
||||
assert "created_at" in body
|
||||
assert "updated_at" in body
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# PATCH /api/modbus/devices/{uuid}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_patch_device_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.patch(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
json={"enabled": False},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_patch_device_missing_csrf_returns_403(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.patch(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
json={"enabled": False},
|
||||
)
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_patch_device_enable_disable(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine, enabled=True)
|
||||
|
||||
# Disable
|
||||
resp = client.patch(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
json={"enabled": False},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["enabled"] is False
|
||||
|
||||
# Re-enable
|
||||
resp = client.patch(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
json={"enabled": True},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["enabled"] is True
|
||||
|
||||
|
||||
def test_patch_device_invalid_profile_returns_422(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.patch(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
json={"profile": "nonexistent_xyz"},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
def test_patch_device_updates_correct_fields(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.patch(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
json={"friendly_name": "Renamed Meter", "port": 503},
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["friendly_name"] == "Renamed Meter"
|
||||
assert body["port"] == 503
|
||||
# Other fields unchanged
|
||||
assert body["uuid"] == device.uuid
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# DELETE /api/modbus/devices/{uuid}
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_delete_device_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.delete(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_delete_device_missing_csrf_returns_403(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.delete(f"/api/modbus/devices/{device.uuid}")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_delete_device_no_readings_succeeds(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.delete(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 204
|
||||
|
||||
# Confirm device is gone
|
||||
with Session(engine) as session:
|
||||
devices = session.query(ModbusDevice).all()
|
||||
assert len(devices) == 0
|
||||
|
||||
|
||||
def test_delete_device_with_readings_returns_409(modbus_client):
|
||||
"""Delete with existing readings must return 409 (application-layer check, not FK)."""
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
# Add a reading
|
||||
_make_reading(
|
||||
engine,
|
||||
device_id=device.id,
|
||||
recorded_at=datetime.now(UTC),
|
||||
payload={"voltage": 230.0},
|
||||
)
|
||||
|
||||
resp = client.delete(
|
||||
f"/api/modbus/devices/{device.uuid}",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 409
|
||||
assert "reading" in resp.json()["detail"].lower()
|
||||
|
||||
# Device still exists
|
||||
with Session(engine) as session:
|
||||
devices = session.query(ModbusDevice).all()
|
||||
assert len(devices) == 1
|
||||
|
||||
|
||||
def test_delete_device_not_found_returns_404(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.delete(
|
||||
"/api/modbus/devices/00000000-0000-0000-0000-000000000000",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/modbus/devices/{uuid}/latest
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_latest_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/latest")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_latest_no_readings_returns_found_false(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/latest")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["found"] is False
|
||||
assert body["recorded_at"] is None
|
||||
assert body["payload"] is None
|
||||
|
||||
|
||||
def test_latest_returns_most_recent_reading(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
earlier = now - timedelta(minutes=10)
|
||||
later = now - timedelta(minutes=1)
|
||||
|
||||
_make_reading(engine, device.id, earlier, {"voltage": 228.0})
|
||||
_make_reading(engine, device.id, later, {"voltage": 232.0})
|
||||
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/latest")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["found"] is True
|
||||
assert body["payload"] == {"voltage": 232.0}
|
||||
assert body["recorded_at"] is not None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/modbus/devices/{uuid}/readings
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_readings_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/readings")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_readings_empty_returns_empty_list(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/readings")
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["items"] == []
|
||||
|
||||
|
||||
def test_readings_returns_all_by_default_ordered_ascending(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
t1 = now - timedelta(minutes=30)
|
||||
t2 = now - timedelta(minutes=20)
|
||||
t3 = now - timedelta(minutes=10)
|
||||
|
||||
_make_reading(engine, device.id, t3, {"voltage": 233.0})
|
||||
_make_reading(engine, device.id, t1, {"voltage": 231.0})
|
||||
_make_reading(engine, device.id, t2, {"voltage": 232.0})
|
||||
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/readings")
|
||||
assert resp.status_code == 200
|
||||
items = resp.json()["items"]
|
||||
assert len(items) == 3
|
||||
# Ascending order
|
||||
voltages = [item["payload"]["voltage"] for item in items]
|
||||
assert voltages == [231.0, 232.0, 233.0]
|
||||
|
||||
|
||||
def test_readings_time_window_filters_correctly(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
t_old = now - timedelta(hours=2)
|
||||
t_mid = now - timedelta(hours=1)
|
||||
t_new = now - timedelta(minutes=5)
|
||||
|
||||
_make_reading(engine, device.id, t_old, {"voltage": 228.0})
|
||||
_make_reading(engine, device.id, t_mid, {"voltage": 230.0})
|
||||
_make_reading(engine, device.id, t_new, {"voltage": 232.0})
|
||||
|
||||
# Only the last 90 minutes
|
||||
start = (now - timedelta(minutes=90)).isoformat()
|
||||
resp = client.get(
|
||||
f"/api/modbus/devices/{device.uuid}/readings",
|
||||
params={"start": start},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
items = resp.json()["items"]
|
||||
assert len(items) == 2
|
||||
voltages = {item["payload"]["voltage"] for item in items}
|
||||
assert 228.0 not in voltages
|
||||
|
||||
|
||||
def test_readings_limit_applied(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
now = datetime.now(UTC)
|
||||
for i in range(10):
|
||||
_make_reading(engine, device.id, now - timedelta(minutes=i), {"voltage": float(i)})
|
||||
|
||||
resp = client.get(
|
||||
f"/api/modbus/devices/{device.uuid}/readings",
|
||||
params={"limit": 3},
|
||||
)
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()["items"]) == 3
|
||||
|
||||
|
||||
def test_readings_limit_exceeds_max_returns_422(modbus_client):
|
||||
"""A limit > 5000 must be rejected by FastAPI query validation (422)."""
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.get(
|
||||
f"/api/modbus/devices/{device.uuid}/readings",
|
||||
params={"limit": 9999},
|
||||
)
|
||||
assert resp.status_code == 422
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# GET /api/modbus/devices/{uuid}/metrics
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_metrics_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/metrics")
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_metrics_returns_profile_catalogue(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine, profile="sdm120")
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/metrics")
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["profile"] == "sdm120"
|
||||
assert isinstance(body["metrics"], list)
|
||||
assert len(body["metrics"]) > 0
|
||||
# Each metric has required fields
|
||||
for m in body["metrics"]:
|
||||
assert "key" in m
|
||||
assert "label" in m
|
||||
assert "unit" in m
|
||||
assert "device_class" in m
|
||||
|
||||
# SDM120 must include voltage
|
||||
keys = [m["key"] for m in body["metrics"]]
|
||||
assert "voltage" in keys
|
||||
|
||||
# Label is human-readable (not raw snake_case for "active_power")
|
||||
active_power = next((m for m in body["metrics"] if m["key"] == "active_power"), None)
|
||||
assert active_power is not None
|
||||
assert "Active Power" == active_power["label"]
|
||||
|
||||
|
||||
def test_metrics_label_derived_from_key(modbus_client):
|
||||
"""Verify the label derivation helper converts snake_case to Title Case."""
|
||||
from app.api.routes.api.modbus import _label_from_key
|
||||
|
||||
assert _label_from_key("voltage") == "Voltage"
|
||||
assert _label_from_key("active_power") == "Active Power"
|
||||
assert _label_from_key("import_energy") == "Import Energy"
|
||||
assert _label_from_key("power_factor") == "Power Factor"
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# POST /api/modbus/devices/{uuid}/test
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_test_read_unauthenticated_returns_401(modbus_client):
|
||||
client, engine = modbus_client
|
||||
device = _make_device(engine)
|
||||
resp = client.post(
|
||||
f"/api/modbus/devices/{device.uuid}/test",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 401
|
||||
|
||||
|
||||
def test_test_read_missing_csrf_returns_403(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
resp = client.post(f"/api/modbus/devices/{device.uuid}/test")
|
||||
assert resp.status_code == 403
|
||||
|
||||
|
||||
def test_test_read_success_returns_payload_and_does_not_persist(modbus_client):
|
||||
"""Successful test-read returns payload and stores nothing in modbus_reading."""
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
known_payload = {"voltage": 230.2, "current": 1.3}
|
||||
|
||||
with (
|
||||
patch("app.api.routes.api.modbus.modbus_driver.read_blocks") as mock_read,
|
||||
patch("app.api.routes.api.modbus.decode_profile") as mock_decode,
|
||||
):
|
||||
mock_read.return_value = {0x0000: 0x4366, 0x0001: 0x3334}
|
||||
mock_decode.return_value = known_payload
|
||||
|
||||
resp = client.post(
|
||||
f"/api/modbus/devices/{device.uuid}/test",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["ok"] is True
|
||||
assert body["payload"] == known_payload
|
||||
assert body["error"] is None
|
||||
|
||||
# Verify nothing was persisted
|
||||
with Session(engine) as session:
|
||||
readings = session.query(ModbusReading).all()
|
||||
assert len(readings) == 0
|
||||
|
||||
|
||||
def test_test_read_driver_failure_returns_ok_false(modbus_client):
|
||||
"""When the driver raises, test-read returns ok=False with an error message."""
|
||||
from app.integrations.modbus.driver import ModbusConnectionError
|
||||
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
|
||||
with patch("app.api.routes.api.modbus.modbus_driver.read_blocks") as mock_read:
|
||||
mock_read.side_effect = ModbusConnectionError("Cannot reach gateway")
|
||||
|
||||
resp = client.post(
|
||||
f"/api/modbus/devices/{device.uuid}/test",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert body["ok"] is False
|
||||
assert body["error"] is not None
|
||||
assert "Cannot reach gateway" in body["error"]
|
||||
|
||||
# Still no readings in DB
|
||||
with Session(engine) as session:
|
||||
readings = session.query(ModbusReading).all()
|
||||
assert len(readings) == 0
|
||||
|
||||
|
||||
def test_test_read_not_found_returns_404(modbus_client):
|
||||
client, _ = modbus_client
|
||||
_login(client)
|
||||
resp = client.post(
|
||||
"/api/modbus/devices/00000000-0000-0000-0000-000000000000/test",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp.status_code == 404
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Row count integrity: CRUD operations produce the right row count
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_crud_row_counts(modbus_client):
|
||||
"""Create 2 devices, verify row count; delete one (no readings), verify count drops."""
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
|
||||
resp1 = client.post(
|
||||
"/api/modbus/devices",
|
||||
json=_create_device_payload(friendly_name="Meter A"),
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp1.status_code == 201
|
||||
uuid_a = resp1.json()["uuid"]
|
||||
|
||||
resp2 = client.post(
|
||||
"/api/modbus/devices",
|
||||
json=_create_device_payload(friendly_name="Meter B"),
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp2.status_code == 201
|
||||
|
||||
# 2 devices exist
|
||||
with Session(engine) as session:
|
||||
count = session.query(ModbusDevice).count()
|
||||
assert count == 2
|
||||
|
||||
# Delete Meter A (no readings)
|
||||
resp_del = client.delete(
|
||||
f"/api/modbus/devices/{uuid_a}",
|
||||
headers={"X-CSRF-Token": _CSRF},
|
||||
)
|
||||
assert resp_del.status_code == 204
|
||||
|
||||
# 1 device remains
|
||||
with Session(engine) as session:
|
||||
count = session.query(ModbusDevice).count()
|
||||
assert count == 1
|
||||
|
||||
|
||||
def test_readings_each_item_has_recorded_at_and_payload(modbus_client):
|
||||
client, engine = modbus_client
|
||||
_login(client)
|
||||
device = _make_device(engine)
|
||||
now = datetime.now(UTC)
|
||||
_make_reading(engine, device.id, now, {"voltage": 230.0, "current": 1.5})
|
||||
|
||||
resp = client.get(f"/api/modbus/devices/{device.uuid}/readings")
|
||||
assert resp.status_code == 200
|
||||
items = resp.json()["items"]
|
||||
assert len(items) == 1
|
||||
item = items[0]
|
||||
assert "recorded_at" in item
|
||||
assert "payload" in item
|
||||
assert item["payload"]["voltage"] == 230.0
|
||||
Reference in New Issue
Block a user