alexei.dolgolyov
898912f8b1
Bundle the remaining backend touch-ups that the production review landed individually as small surgical edits across many modules: - MQTT runtime: fire-and-forget task tracking + drain resilience. - mqtt_source + store + storage/color_strip_source: secret_box encryption for credentials with auto-migration of plaintext fields. - devices/discovery_watcher: task tracking on watcher start/stop. - devices/wled_client + wled_provider: URL scheme inference helper applied at the create/update boundary so bare hostnames stay valid. - core/capture/screen_capture: hardened error paths. - core/processing (mapped/processed/processor_manager/video/wled_target): smaller follow-throughs from the registry refactor that landed earlier on the branch. - utils/safe_source + utils/file_ops + utils/__init__: shared URL + IP classification helpers + larger streaming upload size caps. - api/auth: WebSocket Origin allow-list + /docs auth-gate. - api/dependencies: register the new HTTP-endpoint store. - api/routes (assets, backup, webhooks): streaming-upload caps + asyncio.gather return_exceptions on broadcast loops. - tests/test_api + tests/e2e/test_backup_flow: cover the new caps and the Origin allow-list.
105 lines
3.2 KiB
Python
105 lines
3.2 KiB
Python
"""Tests for API endpoints (public + authenticated)."""
|
|
|
|
import os
|
|
import sys
|
|
|
|
import pytest
|
|
|
|
from ledgrab import __version__
|
|
from ledgrab.config import get_config
|
|
|
|
_has_display = bool(
|
|
os.environ.get("DISPLAY") or sys.platform == "win32" or sys.platform == "darwin"
|
|
)
|
|
requires_display = pytest.mark.skipif(not _has_display, reason="No display available (headless CI)")
|
|
|
|
# Build auth header from the first configured API key
|
|
_config = get_config()
|
|
_api_key = next(iter(_config.auth.api_keys.values()), "")
|
|
AUTH_HEADERS = {"Authorization": f"Bearer {_api_key}"} if _api_key else {}
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
"""Provide a TestClient backed by the isolated test database."""
|
|
from fastapi.testclient import TestClient
|
|
from ledgrab.main import app
|
|
|
|
with TestClient(app, raise_server_exceptions=False) as c:
|
|
yield c
|
|
|
|
|
|
def test_root_endpoint(client):
|
|
"""Test root endpoint returns the HTML dashboard."""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
|
|
def test_health_check(client):
|
|
"""Test health check endpoint."""
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
assert data["version"] == __version__
|
|
assert "timestamp" in data
|
|
|
|
|
|
def test_version_endpoint(client):
|
|
"""Test version endpoint."""
|
|
response = client.get("/api/v1/version")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["version"] == __version__
|
|
assert "python_version" in data
|
|
assert data["api_version"] == "v1"
|
|
|
|
|
|
@requires_display
|
|
def test_get_displays(client):
|
|
"""Test get displays endpoint (requires auth and a real display)."""
|
|
response = client.get("/api/v1/config/displays", headers=AUTH_HEADERS)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "displays" in data
|
|
assert "count" in data
|
|
assert isinstance(data["displays"], list)
|
|
assert data["count"] >= 0
|
|
|
|
# If displays are found, validate structure
|
|
if data["count"] > 0:
|
|
display = data["displays"][0]
|
|
assert "index" in display
|
|
assert "name" in display
|
|
assert "width" in display
|
|
assert "height" in display
|
|
assert "is_primary" in display
|
|
|
|
|
|
def test_openapi_docs(client):
|
|
"""Test OpenAPI documentation is available to authenticated clients."""
|
|
response = client.get("/openapi.json", headers=AUTH_HEADERS)
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["info"]["version"] == __version__
|
|
|
|
|
|
def test_swagger_ui(client):
|
|
"""Test Swagger UI is available to authenticated clients."""
|
|
response = client.get("/docs", headers=AUTH_HEADERS)
|
|
assert response.status_code == 200
|
|
assert "text/html" in response.headers["content-type"]
|
|
|
|
|
|
def test_openapi_docs_requires_auth(client):
|
|
"""OpenAPI surface must NOT be reachable without auth (info disclosure)."""
|
|
response = client.get("/openapi.json")
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_swagger_ui_requires_auth(client):
|
|
"""Swagger UI must NOT be reachable without auth."""
|
|
response = client.get("/docs")
|
|
assert response.status_code == 401
|