alexei.dolgolyov
17dd2e02ba
Multi-dimension review of v0.8.2. Excludes the deliberately deferred default_config.yaml weak-default-key item (C1). Backend: - calibration: create_default_calibration no longer exceeds led_count for small odd counts (bounded trim + regression test) - game-integration: generic webhook now requires auth_token; constant-time compare_digest in all adapters; per-IP failed-auth rate limit on the ingest route; auth_token encrypted at rest via secret_box (migration-safe) - playlist engine: serialize _state/_task under the lifecycle lock to close a delete-mid-play race (+ concurrency tests) - main: stop the calibration session on shutdown (restore prior target) - home_assistant: validate HA host via the LAN classifier on create/update - perf: drop slow preview-WS clients instead of blocking the send loop; cache composite full-strip resize linspaces; effect_stream lava reuses scratch Frontend: - setup/auto-calibration wizard: guard _state after awaits (cancel-safe), await session teardown before output start, busy-gate skip-calibration, manual display input keeps focus, move focus on step change - calibration: destroy EntitySelect on modal close - color-strips test: dirty-flag-gated render + cached ctx/ImageData - a11y/TV: focus-visible for new wizard/auto-cal/corner controls, aria-labels on the spatial corner/edge picker; theme-aware syntax tokens; dead/undefined CSS tokens removed; .modal-error styled; i18n titles (en/ru/zh) Android: - ApiKeyManager: EncryptedSharedPreferences with verified, data-safe legacy migration that never rotates an existing key - CaptureService: validate MediaProjection token before promoting; satisfy the startForeground 5s contract on the bail path - NotificationListener: connection-scoped executor with lazy fallback - BLE: request BLUETOOTH_SCAN/CONNECT at runtime + guard handler-thread SecurityExceptions - Root: cancellation-aware su grant probe Adds 14 tests. Gate: ruff + tsc 5.9.3 + esbuild + pytest (2185 passed) + compileDebugKotlin all green.
553 lines
18 KiB
Python
553 lines
18 KiB
Python
"""Tests for game integration API routes.
|
|
|
|
Uses FastAPI TestClient with dependency overrides to test route handlers
|
|
in isolation from the real application.
|
|
"""
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from ledgrab.api.routes.game_integration import router
|
|
from ledgrab.core.game_integration.adapter_registry import AdapterRegistry
|
|
from ledgrab.core.game_integration.base_adapter import GameAdapter
|
|
from ledgrab.core.game_integration.event_bus import GameEventBus
|
|
from ledgrab.core.game_integration.events import GameEvent
|
|
from ledgrab.storage.game_integration_store import GameIntegrationStore
|
|
from ledgrab.api import dependencies as deps
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Test adapter for ingestion tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class _TestAdapter(GameAdapter):
|
|
ADAPTER_TYPE = "test_adapter"
|
|
DISPLAY_NAME = "Test Adapter"
|
|
GAME_NAME = "Test Game"
|
|
SUPPORTED_EVENTS = ["health", "kill"]
|
|
|
|
@classmethod
|
|
def parse_payload(cls, payload, adapter_config, prev_state):
|
|
events = []
|
|
if "health" in payload:
|
|
events.append(
|
|
GameEvent(
|
|
adapter_id=adapter_config.get("integration_id", "test"),
|
|
event_type="health",
|
|
value=payload["health"] / 100.0,
|
|
)
|
|
)
|
|
return events, prev_state
|
|
|
|
@classmethod
|
|
def validate_auth(cls, headers, payload, adapter_config):
|
|
token = adapter_config.get("auth_token")
|
|
if not token:
|
|
return True
|
|
return headers.get("x-auth-token") == token
|
|
|
|
@classmethod
|
|
def get_config_schema(cls):
|
|
return {
|
|
"type": "object",
|
|
"properties": {
|
|
"auth_token": {"type": "string", "description": "Auth token"},
|
|
},
|
|
}
|
|
|
|
@classmethod
|
|
def get_setup_instructions(cls):
|
|
return "Configure the test adapter with an auth token."
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _make_app():
|
|
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 game_store(_route_db):
|
|
return GameIntegrationStore(_route_db)
|
|
|
|
|
|
@pytest.fixture
|
|
def event_bus():
|
|
return GameEventBus()
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _register_test_adapter():
|
|
"""Register and clean up the test adapter for each test."""
|
|
AdapterRegistry.register(_TestAdapter)
|
|
yield
|
|
AdapterRegistry.clear_registry()
|
|
|
|
|
|
@pytest.fixture
|
|
def client(game_store, event_bus):
|
|
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"
|
|
app.dependency_overrides[deps.get_game_integration_store] = lambda: game_store
|
|
app.dependency_overrides[deps.get_game_event_bus] = lambda: event_bus
|
|
|
|
return TestClient(app, raise_server_exceptions=False)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helper
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _create_integration(client, name="Test Integration", adapter_type="test_adapter", **kwargs):
|
|
"""Helper to create an integration via API."""
|
|
body = {
|
|
"name": name,
|
|
"adapter_type": adapter_type,
|
|
**kwargs,
|
|
}
|
|
resp = client.post("/api/v1/game-integrations", json=body)
|
|
assert resp.status_code == 201, resp.text
|
|
return resp.json()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# CRUD tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestCreateIntegration:
|
|
def test_create_basic(self, client):
|
|
data = _create_integration(client)
|
|
assert data["id"].startswith("gi_")
|
|
assert data["name"] == "Test Integration"
|
|
assert data["adapter_type"] == "test_adapter"
|
|
assert data["enabled"] is True
|
|
|
|
def test_create_with_mappings(self, client):
|
|
data = _create_integration(
|
|
client,
|
|
event_mappings=[
|
|
{"event_type": "health", "effect": "gradient", "color": [255, 0, 0]},
|
|
{"event_type": "kill", "effect": "flash", "color": [0, 255, 0]},
|
|
],
|
|
)
|
|
assert len(data["event_mappings"]) == 2
|
|
assert data["event_mappings"][0]["event_type"] == "health"
|
|
|
|
def test_create_with_config(self, client):
|
|
data = _create_integration(
|
|
client,
|
|
adapter_config={"auth_token": "secret123"},
|
|
description="My game",
|
|
tags=["fps"],
|
|
)
|
|
assert data["adapter_config"] == {"auth_token": "secret123"}
|
|
assert data["description"] == "My game"
|
|
assert data["tags"] == ["fps"]
|
|
|
|
def test_create_duplicate_name(self, client):
|
|
_create_integration(client, name="Unique")
|
|
resp = client.post(
|
|
"/api/v1/game-integrations",
|
|
json={"name": "Unique", "adapter_type": "test_adapter"},
|
|
)
|
|
assert resp.status_code == 400
|
|
assert "already exists" in resp.json()["detail"]
|
|
|
|
def test_create_empty_name(self, client):
|
|
resp = client.post(
|
|
"/api/v1/game-integrations",
|
|
json={"name": "", "adapter_type": "test_adapter"},
|
|
)
|
|
assert resp.status_code == 422 # Pydantic validation
|
|
|
|
|
|
class TestListIntegrations:
|
|
def test_list_empty(self, client):
|
|
resp = client.get("/api/v1/game-integrations")
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
assert data["count"] == 0
|
|
assert data["integrations"] == []
|
|
|
|
def test_list_multiple(self, client):
|
|
_create_integration(client, name="Int 1")
|
|
_create_integration(client, name="Int 2")
|
|
|
|
resp = client.get("/api/v1/game-integrations")
|
|
data = resp.json()
|
|
assert data["count"] == 2
|
|
names = {i["name"] for i in data["integrations"]}
|
|
assert names == {"Int 1", "Int 2"}
|
|
|
|
|
|
class TestGetIntegration:
|
|
def test_get_existing(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.get(f"/api/v1/game-integrations/{created['id']}")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "Test Integration"
|
|
|
|
def test_get_nonexistent(self, client):
|
|
resp = client.get("/api/v1/game-integrations/gi_nonexist")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestUpdateIntegration:
|
|
def test_update_name(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.put(
|
|
f"/api/v1/game-integrations/{created['id']}",
|
|
json={"name": "Updated Name"},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["name"] == "Updated Name"
|
|
|
|
def test_update_enabled(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.put(
|
|
f"/api/v1/game-integrations/{created['id']}",
|
|
json={"enabled": False},
|
|
)
|
|
assert resp.status_code == 200
|
|
assert resp.json()["enabled"] is False
|
|
|
|
def test_update_nonexistent(self, client):
|
|
resp = client.put(
|
|
"/api/v1/game-integrations/gi_nonexist",
|
|
json={"name": "x"},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_update_to_duplicate_name(self, client):
|
|
_create_integration(client, name="Name A")
|
|
b = _create_integration(client, name="Name B")
|
|
resp = client.put(
|
|
f"/api/v1/game-integrations/{b['id']}",
|
|
json={"name": "Name A"},
|
|
)
|
|
assert resp.status_code == 400
|
|
|
|
|
|
class TestDeleteIntegration:
|
|
def test_delete_existing(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.delete(f"/api/v1/game-integrations/{created['id']}")
|
|
assert resp.status_code == 204
|
|
|
|
# Verify it's gone
|
|
resp = client.get(f"/api/v1/game-integrations/{created['id']}")
|
|
assert resp.status_code == 404
|
|
|
|
def test_delete_nonexistent(self, client):
|
|
resp = client.delete("/api/v1/game-integrations/gi_nonexist")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Event ingestion tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEventIngestion:
|
|
def test_ingest_basic(self, client, event_bus):
|
|
created = _create_integration(client)
|
|
integration_id = created["id"]
|
|
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/event",
|
|
json={"data": {"health": 75}},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
# Verify event was published to bus
|
|
recent = event_bus.get_recent_events()
|
|
assert len(recent) == 1
|
|
assert recent[0].event_type == "health"
|
|
assert recent[0].value == 0.75
|
|
|
|
def test_ingest_disabled_integration(self, client):
|
|
created = _create_integration(client)
|
|
integration_id = created["id"]
|
|
|
|
# Disable the integration
|
|
client.put(
|
|
f"/api/v1/game-integrations/{integration_id}",
|
|
json={"enabled": False},
|
|
)
|
|
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/event",
|
|
json={"data": {"health": 50}},
|
|
)
|
|
assert resp.status_code == 409
|
|
|
|
def test_ingest_nonexistent_integration(self, client):
|
|
resp = client.post(
|
|
"/api/v1/game-integrations/gi_nonexist/event",
|
|
json={"data": {"health": 50}},
|
|
)
|
|
assert resp.status_code == 404
|
|
|
|
def test_ingest_auth_failure(self, client):
|
|
created = _create_integration(
|
|
client,
|
|
adapter_config={"auth_token": "correct_token"},
|
|
)
|
|
integration_id = created["id"]
|
|
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/event",
|
|
json={"data": {"health": 50}},
|
|
headers={"x-auth-token": "wrong_token"},
|
|
)
|
|
assert resp.status_code == 403
|
|
|
|
def test_ingest_auth_success(self, client, event_bus):
|
|
created = _create_integration(
|
|
client,
|
|
adapter_config={"auth_token": "correct_token"},
|
|
)
|
|
integration_id = created["id"]
|
|
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/event",
|
|
json={"data": {"health": 50}},
|
|
headers={"x-auth-token": "correct_token"},
|
|
)
|
|
assert resp.status_code == 204
|
|
|
|
recent = event_bus.get_recent_events()
|
|
assert len(recent) == 1
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Failed-auth rate limiting (brute-force defence on the ingest route)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.fixture
|
|
def _reset_auth_fail_limiter():
|
|
"""Clear the module-level failed-auth hit map before and after each test.
|
|
|
|
The limiter keeps per-IP state in a process-global dict, so without this
|
|
reset, attempts from earlier tests would bleed into later ones.
|
|
"""
|
|
from ledgrab.api.routes import game_integration as gi
|
|
|
|
gi._auth_fail_hits.clear()
|
|
yield
|
|
gi._auth_fail_hits.clear()
|
|
|
|
|
|
class TestIngestRateLimiting:
|
|
def test_failed_auth_attempts_are_rate_limited(self, client, _reset_auth_fail_limiter):
|
|
from ledgrab.api.routes import game_integration as gi
|
|
|
|
created = _create_integration(
|
|
client,
|
|
adapter_config={"auth_token": "correct_token"},
|
|
)
|
|
integration_id = created["id"]
|
|
url = f"/api/v1/game-integrations/{integration_id}/event"
|
|
bad = {"x-auth-token": "wrong_token"}
|
|
|
|
# Burn through the failed-auth budget — each returns 403.
|
|
for _ in range(gi._AUTH_FAIL_LIMIT):
|
|
resp = client.post(url, json={"data": {"health": 1}}, headers=bad)
|
|
assert resp.status_code == 403
|
|
|
|
# The next attempt from the same IP is throttled with 429.
|
|
resp = client.post(url, json={"data": {"health": 1}}, headers=bad)
|
|
assert resp.status_code == 429
|
|
|
|
def test_successful_ingest_not_rate_limited(self, client, event_bus, _reset_auth_fail_limiter):
|
|
from ledgrab.api.routes import game_integration as gi
|
|
|
|
created = _create_integration(
|
|
client,
|
|
adapter_config={"auth_token": "correct_token"},
|
|
)
|
|
integration_id = created["id"]
|
|
url = f"/api/v1/game-integrations/{integration_id}/event"
|
|
good = {"x-auth-token": "correct_token"}
|
|
|
|
# High-rate legitimate ingestion well past the failed-auth threshold
|
|
# must NOT be throttled — only failures count toward the limit.
|
|
for _ in range(gi._AUTH_FAIL_LIMIT + 10):
|
|
resp = client.post(url, json={"data": {"health": 50}}, headers=good)
|
|
assert resp.status_code == 204
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Status / diagnostics tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestStatus:
|
|
def test_status_no_events(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.get(f"/api/v1/game-integrations/{created['id']}/status")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["integration_id"] == created["id"]
|
|
assert data["enabled"] is True
|
|
assert data["connected"] is False
|
|
assert data["event_count"] == 0
|
|
|
|
def test_status_after_events(self, client):
|
|
created = _create_integration(client)
|
|
integration_id = created["id"]
|
|
|
|
# Send an event
|
|
client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/event",
|
|
json={"data": {"health": 80}},
|
|
)
|
|
|
|
resp = client.get(f"/api/v1/game-integrations/{integration_id}/status")
|
|
data = resp.json()
|
|
assert data["event_count"] == 1
|
|
assert data["connected"] is True
|
|
assert "health" in data["event_counts_by_type"]
|
|
|
|
def test_status_nonexistent(self, client):
|
|
resp = client.get("/api/v1/game-integrations/gi_nonexist/status")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
class TestRecentEvents:
|
|
def test_recent_events_empty(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.get(f"/api/v1/game-integrations/{created['id']}/events")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["count"] == 0
|
|
assert data["events"] == []
|
|
|
|
def test_recent_events_nonexistent(self, client):
|
|
resp = client.get("/api/v1/game-integrations/gi_nonexist/events")
|
|
assert resp.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Adapter metadata tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestAdapterMetadata:
|
|
def test_list_adapters(self, client):
|
|
resp = client.get("/api/v1/game-adapters")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["count"] >= 1
|
|
|
|
adapter = data["adapters"][0]
|
|
assert adapter["adapter_type"] == "test_adapter"
|
|
assert adapter["display_name"] == "Test Adapter"
|
|
assert adapter["game_name"] == "Test Game"
|
|
assert "health" in adapter["supported_events"]
|
|
assert "config_schema" in adapter
|
|
assert "setup_instructions" in adapter
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Preset tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestPresets:
|
|
def test_list_presets(self, client):
|
|
resp = client.get("/api/v1/game-integrations/presets")
|
|
assert resp.status_code == 200
|
|
|
|
data = resp.json()
|
|
assert data["count"] >= 4
|
|
keys = {p["key"] for p in data["presets"]}
|
|
assert "fps_combat" in keys
|
|
assert "moba_health" in keys
|
|
assert "racing" in keys
|
|
assert "generic_alert" in keys
|
|
|
|
def test_preset_has_mappings(self, client):
|
|
resp = client.get("/api/v1/game-integrations/presets")
|
|
data = resp.json()
|
|
for preset in data["presets"]:
|
|
assert len(preset["event_mappings"]) > 0
|
|
for m in preset["event_mappings"]:
|
|
assert "event_type" in m
|
|
assert "effect" in m
|
|
assert "color" in m
|
|
|
|
def test_apply_preset_replace(self, client):
|
|
created = _create_integration(
|
|
client,
|
|
event_mappings=[{"event_type": "kill", "effect": "flash", "color": [255, 0, 0]}],
|
|
)
|
|
integration_id = created["id"]
|
|
assert len(created["event_mappings"]) == 1
|
|
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/apply-preset",
|
|
json={"preset_key": "fps_combat", "replace": True},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# Should have replaced the single mapping with preset mappings
|
|
assert len(data["event_mappings"]) >= 3
|
|
|
|
def test_apply_preset_append(self, client):
|
|
created = _create_integration(
|
|
client,
|
|
event_mappings=[{"event_type": "kill", "effect": "flash", "color": [255, 0, 0]}],
|
|
)
|
|
integration_id = created["id"]
|
|
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{integration_id}/apply-preset",
|
|
json={"preset_key": "generic_alert", "replace": False},
|
|
)
|
|
assert resp.status_code == 200
|
|
data = resp.json()
|
|
# Should have original + preset mappings
|
|
assert len(data["event_mappings"]) >= 5
|
|
|
|
def test_apply_preset_unknown_key(self, client):
|
|
created = _create_integration(client)
|
|
resp = client.post(
|
|
f"/api/v1/game-integrations/{created['id']}/apply-preset",
|
|
json={"preset_key": "nonexistent"},
|
|
)
|
|
assert resp.status_code == 404
|
|
assert "not found" in resp.json()["detail"]
|
|
|
|
def test_apply_preset_unknown_integration(self, client):
|
|
resp = client.post(
|
|
"/api/v1/game-integrations/gi_nonexist/apply-preset",
|
|
json={"preset_key": "fps_combat"},
|
|
)
|
|
assert resp.status_code == 404
|