286 lines
9.2 KiB
Python
286 lines
9.2 KiB
Python
"""Integration tests for the main FastAPI application."""
|
|
|
|
import asyncio
|
|
import json
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch, Mock
|
|
from fastapi.testclient import TestClient
|
|
from datetime import datetime
|
|
|
|
# Import after mocking
|
|
from app.models import PortStatus, CodecType
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_recorder_manager():
|
|
"""Create a mock RecorderManager."""
|
|
manager = MagicMock()
|
|
|
|
# Mock get_all_status to return sample port statuses (synchronous)
|
|
mock_status = PortStatus(
|
|
port_index=0,
|
|
is_recording=False,
|
|
frame_count=0,
|
|
fps=29.97,
|
|
bitrate_mbps=50.0,
|
|
uptime_seconds=0,
|
|
current_file="/recordings/port0.mov",
|
|
codec=CodecType.PRORES
|
|
)
|
|
manager.get_all_status.return_value = [mock_status]
|
|
manager.stop_recording = AsyncMock()
|
|
|
|
return manager
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_scte35_manager():
|
|
"""Create a mock SCTE35Manager."""
|
|
manager = MagicMock()
|
|
return manager
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_hls_manager():
|
|
"""Create a mock HLSPreviewManager."""
|
|
manager = MagicMock()
|
|
manager.stop_preview = AsyncMock()
|
|
return manager
|
|
|
|
|
|
@pytest.fixture
|
|
def mock_settings():
|
|
"""Create mock Settings."""
|
|
settings = MagicMock()
|
|
settings.deltacast_port_count = 4
|
|
settings.ffmpeg_path = "/usr/bin/ffmpeg"
|
|
settings.recording_dir = "/recordings"
|
|
return settings
|
|
|
|
|
|
@pytest.fixture
|
|
def test_app(mock_recorder_manager, mock_scte35_manager, mock_hls_manager, mock_settings):
|
|
"""Create a test FastAPI app with mocked managers."""
|
|
|
|
# Patch the lifespan and create app without running startup
|
|
with patch('app.main.RecorderManager', return_value=mock_recorder_manager), \
|
|
patch('app.main.SCTE35Manager', return_value=mock_scte35_manager), \
|
|
patch('app.main.HLSPreviewManager', return_value=mock_hls_manager), \
|
|
patch('app.main.settings', mock_settings):
|
|
|
|
from app.main import app
|
|
|
|
# Manually inject the mocked managers for routes
|
|
from app.api import routes as routes_module
|
|
routes_module.recorder_manager = mock_recorder_manager
|
|
routes_module.scte35_manager = mock_scte35_manager
|
|
|
|
return app
|
|
|
|
|
|
@pytest.fixture
|
|
def client(test_app):
|
|
"""Create a TestClient for the test app."""
|
|
return TestClient(test_app)
|
|
|
|
|
|
class TestAppInitialization:
|
|
"""Tests for app initialization and configuration."""
|
|
|
|
def test_app_has_health_endpoint(self, client):
|
|
"""App responds to GET /api/health"""
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "ok"
|
|
assert "ports" in data
|
|
assert "recording_count" in data
|
|
|
|
def test_app_has_ports_endpoint(self, client):
|
|
"""App responds to GET /api/ports"""
|
|
response = client.get("/api/ports")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) > 0
|
|
assert "port_index" in data[0]
|
|
|
|
def test_cors_middleware_enabled(self, test_app):
|
|
"""CORS middleware is configured in the app"""
|
|
# Check that CORSMiddleware is in the middleware stack
|
|
from fastapi.middleware.cors import CORSMiddleware
|
|
|
|
has_cors = any(
|
|
isinstance(middleware.cls, type)
|
|
and issubclass(middleware.cls, CORSMiddleware)
|
|
for middleware in test_app.user_middleware
|
|
)
|
|
assert has_cors
|
|
|
|
def test_hls_endpoint_404_for_missing_file(self, client):
|
|
"""GET /hls/nonexistent.m3u8 returns 404"""
|
|
response = client.get("/hls/nonexistent.m3u8")
|
|
assert response.status_code == 404
|
|
assert "not found" in response.json()["detail"].lower()
|
|
|
|
|
|
class TestBroadcastPortStatus:
|
|
"""Tests for broadcast_port_status background task."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_broadcast_port_status_sends_data(self, mock_recorder_manager, mock_settings):
|
|
"""broadcast_port_status sends correct data structure"""
|
|
|
|
# Mock connection manager to capture broadcast
|
|
broadcast_called = asyncio.Event()
|
|
broadcast_data = {}
|
|
|
|
async def mock_broadcast(data):
|
|
broadcast_data.update(data)
|
|
broadcast_called.set()
|
|
|
|
# Create the task
|
|
from app.main import broadcast_port_status
|
|
|
|
with patch('app.main.recorder_manager', mock_recorder_manager), \
|
|
patch('app.main.connection_manager') as mock_conn_manager:
|
|
|
|
mock_conn_manager.broadcast = mock_broadcast
|
|
|
|
# Run broadcast once
|
|
task = asyncio.create_task(broadcast_port_status())
|
|
|
|
# Wait for broadcast
|
|
try:
|
|
await asyncio.wait_for(broadcast_called.wait(), timeout=2.0)
|
|
except asyncio.TimeoutError:
|
|
pytest.fail("Broadcast did not complete within timeout")
|
|
finally:
|
|
task.cancel()
|
|
try:
|
|
await task
|
|
except asyncio.CancelledError:
|
|
pass
|
|
|
|
# Verify broadcast data structure
|
|
assert "type" in broadcast_data
|
|
assert broadcast_data["type"] == "port_status"
|
|
assert "ports" in broadcast_data
|
|
assert isinstance(broadcast_data["ports"], list)
|
|
if broadcast_data["ports"]:
|
|
assert "port_index" in broadcast_data["ports"][0]
|
|
|
|
|
|
class TestWebSocketIntegration:
|
|
"""Tests for WebSocket functionality."""
|
|
|
|
def test_websocket_endpoint_exists(self, client):
|
|
"""WebSocket endpoint is defined in app routes"""
|
|
from app.main import app
|
|
|
|
ws_routes = [
|
|
route for route in app.routes
|
|
if hasattr(route, 'path') and route.path == "/ws"
|
|
]
|
|
assert len(ws_routes) > 0
|
|
|
|
def test_connection_manager_initialization(self):
|
|
"""ConnectionManager is initialized with empty connections"""
|
|
from app.api.websocket import ConnectionManager
|
|
|
|
manager = ConnectionManager()
|
|
assert manager.active_connections == []
|
|
|
|
|
|
class TestHLSEndpoint:
|
|
"""Tests for HLS file serving endpoint."""
|
|
|
|
def test_hls_endpoint_returns_404_for_missing_file(self, client):
|
|
"""HLS endpoint returns 404 for non-existent file"""
|
|
response = client.get("/hls/missing-segment.ts")
|
|
assert response.status_code == 404
|
|
|
|
def test_hls_endpoint_handles_various_filenames(self, client):
|
|
"""HLS endpoint correctly handles various filename patterns"""
|
|
test_files = [
|
|
"stream.m3u8",
|
|
"segment-001.ts",
|
|
"playlist_index.m3u8",
|
|
"../../../etc/passwd", # Path traversal attempt
|
|
]
|
|
|
|
for filename in test_files:
|
|
response = client.get(f"/hls/{filename}")
|
|
# Should all return 404 since files don't exist
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestAPIRouter:
|
|
"""Tests to verify API router is properly included."""
|
|
|
|
def test_api_router_is_included(self, client):
|
|
"""API router is included in app"""
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
|
|
def test_api_prefix_is_correct(self, client):
|
|
"""API routes have /api prefix"""
|
|
response = client.get("/api/health")
|
|
assert response.status_code == 200
|
|
|
|
# Routes without /api should not work
|
|
response = client.get("/health")
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestErrorHandling:
|
|
"""Tests for error handling in the application."""
|
|
|
|
def test_404_for_undefined_route(self, client):
|
|
"""Undefined routes return 404"""
|
|
response = client.get("/api/nonexistent")
|
|
assert response.status_code == 404
|
|
|
|
def test_404_error_is_dict(self, client):
|
|
"""404 errors return JSON response"""
|
|
response = client.get("/api/nonexistent")
|
|
assert isinstance(response.json(), dict)
|
|
assert "detail" in response.json()
|
|
|
|
|
|
class TestAppMetadata:
|
|
"""Tests for app metadata and OpenAPI documentation."""
|
|
|
|
def test_app_has_title(self, client):
|
|
"""App has correct title"""
|
|
from app.main import app
|
|
assert app.title == "Deltacast SDI Recorder"
|
|
|
|
def test_app_has_version(self, client):
|
|
"""App has version information"""
|
|
from app.main import app
|
|
assert app.version == "1.0.0"
|
|
|
|
def test_openapi_schema_available(self, client):
|
|
"""OpenAPI schema is available"""
|
|
response = client.get("/openapi.json")
|
|
assert response.status_code == 200
|
|
schema = response.json()
|
|
assert "openapi" in schema
|
|
assert "paths" in schema
|
|
|
|
|
|
class TestManagerInjection:
|
|
"""Tests for proper manager injection into routes module."""
|
|
|
|
def test_managers_injected_into_routes(self, mock_recorder_manager, mock_scte35_manager):
|
|
"""Managers are correctly injected into routes module"""
|
|
from app.api import routes as routes_module
|
|
|
|
# Managers should be set by the fixture
|
|
assert routes_module.recorder_manager is not None
|
|
assert routes_module.scte35_manager is not None
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|