alexei.dolgolyov/ledgrab
1
1
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 14 Wiki Activity
Files
d32961085da07172db1e35a126480598a0c0170f
ledgrab/server/tests/api/routes/test_setup_routes.py
T
alexei.dolgolyov 6cd5e057da fix(setup): register scaffolded target with ProcessorManager + final-review hardening 
Final-review blocker: the setup scaffold created the LED output target in the
store but never registered it with the ProcessorManager, so the wizard's
"Start" step 404'd on a fresh setup (target not found) — the lights never
started despite a success screen. Now the scaffold calls
target.register_with_manager(manager) right after create (mirroring the
canonical POST /output-targets route, same ValueError guard), so
start_processing finds the target. Rollback unregisters via
manager.remove_target before deleting the store entity, so a post-registration
failure leaves no half-registered target.

Also from the final review:
- solve corner_indices elements now bounded ge=0 (clear 422 instead of silent
  modulo-wrap).
- setup-wizard.ts: reuse tutorials' suppressGettingStartedTour()/TOUR_KEY
  instead of a duplicated 'tour_completed' literal; drop a duplicate manual-form
  submit listener.

Tests: + adversarial pass over the whole feature (solver/session/scaffold edge
cases) and a scaffold->register->startable regression test. Full suite
2149 passed / 2 skipped; tsc clean; build passes; ruff clean.
2026-06-08 16:55:36 +03:00

601 lines
22 KiB
Python
Raw Blame History

