alexei.dolgolyov
0409cd8b66
Backend engine for guided LED-chase calibration, driven by the upcoming
auto-calibration UI (phase 3) and first-run wizard (phase 4).
- solve_calibration(): pure function mapping start corner + direction + 4
corner-tap indices to per-edge LED counts, consistent with EDGE_ORDER/
EDGE_REVERSE so it round-trips through build_segments().
- CalibrationChaseMixin.set_calibration_pixel(): light a specific LED index
(+ optional window) on a device, reusing the device_test_mode idle-client
send path.
- CalibrationSession: single-active session with start/position/stop/cancel,
a 60s idle-timeout watchdog, and a concurrency lock so interleaved calls
can't corrupt the stop/restore bookkeeping — start() stops + remembers any
running target on the device and stop/cancel/timeout always restore it
(never leaves the device dark or stuck in chase).
- Routes /api/v1/calibration/{session,session/position,session/stop,
session/cancel,session/state,solve} (all AuthRequired, bounds-validated);
calibration is persisted by reusing the existing PUT /color-strip-sources/
{id} (hot-reloads running streams) rather than a duplicate endpoint.
- Tests: 19 solver pure-logic + 19 route/bounds. docs/API.md updated.
Part of the edge-calibration + first-run-wizard feature (Big Bang; intermediate
phase — full build/suite gated at the final phase).
355 lines
11 KiB
Python
355 lines
11 KiB
Python
"""Happy-path and bounds-validation tests for calibration API routes.
|
|
|
|
Runs with the full app test-client stack but mocks the ProcessorManager
|
|
so no real LED devices are required.
|
|
|
|
Note: Deep adversarial coverage is deferred to the Phase 4 test-writer
|
|
(Big Bang strategy).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from fastapi import FastAPI
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from ledgrab.api.routes.calibration import router
|
|
from ledgrab.core.capture.calibration_session import get_calibration_session
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_manager() -> MagicMock:
|
|
"""A minimal fake ProcessorManager."""
|
|
mgr = MagicMock()
|
|
# Simulate a registered device with 100 LEDs
|
|
ds = MagicMock()
|
|
ds.led_count = 100
|
|
mgr._devices = {"dev1": ds}
|
|
mgr.get_processing_target_for_device = MagicMock(return_value=None)
|
|
mgr.stop_processing = AsyncMock()
|
|
mgr.start_processing = AsyncMock()
|
|
mgr.send_clear_pixels = AsyncMock()
|
|
mgr.set_calibration_pixel = AsyncMock()
|
|
return mgr
|
|
|
|
|
|
@pytest_asyncio.fixture(autouse=True)
|
|
async def reset_session():
|
|
"""Reset the module-level CalibrationSession singleton before each test."""
|
|
|
|
import asyncio
|
|
|
|
def _clear(session) -> None:
|
|
session._active = False
|
|
session._device_id = None
|
|
session._led_count = 0
|
|
session._prior_target_id = None
|
|
session._last_activity = None
|
|
session._manager = None
|
|
# Reset lock so a test that aborted mid-await doesn't leave it locked
|
|
session._lock = asyncio.Lock()
|
|
|
|
session = get_calibration_session()
|
|
# Cancel any leftover watchdog task before clearing
|
|
if session._timeout_task and not session._timeout_task.done():
|
|
session._timeout_task.cancel()
|
|
try:
|
|
await session._timeout_task
|
|
except Exception:
|
|
pass
|
|
session._timeout_task = None
|
|
_clear(session)
|
|
|
|
yield
|
|
|
|
# Cleanup after test
|
|
if session._timeout_task and not session._timeout_task.done():
|
|
session._timeout_task.cancel()
|
|
try:
|
|
await session._timeout_task
|
|
except Exception:
|
|
pass
|
|
session._timeout_task = None
|
|
_clear(session)
|
|
|
|
|
|
@pytest.fixture()
|
|
def app(mock_manager: MagicMock) -> FastAPI:
|
|
"""Tiny FastAPI app with only the calibration router and auth disabled."""
|
|
from fastapi import FastAPI
|
|
from ledgrab.api.auth import verify_api_key
|
|
from ledgrab.api import dependencies as deps_mod
|
|
|
|
_app = FastAPI()
|
|
_app.include_router(router)
|
|
# Override the underlying dependency that AuthRequired resolves to
|
|
_app.dependency_overrides[verify_api_key] = lambda: "test-token"
|
|
_app.dependency_overrides[deps_mod.get_processor_manager] = lambda: mock_manager
|
|
return _app
|
|
|
|
|
|
@pytest_asyncio.fixture()
|
|
async def client(app: FastAPI):
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session start
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_session_success(client: AsyncClient, mock_manager: MagicMock):
|
|
resp = await client.post("/api/v1/calibration/session", json={"device_id": "dev1"})
|
|
assert resp.status_code == 201
|
|
data = resp.json()
|
|
assert data["active"] is True
|
|
assert data["device_id"] == "dev1"
|
|
assert data["led_count"] == 100
|
|
mock_manager.send_clear_pixels.assert_awaited_once_with("dev1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_session_unknown_device(client: AsyncClient):
|
|
resp = await client.post("/api/v1/calibration/session", json={"device_id": "does_not_exist"})
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session position
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_position_success(client: AsyncClient, mock_manager: MagicMock):
|
|
# Start session first
|
|
await client.post("/api/v1/calibration/session", json={"device_id": "dev1"})
|
|
resp = await client.post(
|
|
"/api/v1/calibration/session/position", json={"index": 42, "window": 2}
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["active"] is True
|
|
mock_manager.set_calibration_pixel.assert_awaited_with("dev1", 42, window=2)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_position_out_of_range(client: AsyncClient, mock_manager: MagicMock):
|
|
"""index >= led_count → 400."""
|
|
await client.post("/api/v1/calibration/session", json={"device_id": "dev1"})
|
|
resp = await client.post(
|
|
"/api/v1/calibration/session/position", json={"index": 100, "window": 1}
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_position_negative_index_422(client: AsyncClient):
|
|
"""index < 0 → Pydantic 422."""
|
|
resp = await client.post(
|
|
"/api/v1/calibration/session/position", json={"index": -1, "window": 1}
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_position_no_active_session(client: AsyncClient):
|
|
"""Calling position without starting a session → 400."""
|
|
resp = await client.post("/api/v1/calibration/session/position", json={"index": 5, "window": 1})
|
|
assert resp.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session stop
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_session_clears_device(client: AsyncClient, mock_manager: MagicMock):
|
|
await client.post("/api/v1/calibration/session", json={"device_id": "dev1"})
|
|
resp = await client.post("/api/v1/calibration/session/stop")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["active"] is False
|
|
# send_clear_pixels called at start AND at stop
|
|
assert mock_manager.send_clear_pixels.await_count == 2
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_restores_prior_target(client: AsyncClient, mock_manager: MagicMock):
|
|
"""When a target was running, stop should restart it."""
|
|
mock_manager.get_processing_target_for_device = MagicMock(return_value="tgt1")
|
|
await client.post("/api/v1/calibration/session", json={"device_id": "dev1"})
|
|
await client.post("/api/v1/calibration/session/stop")
|
|
mock_manager.start_processing.assert_awaited_once_with("tgt1")
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_stop_no_active_session_is_ok(client: AsyncClient):
|
|
"""stop when inactive → 200 with active=False."""
|
|
resp = await client.post("/api/v1/calibration/session/stop")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["active"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Session state
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_state_inactive(client: AsyncClient):
|
|
resp = await client.get("/api/v1/calibration/session/state")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["active"] is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_state_active(client: AsyncClient, mock_manager: MagicMock):
|
|
await client.post("/api/v1/calibration/session", json={"device_id": "dev1"})
|
|
resp = await client.get("/api/v1/calibration/session/state")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["active"] is True
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Solve endpoint
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_with_device_id(client: AsyncClient):
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"device_id": "dev1",
|
|
"start_position": "bottom_left",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 30, 60, 80],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["mode"] == "simple"
|
|
# bottom_left/clockwise EDGE_ORDER: left, top, right, bottom
|
|
# left=30, top=30, right=20, bottom=20 → total=100
|
|
assert data["leds_left"] == 30
|
|
assert data["leds_top"] == 30
|
|
assert data["leds_right"] == 20
|
|
assert data["leds_bottom"] == 20
|
|
assert data["layout"] == "clockwise"
|
|
assert data["start_position"] == "bottom_left"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_with_led_count(client: AsyncClient):
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"led_count": 80,
|
|
"start_position": "top_left",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 20, 40, 60],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert sum([data["leds_top"], data["leds_right"], data["leds_bottom"], data["leds_left"]]) == 80
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_missing_device_and_led_count(client: AsyncClient):
|
|
"""Omitting both device_id and led_count → 422 (model validator)."""
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"start_position": "bottom_left",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 25, 50, 75],
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_unknown_device(client: AsyncClient):
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"device_id": "no_such_device",
|
|
"start_position": "bottom_left",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 25, 50, 75],
|
|
},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_invalid_start_position_422(client: AsyncClient):
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"led_count": 100,
|
|
"start_position": "invalid_corner",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 25, 50, 75],
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_invalid_layout_422(client: AsyncClient):
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"led_count": 100,
|
|
"start_position": "bottom_left",
|
|
"layout": "diagonal",
|
|
"corner_indices": [0, 25, 50, 75],
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_wrong_corner_count_422(client: AsyncClient):
|
|
"""Only 3 corner indices → 422 (min_length=4)."""
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"led_count": 100,
|
|
"start_position": "bottom_left",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 25, 50],
|
|
},
|
|
)
|
|
assert resp.status_code == 422
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_solve_with_offset(client: AsyncClient):
|
|
resp = await client.post(
|
|
"/api/v1/calibration/solve",
|
|
json={
|
|
"led_count": 100,
|
|
"start_position": "bottom_left",
|
|
"layout": "clockwise",
|
|
"corner_indices": [0, 25, 50, 75],
|
|
"offset": 7,
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["offset"] == 7
|