alexei.dolgolyov
907bdaf043
TestWLEDSchemeInference in test_devices_routes covers the POST/PUT create-and-update flow with a stubbed WLED provider so the infer_http_scheme integration hop has end-to-end coverage instead of just the unit tests. test_url_scheme grows public IPv6 (Cloudflare / Google / Quad9 DNS), bracketed-form, and ULA cases. Adds an explicit pin for the Python ipaddress documentation-prefix quirk (2001:db8::/32 is is_private, so it routes to http:// even though some audits colloquially call it "public").
507 lines
18 KiB
Python
507 lines
18 KiB
Python
"""Tests for device CRUD routes.
|
|
|
|
These tests exercise the FastAPI route handlers using dependency override
|
|
to inject test stores, avoiding real hardware dependencies.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from ledgrab.api.routes.devices import router
|
|
from ledgrab.storage.device_store import Device, DeviceStore
|
|
from ledgrab.storage.output_target_store import OutputTargetStore
|
|
from ledgrab.core.processing.processor_manager import ProcessorManager
|
|
from ledgrab.api import dependencies as deps
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App + fixtures (isolated from the real main app)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_app():
|
|
"""Build a minimal FastAPI app with just the devices router + overrides."""
|
|
app = FastAPI()
|
|
app.include_router(router)
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def _route_db(tmp_path):
|
|
from ledgrab.storage.database import Database
|
|
|
|
db = Database(tmp_path / "test.db")
|
|
yield db
|
|
db.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def device_store(_route_db):
|
|
return DeviceStore(_route_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def output_target_store(_route_db):
|
|
return OutputTargetStore(_route_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def processor_manager():
|
|
"""A mock ProcessorManager — avoids real hardware."""
|
|
m = MagicMock(spec=ProcessorManager)
|
|
m.add_device = MagicMock()
|
|
m.remove_device = AsyncMock()
|
|
m.update_device_info = MagicMock()
|
|
m.find_device_state = MagicMock(return_value=None)
|
|
m.get_all_device_health_dicts = MagicMock(return_value=[])
|
|
return m
|
|
|
|
|
|
@pytest.fixture
|
|
def client(device_store, output_target_store, processor_manager):
|
|
app = _make_app()
|
|
|
|
# Override auth to always pass
|
|
from ledgrab.api.auth import verify_api_key
|
|
|
|
app.dependency_overrides[verify_api_key] = lambda: "test-user"
|
|
|
|
# Override stores and manager
|
|
app.dependency_overrides[deps.get_device_store] = lambda: device_store
|
|
app.dependency_overrides[deps.get_output_target_store] = lambda: output_target_store
|
|
app.dependency_overrides[deps.get_processor_manager] = lambda: processor_manager
|
|
|
|
return TestClient(app, raise_server_exceptions=False)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper to pre-populate a device
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _seed_device(store: DeviceStore, name="Test Device", led_count=100) -> Device:
|
|
return store.create_device(
|
|
name=name,
|
|
url="http://192.168.1.100",
|
|
led_count=led_count,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# LIST
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestListDevices:
|
|
def test_list_empty(self, client):
|
|
resp = client.get("/api/v1/devices")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["count"] == 0
|
|
assert data["devices"] == []
|
|
|
|
def test_list_with_devices(self, client, device_store):
|
|
_seed_device(device_store, "Dev A")
|
|
_seed_device(device_store, "Dev B")
|
|
resp = client.get("/api/v1/devices")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["count"] == 2
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET by ID
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestGetDevice:
|
|
def test_get_existing(self, client, device_store):
|
|
d = _seed_device(device_store)
|
|
resp = client.get(f"/api/v1/devices/{d.id}")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["id"] == d.id
|
|
assert data["name"] == "Test Device"
|
|
|
|
def test_get_not_found(self, client):
|
|
resp = client.get("/api/v1/devices/nonexistent")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# UPDATE
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestUpdateDevice:
|
|
def test_update_name(self, client, device_store):
|
|
d = _seed_device(device_store)
|
|
resp = client.put(
|
|
f"/api/v1/devices/{d.id}",
|
|
json={"name": "Renamed"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "Renamed"
|
|
|
|
def test_update_led_count(self, client, device_store):
|
|
d = _seed_device(device_store, led_count=100)
|
|
resp = client.put(
|
|
f"/api/v1/devices/{d.id}",
|
|
json={"led_count": 300},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["led_count"] == 300
|
|
|
|
def test_update_not_found(self, client):
|
|
resp = client.put(
|
|
"/api/v1/devices/missing_id",
|
|
json={"name": "X"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DELETE
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestDeleteDevice:
|
|
def test_delete_existing(self, client, device_store):
|
|
d = _seed_device(device_store)
|
|
resp = client.delete(f"/api/v1/devices/{d.id}")
|
|
assert resp.status_code == 204
|
|
assert device_store.count() == 0
|
|
|
|
def test_delete_not_found(self, client):
|
|
resp = client.delete("/api/v1/devices/missing_id")
|
|
assert resp.status_code == 404
|
|
|
|
def test_delete_referenced_by_target_returns_409(
|
|
self, client, device_store, output_target_store
|
|
):
|
|
d = _seed_device(device_store)
|
|
output_target_store.create_target(
|
|
name="Target",
|
|
target_type="led",
|
|
device_id=d.id,
|
|
)
|
|
resp = client.delete(f"/api/v1/devices/{d.id}")
|
|
assert resp.status_code == 409
|
|
assert "referenced" in resp.json()["detail"].lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Batch states
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBatchStates:
|
|
def test_batch_states(self, client):
|
|
resp = client.get("/api/v1/devices/batch/states")
|
|
assert resp.status_code == 200
|
|
assert "states" in resp.json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# PAIRING
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _PairableProviderStub:
|
|
"""Test double that exercises the four pair_device outcomes.
|
|
|
|
Registered into the provider registry at test-time and unregistered
|
|
afterwards so the global registry stays clean across tests.
|
|
"""
|
|
|
|
def __init__(self, outcome: str):
|
|
self._outcome = outcome
|
|
|
|
@property
|
|
def device_type(self) -> str:
|
|
return "_pair_test"
|
|
|
|
@property
|
|
def capabilities(self) -> set:
|
|
return {"requires_pairing"}
|
|
|
|
def create_client(self, config, *, deps): # pragma: no cover -- not used here
|
|
raise AssertionError("Stub provider should not be asked to create a client")
|
|
|
|
async def check_health(self, url, http_client, prev_health=None): # pragma: no cover
|
|
raise AssertionError("Stub provider should not be asked for health")
|
|
|
|
async def validate_device(self, url): # pragma: no cover
|
|
return {}
|
|
|
|
async def pair_device(self, url: str):
|
|
from ledgrab.core.devices.led_client import PairingNotReady
|
|
|
|
if self._outcome == "success":
|
|
return {"_pair_test_token": "token-abc", "_pair_test_meta": 42}
|
|
if self._outcome == "not_ready":
|
|
raise PairingNotReady("Press the device button and try again.")
|
|
if self._outcome == "invalid_url":
|
|
raise ValueError("URL is missing a host")
|
|
if self._outcome == "boom":
|
|
raise RuntimeError("transport blew up")
|
|
if self._outcome == "malformed":
|
|
return "not-a-dict" # type: ignore[return-value]
|
|
if self._outcome == "not_implemented":
|
|
raise NotImplementedError("paired? what is paired?")
|
|
raise AssertionError(f"unknown outcome: {self._outcome}")
|
|
|
|
|
|
@pytest.fixture
|
|
def pair_stub_registered():
|
|
"""Register a test provider for the duration of one test."""
|
|
from ledgrab.core.devices import led_client as _led_client
|
|
|
|
registered: dict = {}
|
|
|
|
def _register(outcome: str):
|
|
provider = _PairableProviderStub(outcome)
|
|
_led_client.register_provider(provider) # type: ignore[arg-type]
|
|
registered["type"] = provider.device_type
|
|
return provider
|
|
|
|
yield _register
|
|
|
|
if "type" in registered:
|
|
_led_client._provider_registry.pop(registered["type"], None)
|
|
|
|
|
|
class TestPairDevice:
|
|
def test_returns_provider_fields_on_success(self, client, pair_stub_registered):
|
|
pair_stub_registered("success")
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "_pair_test", "url": "test://host"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert resp.json() == {"fields": {"_pair_test_token": "token-abc", "_pair_test_meta": 42}}
|
|
|
|
def test_unknown_device_type_returns_400(self, client):
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "nonexistent_zzz", "url": "x://y"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "unknown device type" in resp.json()["detail"].lower()
|
|
|
|
def test_not_implemented_returns_400(self, client, pair_stub_registered):
|
|
pair_stub_registered("not_implemented")
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "_pair_test", "url": "test://host"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "does not support pairing" in resp.json()["detail"]
|
|
|
|
def test_pairing_not_ready_returns_409(self, client, pair_stub_registered):
|
|
pair_stub_registered("not_ready")
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "_pair_test", "url": "test://host"},
|
|
)
|
|
assert resp.status_code == 409
|
|
assert "press" in resp.json()["detail"].lower()
|
|
|
|
def test_invalid_url_returns_422(self, client, pair_stub_registered):
|
|
pair_stub_registered("invalid_url")
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "_pair_test", "url": "test://host"},
|
|
)
|
|
assert resp.status_code == 422
|
|
assert "host" in resp.json()["detail"]
|
|
|
|
def test_transport_exception_returns_502(self, client, pair_stub_registered):
|
|
pair_stub_registered("boom")
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "_pair_test", "url": "test://host"},
|
|
)
|
|
assert resp.status_code == 502
|
|
assert "pairing failed" in resp.json()["detail"].lower()
|
|
|
|
def test_malformed_provider_result_returns_500(self, client, pair_stub_registered):
|
|
pair_stub_registered("malformed")
|
|
resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "_pair_test", "url": "test://host"},
|
|
)
|
|
assert resp.status_code == 500
|
|
assert "malformed" in resp.json()["detail"].lower()
|
|
|
|
def test_missing_required_fields_returns_422(self, client):
|
|
resp = client.post("/api/v1/devices/pair", json={"device_type": "nanoleaf"})
|
|
assert resp.status_code == 422
|
|
|
|
|
|
class TestWLEDSchemeInference:
|
|
"""End-to-end pin for the URL-scheme inference applied to WLED devices.
|
|
|
|
The :func:`infer_http_scheme` helper has its own exhaustive unit
|
|
coverage in :file:`tests/test_url_scheme.py`; this class adds the
|
|
*integration* hop that REVIEW_TODO calls out — POST/PUT a bare host
|
|
and assert the stored device carries the inferred scheme so a future
|
|
refactor of the route handler can't quietly drop the call.
|
|
|
|
The WLED provider is stubbed out so the test does not need a real
|
|
device on the network — we just need ``validate_device`` to return a
|
|
successful payload.
|
|
"""
|
|
|
|
@pytest.fixture
|
|
def _stub_wled_validate(self, monkeypatch):
|
|
async def fake_validate(self, url): # noqa: ARG001 — provider self
|
|
return {"led_count": 30}
|
|
|
|
from ledgrab.core.devices.wled_provider import WLEDDeviceProvider
|
|
|
|
monkeypatch.setattr(WLEDDeviceProvider, "validate_device", fake_validate)
|
|
return fake_validate
|
|
|
|
def test_create_wled_with_bare_private_ip_normalises_to_http(
|
|
self, client, device_store, _stub_wled_validate
|
|
):
|
|
resp = client.post(
|
|
"/api/v1/devices",
|
|
json={
|
|
"name": "WLED desk",
|
|
"device_type": "wled",
|
|
"url": "192.168.1.42",
|
|
"led_count": 30,
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
device_id = resp.json()["id"]
|
|
assert device_store.get_device(device_id).url == "http://192.168.1.42"
|
|
|
|
def test_create_wled_with_public_host_normalises_to_https(
|
|
self, client, device_store, _stub_wled_validate
|
|
):
|
|
resp = client.post(
|
|
"/api/v1/devices",
|
|
json={
|
|
"name": "WLED cloud",
|
|
"device_type": "wled",
|
|
"url": "wled.example.com",
|
|
"led_count": 30,
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
device_id = resp.json()["id"]
|
|
assert device_store.get_device(device_id).url == "https://wled.example.com"
|
|
|
|
def test_create_wled_strips_trailing_slash_then_infers(
|
|
self, client, device_store, _stub_wled_validate
|
|
):
|
|
resp = client.post(
|
|
"/api/v1/devices",
|
|
json={
|
|
"name": "WLED rack",
|
|
"device_type": "wled",
|
|
"url": "wled-rack.local/",
|
|
"led_count": 30,
|
|
},
|
|
)
|
|
assert resp.status_code == 201, resp.text
|
|
device_id = resp.json()["id"]
|
|
assert device_store.get_device(device_id).url == "http://wled-rack.local"
|
|
|
|
def test_update_wled_with_bare_host_normalises_url(self, client, device_store):
|
|
existing = device_store.create_device(
|
|
name="WLED desk",
|
|
url="http://192.168.1.42",
|
|
led_count=30,
|
|
device_type="wled",
|
|
)
|
|
resp = client.put(
|
|
f"/api/v1/devices/{existing.id}",
|
|
json={"url": "10.0.0.5"},
|
|
)
|
|
assert resp.status_code == 200, resp.text
|
|
assert device_store.get_device(existing.id).url == "http://10.0.0.5"
|
|
|
|
|
|
class TestPairThenCreateFlow:
|
|
"""End-to-end coverage: pair, then persist; assert the token is
|
|
encrypted at rest and decrypted in to_config(), and that the API
|
|
response strips the secret.
|
|
|
|
Closes the LOW gap in the pre-merge review: pair-route tests stopped
|
|
at the 200 response, never carrying the returned fields through to
|
|
storage to verify the round-trip and the response-strip.
|
|
"""
|
|
|
|
def test_pair_then_create_persists_encrypted_token(self, client, device_store):
|
|
from ledgrab.api.routes.devices import _device_to_response
|
|
from ledgrab.core.devices import led_client as _led_client
|
|
from ledgrab.core.devices.led_client import DeviceHealth
|
|
|
|
class _NanoleafLikeStub:
|
|
@property
|
|
def device_type(self):
|
|
return "nanoleaf_like_stub"
|
|
|
|
@property
|
|
def capabilities(self):
|
|
return {"manual_led_count", "requires_pairing"}
|
|
|
|
async def pair_device(self, url):
|
|
return {"nanoleaf_token": "secret-paired-token"}
|
|
|
|
async def validate_device(self, url):
|
|
return {"led_count": 9}
|
|
|
|
def create_client(self, config, *, deps):
|
|
raise AssertionError("not used")
|
|
|
|
async def check_health(self, url, http_client, prev_health=None):
|
|
return DeviceHealth(online=True)
|
|
|
|
_led_client.register_provider(_NanoleafLikeStub())
|
|
try:
|
|
# Step 1: pair via the route
|
|
pair_resp = client.post(
|
|
"/api/v1/devices/pair",
|
|
json={"device_type": "nanoleaf_like_stub", "url": "stub://1.2.3.4"},
|
|
)
|
|
assert pair_resp.status_code == 200, pair_resp.text
|
|
fields = pair_resp.json()["fields"]
|
|
assert fields == {"nanoleaf_token": "secret-paired-token"}
|
|
|
|
# Step 2: persist via the store (skip the route's create path
|
|
# which would require a real validate_device handshake)
|
|
device = device_store.create_device(
|
|
name="E2E Paired",
|
|
url="stub://1.2.3.4",
|
|
led_count=9,
|
|
device_type="nanoleaf",
|
|
**fields,
|
|
)
|
|
|
|
# In-memory device holds plaintext
|
|
assert device.nanoleaf_token == "secret-paired-token"
|
|
|
|
# to_config surfaces plaintext to the provider
|
|
config = device.to_config()
|
|
assert config.nanoleaf_token == "secret-paired-token"
|
|
|
|
# Persisted row holds ciphertext (envelope prefix)
|
|
persisted = device.to_dict()
|
|
assert persisted["nanoleaf_token"].startswith("ENC:v1:")
|
|
assert persisted["nanoleaf_token"] != "secret-paired-token"
|
|
|
|
# API response strips the token; only a boolean flag remains
|
|
resp = _device_to_response(device).model_dump()
|
|
assert "nanoleaf_token" not in resp
|
|
assert resp.get("nanoleaf_paired") is True
|
|
finally:
|
|
_led_client._provider_registry.pop("nanoleaf_like_stub", None)
|