"""Tests for the setup scaffold endpoint and the onboarding preference endpoints.
Coverage:
- scaffold happy path (device_id-based; 4 entities created, correct linking ids,
entity events fired ONLY after full success)
- scaffold reuses existing capture template
- scaffold partial-failure rollback (force a later step to fail → no orphans AND
no stray "created" events emitted for the rolled-back entities)
- scaffold 404 for unknown/missing device_id
- scaffold 422 for display_index out of range (> 63)
- scaffold 422 when device_id field is absent (Pydantic validation)
- onboarding GET default (onboarded=false, completed_at=null)
- onboarding PUT round-trip (timestamps auto-stamped)
- integration: scaffold → PUT calibration on the CSS → GET CSS round-trips with it
Deep adversarial coverage is deferred to the Phase 4 test-writer (Big Bang strategy).
"""
from __future__ import annotations
import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient
from unittest.mock import MagicMock, patch
# ---------------------------------------------------------------------------
# Shared fixtures
# ---------------------------------------------------------------------------
@pytest.fixture
def tmp_db(tmp_path):
from ledgrab.storage.database import Database
db = Database(tmp_path / "test_setup.db")
yield db
db.close()
@pytest.fixture
def device_store(tmp_db):
from ledgrab.storage import DeviceStore
return DeviceStore(tmp_db)
@pytest.fixture
def template_store(tmp_db):
from ledgrab.storage.template_store import TemplateStore
return TemplateStore(tmp_db)
@pytest.fixture
def picture_source_store(tmp_db):
from ledgrab.storage.picture_source_store import PictureSourceStore
return PictureSourceStore(tmp_db)
@pytest.fixture
def css_store(tmp_db):
from ledgrab.storage.color_strip_store import ColorStripStore
return ColorStripStore(tmp_db)
@pytest.fixture
def output_target_store(tmp_db):
from ledgrab.storage.output_target_store import OutputTargetStore
return OutputTargetStore(tmp_db)
@pytest.fixture
def sample_device(device_store):
return device_store.create_device(
name="Test LED Strip",
url="http://192.168.1.10",
led_count=60,
)
@pytest.fixture
def event_log():
"""Collect fire_entity_event calls for assertion."""
log = []
return log
@pytest.fixture
def mock_manager():
"""A MagicMock ProcessorManager that silently accepts all calls."""
mgr = MagicMock()
mgr.remove_target = MagicMock()
return mgr
@pytest.fixture
def setup_client(
tmp_db,
device_store,
template_store,
picture_source_store,
css_store,
output_target_store,
event_log,
mock_manager,
):
from ledgrab.api.routes.setup import router
from ledgrab.api.auth import verify_api_key
from ledgrab.api import dependencies as deps
app = FastAPI()
app.include_router(router)
app.dependency_overrides[verify_api_key] = lambda: "test"
app.dependency_overrides[deps.get_device_store] = lambda: device_store
app.dependency_overrides[deps.get_template_store] = lambda: template_store
app.dependency_overrides[deps.get_picture_source_store] = lambda: picture_source_store
app.dependency_overrides[deps.get_color_strip_store] = lambda: css_store
app.dependency_overrides[deps.get_output_target_store] = lambda: output_target_store
app.dependency_overrides[deps.get_processor_manager] = lambda: mock_manager
# Capture entity events
def _fire(entity_type, action, entity_id):
event_log.append((entity_type, action, entity_id))
with patch("ledgrab.api.routes.setup.fire_entity_event", side_effect=_fire):
yield TestClient(app, raise_server_exceptions=False)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _scaffold(client, **overrides):
"""POST a scaffold request with sensible defaults."""
body = {"display_index": 0, **overrides}
return client.post("/api/v1/setup/scaffold", json=body)
# ---------------------------------------------------------------------------
# Scaffold: happy path (using existing device)
# ---------------------------------------------------------------------------
class TestScaffoldHappyPath:
def test_returns_201(self, setup_client, sample_device, template_store):
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 201, resp.text
def test_response_contains_all_ids(self, setup_client, sample_device):
resp = _scaffold(setup_client, device_id=sample_device.id)
data = resp.json()
assert data["device_id"] == sample_device.id
assert data["capture_template_id"].startswith("tpl_")
assert data["picture_source_id"].startswith("ps_")
assert data["color_strip_source_id"].startswith("css_")
assert data["output_target_id"].startswith("pt_")
def test_response_has_no_device_created_field(self, setup_client, sample_device):
"""ScaffoldResponse no longer includes device_created — devices are always pre-existing."""
resp = _scaffold(setup_client, device_id=sample_device.id)
assert "device_created" not in resp.json()
def test_entities_are_persisted(
self,
setup_client,
sample_device,
picture_source_store,
css_store,
output_target_store,
):
resp = _scaffold(setup_client, device_id=sample_device.id)
data = resp.json()
# All entities retrievable from the stores
ps = picture_source_store.get(data["picture_source_id"])
assert ps is not None
css = css_store.get_source(data["color_strip_source_id"])
assert css is not None
ot = output_target_store.get(data["output_target_id"])
assert ot is not None
def test_entity_links_are_correct(
self,
setup_client,
sample_device,
picture_source_store,
css_store,
output_target_store,
):
resp = _scaffold(setup_client, device_id=sample_device.id)
data = resp.json()
ps = picture_source_store.get(data["picture_source_id"])
assert ps.capture_template_id == data["capture_template_id"]
assert ps.display_index == 0
css = css_store.get_source(data["color_strip_source_id"])
assert css.picture_source_id == data["picture_source_id"]
ot = output_target_store.get(data["output_target_id"])
assert ot.device_id == sample_device.id
assert ot.color_strip_source_id == data["color_strip_source_id"]
def test_entity_events_fire_after_success(self, setup_client, sample_device, event_log):
"""Events must be emitted for all created entities — and only after success."""
_scaffold(setup_client, device_id=sample_device.id)
types_fired = {(et, act) for et, act, _ in event_log}
assert ("picture_source", "created") in types_fired
assert ("color_strip_source", "created") in types_fired
assert ("output_target", "created") in types_fired
# Device is pre-existing — no device "created" event expected
assert ("device", "created") not in types_fired
def test_events_fired_only_once_per_entity(self, setup_client, sample_device, event_log):
"""No duplicate events for a single scaffold call."""
_scaffold(setup_client, device_id=sample_device.id)
ps_created = [(et, act, eid) for et, act, eid in event_log if et == "picture_source"]
css_created = [(et, act, eid) for et, act, eid in event_log if et == "color_strip_source"]
ot_created = [(et, act, eid) for et, act, eid in event_log if et == "output_target"]
assert len(ps_created) == 1
assert len(css_created) == 1
assert len(ot_created) == 1
# ---------------------------------------------------------------------------
# Scaffold: reuse existing capture template
# ---------------------------------------------------------------------------
class TestScaffoldReusesTemplate:
def test_reuse_existing_template(self, setup_client, sample_device, template_store):
"""TemplateStore auto-creates a 'Default' template; the scaffold must reuse it."""
all_templates_before = template_store.get_all_templates()
assert len(all_templates_before) >= 1, "TemplateStore should auto-create one"
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 201, resp.text
data = resp.json()
assert data["capture_template_reused"] is True
# No new templates created
all_templates_after = template_store.get_all_templates()
assert len(all_templates_after) == len(all_templates_before)
# ---------------------------------------------------------------------------
# Scaffold: validation errors
# ---------------------------------------------------------------------------
class TestScaffoldValidation:
def test_unknown_device_id_returns_404(self, setup_client):
resp = _scaffold(setup_client, device_id="device_doesnotexist")
assert resp.status_code == 404, resp.text
def test_missing_device_id_returns_422(self, setup_client):
"""device_id is now required — omitting it must yield 422."""
resp = setup_client.post("/api/v1/setup/scaffold", json={"display_index": 0})
assert resp.status_code == 422, resp.text
def test_display_index_above_max_returns_422(self, setup_client, sample_device):
"""display_index > 63 must be rejected with 422."""
resp = _scaffold(setup_client, device_id=sample_device.id, display_index=64)
assert resp.status_code == 422, resp.text
def test_display_index_at_max_accepted(self, setup_client, sample_device):
"""display_index == 63 is at the upper bound and must be accepted."""
resp = _scaffold(setup_client, device_id=sample_device.id, display_index=63)
assert resp.status_code == 201, resp.text
def test_display_index_negative_returns_422(self, setup_client, sample_device):
resp = _scaffold(setup_client, device_id=sample_device.id, display_index=-1)
assert resp.status_code == 422, resp.text
def test_custom_display_index_stored(self, setup_client, sample_device, picture_source_store):
resp = _scaffold(setup_client, device_id=sample_device.id, display_index=2)
assert resp.status_code == 201, resp.text
ps = picture_source_store.get(resp.json()["picture_source_id"])
assert ps.display_index == 2
# ---------------------------------------------------------------------------
# Scaffold: rollback on partial failure — no orphans AND no ghost events
# ---------------------------------------------------------------------------
class TestScaffoldRollback:
def test_no_orphans_when_css_creation_fails(
self,
setup_client,
sample_device,
picture_source_store,
css_store,
):
"""Force css_store.create_source to raise; expect the picture source to be deleted too."""
original_create = css_store.create_source
def _fail_create(*args, **kwargs):
raise ValueError("Simulated CSS creation failure")
css_store.create_source = _fail_create
try:
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 400, resp.text
# No picture sources should remain
remaining_ps = picture_source_store.get_all_streams()
assert len(remaining_ps) == 0, "Picture source should have been rolled back"
# No CSS created either
remaining_css = css_store.get_all_sources()
assert len(remaining_css) == 0
finally:
css_store.create_source = original_create
def test_no_orphans_when_output_target_creation_fails(
self,
setup_client,
sample_device,
picture_source_store,
css_store,
output_target_store,
):
"""Force output_target_store.create_wled_target to raise; picture source and CSS rolled back."""
original_create = output_target_store.create_wled_target
def _fail_create(*args, **kwargs):
raise ValueError("Simulated output target failure")
output_target_store.create_wled_target = _fail_create
try:
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 400, resp.text
assert len(picture_source_store.get_all_streams()) == 0
assert len(css_store.get_all_sources()) == 0
assert len(output_target_store.get_all_targets()) == 0
finally:
output_target_store.create_wled_target = original_create
def test_no_created_events_emitted_on_rollback(
self,
setup_client,
sample_device,
css_store,
event_log,
):
"""On failure no 'created' events must leak (deferred-event contract)."""
original_create = css_store.create_source
def _fail_create(*args, **kwargs):
raise ValueError("Simulated CSS failure")
css_store.create_source = _fail_create
try:
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 400, resp.text
created_events = [(et, act) for et, act, _ in event_log if act == "created"]
assert (
created_events == []
), f"No 'created' events should fire on rollback, got: {created_events}"
finally:
css_store.create_source = original_create
def test_reused_template_not_deleted_on_rollback(
self,
setup_client,
sample_device,
template_store,
css_store,
):
"""A reused (pre-existing) capture template must survive rollback."""
templates_before = {t.id for t in template_store.get_all_templates()}
original_create_css = css_store.create_source
def _fail_css(*args, **kwargs):
raise ValueError("Forced failure")
css_store.create_source = _fail_css
try:
_scaffold(setup_client, device_id=sample_device.id)
finally:
css_store.create_source = original_create_css
templates_after = {t.id for t in template_store.get_all_templates()}
assert templates_before == templates_after
def test_device_never_deleted_on_rollback(
self,
setup_client,
sample_device,
device_store,
css_store,
):
"""The pre-existing device must never be touched by rollback."""
original_create = css_store.create_source
def _fail_create(*args, **kwargs):
raise ValueError("Simulated CSS failure")
css_store.create_source = _fail_create
try:
_scaffold(setup_client, device_id=sample_device.id)
finally:
css_store.create_source = original_create
# Device must still exist
device = device_store.get(sample_device.id)
assert device is not None
assert device.name == sample_device.name
# ---------------------------------------------------------------------------
# Onboarding preference
# ---------------------------------------------------------------------------
@pytest.fixture
def pref_client(tmp_db):
from ledgrab.api.routes.preferences import router
from ledgrab.api.auth import verify_api_key
from ledgrab.api import dependencies as deps
app = FastAPI()
app.include_router(router)
app.dependency_overrides[verify_api_key] = lambda: "test"
app.dependency_overrides[deps.get_database] = lambda: tmp_db
return TestClient(app)
class TestOnboarding:
def test_get_default_returns_not_onboarded(self, pref_client):
resp = pref_client.get("/api/v1/preferences/onboarding")
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["onboarded"] is False
assert data["completed_at"] is None
def test_put_onboarded_true_round_trips(self, pref_client):
resp = pref_client.put("/api/v1/preferences/onboarding", json={"onboarded": True})
assert resp.status_code == 200, resp.text
data = resp.json()
assert data["onboarded"] is True
# Server auto-stamps completed_at
assert data["completed_at"] is not None
def test_get_after_put_reflects_stored_value(self, pref_client):
pref_client.put("/api/v1/preferences/onboarding", json={"onboarded": True})
resp = pref_client.get("/api/v1/preferences/onboarding")
assert resp.status_code == 200
assert resp.json()["onboarded"] is True
def test_put_false_clears_completed_at(self, pref_client):
# First set to true
pref_client.put("/api/v1/preferences/onboarding", json={"onboarded": True})
# Then reset
resp = pref_client.put("/api/v1/preferences/onboarding", json={"onboarded": False})
assert resp.status_code == 200
data = resp.json()
assert data["onboarded"] is False
assert data["completed_at"] is None
def test_put_with_explicit_completed_at_preserved(self, pref_client):
ts = "2026-01-01T00:00:00+00:00"
resp = pref_client.put(
"/api/v1/preferences/onboarding",
json={"onboarded": True, "completed_at": ts},
)
assert resp.status_code == 200
assert resp.json()["completed_at"] == ts
# ---------------------------------------------------------------------------
# Integration: scaffold → PUT calibration onto CSS → GET CSS round-trips
# ---------------------------------------------------------------------------
class TestScaffoldCalibrationIntegration:
def test_scaffold_then_update_css_calibration(
self,
setup_client,
sample_device,
css_store,
):
"""Full integration path: scaffold → apply solved calibration via CSS PUT.
Uses the same css_store fixture that the setup_client uses, so the
scaffolded entity is visible to it after creation.
"""
from ledgrab.core.capture.calibration import CalibrationConfig
# Step 1: scaffold
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 201, resp.text
css_id = resp.json()["color_strip_source_id"]
# Confirm entity exists in the shared store
css = css_store.get_source(css_id)
assert css is not None
# Step 2: build a solved calibration (mimics Phase 1 solve output)
solved_cal = CalibrationConfig(
layout="clockwise",
start_position="bottom_left",
leds_top=15,
leds_right=9,
leds_bottom=15,
leds_left=9,
)
# Step 3: persist via store update (the real CSS PUT does this)
css_store.update_source(css_id, calibration=solved_cal)
# Step 4: assert the calibration round-trips
updated = css_store.get_source(css_id)
assert updated.calibration.leds_top == 15
assert updated.calibration.leds_right == 9
assert updated.calibration.layout == "clockwise"
# ---------------------------------------------------------------------------
# Regression: scaffold registers the output target with ProcessorManager
# ---------------------------------------------------------------------------
class TestScaffoldRegistersWithManager:
def test_scaffold_calls_add_target_on_manager(
self,
setup_client,
sample_device,
mock_manager,
):
"""Blocker regression: scaffold must register the created output target
with the ProcessorManager so that a subsequent start call can find it.
WledOutputTarget.register_with_manager calls manager.add_target(...)
— we assert that method was invoked with the scaffolded target id.
"""
resp = _scaffold(setup_client, device_id=sample_device.id)
assert resp.status_code == 201, resp.text
target_id = resp.json()["output_target_id"]
# register_with_manager → manager.add_target(target_id=..., ...)
mock_manager.add_target.assert_called_once()
call_kwargs = mock_manager.add_target.call_args
assert call_kwargs.kwargs.get("target_id") == target_id, (
f"manager.add_target was not called with target_id={target_id!r}; "
f"actual call: {call_kwargs}"
)
def test_scaffold_rollback_calls_remove_target_on_manager(
self,
setup_client,
sample_device,
output_target_store,
mock_manager,
):
"""Rollback after a post-target-creation failure must call manager.remove_target
so no half-registered target lingers in the ProcessorManager.
We inject a failure by making register_with_manager raise RuntimeError
(bypassing the ValueError-only guard), which puts the outer except branch
in play. The target IS already in created_ids at that point, so rollback
must call manager.remove_target(target_id).
"""
created_target_ids: list[str] = []
original_create = output_target_store.create_wled_target
def _spy_create(*args, **kwargs):
target = original_create(*args, **kwargs)
created_target_ids.append(target.id)
return target
output_target_store.create_wled_target = _spy_create
try:
# Patch register_with_manager on WledOutputTarget to raise RuntimeError —
# RuntimeError bypasses the ValueError guard and triggers the outer except,
# so rollback fires with the target already in created_ids.
with patch(
"ledgrab.storage.wled_output_target.WledOutputTarget.register_with_manager",
side_effect=RuntimeError("Injected registration failure for rollback test"),
):
resp = setup_client.post(
"/api/v1/setup/scaffold",
json={"device_id": sample_device.id, "display_index": 0},
)
assert resp.status_code == 500, resp.text
assert len(created_target_ids) == 1, "spy did not record a created target"
target_id = created_target_ids[0]
mock_manager.remove_target.assert_called_with(target_id)
finally:
output_target_store.create_wled_target = original_create
Reference in New Issue View Git Blame Copy Permalink