"""
Test suite for Story 13: Mobile-friendly status dashboard.
Covers FastAPI endpoints, static file serving, API integration, and dashboard functionality.
"""
import pytest
import json
import time
from unittest.mock import Mock, patch, MagicMock
from datetime import datetime, timedelta
from io import BytesIO
import sys
import os
# Add the project root to the path
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from fastapi.testclient import TestClient
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
# Import the main application modules
# We'll create mock versions for testing
# ============================================================================
# FIXTURES
# ============================================================================
@pytest.fixture
def mock_db_connection():
"""Mock database connection for testing."""
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn.cursor.return_value = mock_cursor
mock_cursor.__enter__ = Mock(return_value=mock_cursor)
mock_cursor.__exit__ = Mock(return_value=False)
return mock_conn, mock_cursor
@pytest.fixture
def app():
"""Create FastAPI application instance for testing."""
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.responses import HTMLResponse, JSONResponse
import os
test_app = FastAPI(title="AIVA Voice Command Bridge - Test")
# Mock database state
test_app.state.db_connected = True
test_app.state.api_keys = {"test-api-key-12345": "test_client"}
test_app.state.directives = []
test_app.state.completed_history = []
test_app.state.health_status = {
"api_uptime_seconds": 86400,
"db_connected": True,
"poller_status": "running",
"websocket_clients": 2
}
test_app.state.start_time = datetime.now()
def get_db_connection():
mock_conn = MagicMock()
mock_cursor = MagicMock()
mock_conn.cursor.return_value = mock_cursor
return mock_conn
test_app.state.get_db_connection = get_db_connection
# Add routes
@test_app.middleware("http")
async def auth_middleware(request, call_next):
api_key = request.headers.get("X-API-Key")
if not api_key:
api_key = request.query_params.get("api_key")
# Skip auth for dashboard and health
if request.url.path in ["/bridge/dashboard", "/bridge/health", "/docs", "/openapi.json"]:
return await call_next(request)
if api_key not in test_app.state.api_keys:
return JSONResponse(
status_code=401,
content={"detail": "Invalid or missing API key"}
)
return await call_next(request)
@test_app.get("/bridge/dashboard")
async def get_dashboard():
"""Serve the mobile-friendly status dashboard."""
dashboard_html = '''
Genesis Command Bridge
Auto-refresh: 10s
'''
return HTMLResponse(content=dashboard_html)
@test_app.get("/bridge/status")
async def get_status():
"""Get current bridge status and queue metrics."""
return {
"pending": 3,
"processing": 1,
"completed_today": 47,
"failed_today": 2,
"timestamp": datetime.now().isoformat()
}
@test_app.get("/bridge/directives")
async def get_directives():
"""Get all active directives."""
return [
{
"id": "dir-001",
"command": "ACTIVATE_SECURITY",
"priority": 10,
"status": "pending",
"created_at": (datetime.now() - timedelta(seconds=45)).isoformat(),
"age_seconds": 45,
"source": "phone"
},
{
"id": "dir-002",
"command": "CHECK_STATUS",
"priority": 8,
"status": "processing",
"created_at": (datetime.now() - timedelta(seconds=120)).isoformat(),
"age_seconds": 120,
"source": "phone"
},
{
"id": "dir-003",
"command": "SET_MODE_HOME",
"priority": 5,
"status": "pending",
"created_at": (datetime.now() - timedelta(seconds=30)).isoformat(),
"age_seconds": 30,
"source": "dashboard"
}
]
@test_app.get("/bridge/history")
async def get_history(limit: int = 10):
"""Get recent completed directives."""
return [
{
"id": "hist-001",
"command": "LOCK_DOORS",
"completed_at": (datetime.now() - timedelta(minutes=5)).isoformat(),
"response_time_ms": 234,
"status": "completed"
},
{
"id": "hist-002",
"command": "TURN_ON_LIGHTS",
"completed_at": (datetime.now() - timedelta(minutes=12)).isoformat(),
"response_time_ms": 156,
"status": "completed"
}
]
@test_app.get("/bridge/health")
async def get_health():
"""Get system health status."""
return test_app.state.health_status
@test_app.post("/bridge/directive")
async def create_directive(request: dict):
"""Create a new directive."""
return {
"id": "dir-new-" + str(int(time.time())),
"command": request.get("command", "UNKNOWN"),
"priority": request.get("priority", 5),
"status": "pending",
"created_at": datetime.now().isoformat()
}
@test_app.post("/bridge/history/clear")
async def clear_history():
"""Clear completed history."""
return {"status": "cleared"}
@test_app.post("/bridge/poll")
async def force_poll():
"""Force a poll operation."""
return {"status": "poll_triggered"}
return test_app
@pytest.fixture
def client(app):
"""Create test client."""
return TestClient(app)
@pytest.fixture
def auth_headers():
"""Valid authentication headers."""
return {"X-API-Key": "test-api-key-12345"}
@pytest.fixture
def sample_directives():
"""Sample directive data for testing."""
return [
{
"id": "dir-001",
"command": "ACTIVATE_SECURITY",
"priority": 10,
"status": "pending",
"created_at": (datetime.now() - timedelta(seconds=45)).isoformat(),
"age_seconds": 45,
"source": "phone"
},
{
"id": "dir-002",
"command": "CHECK_STATUS",
"priority": 8,
"status": "processing",
"created_at": (datetime.now() - timedelta(seconds=120)).isoformat(),
"age_seconds": 120,
"source": "phone"
},
{
"id": "dir-003",
"command": "SET_MODE_HOME",
"priority": 5,
"status": "pending",
"created_at": (datetime.now() - timedelta(seconds=30)).isoformat(),
"age_seconds": 30,
"source": "dashboard"
},
{
"id": "dir-004",
"command": "GET_INFO",
"priority": 3,
"status": "pending",
"created_at": (datetime.now() - timedelta(seconds=10)).isoformat(),
"age_seconds": 10,
"source": "api"
}
]
@pytest.fixture
def sample_health():
"""Sample health data for testing."""
return {
"api_uptime_seconds": 86400,
"db_connected": True,
"poller_status": "running",
"websocket_clients": 2
}
# ============================================================================
# UNIT TESTS - Dashboard Endpoint
# ============================================================================
class TestDashboardEndpoint:
"""Tests for the /bridge/dashboard endpoint."""
def test_dashboard_returns_html(self, client):
"""Test that dashboard returns HTML content."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "text/html" in response.headers["content-type"]
def test_dashboard_contains_genesis_title(self, client):
"""Test that dashboard contains Genesis branding."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "Genesis" in response.text
def test_dashboard_contains_mobile_meta_tags(self, client):
"""Test that dashboard has mobile-friendly meta tags."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert 'viewport' in response.text
assert 'width=device-width' in response.text
def test_dashboard_contains_dark_theme_colors(self, client):
"""Test that dashboard uses dark theme colors."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "#0B0C0E" in response.text # Background
assert "#3B82F6" in response.text # Accent
def test_dashboard_contains_priority_styles(self, client):
"""Test that dashboard has priority-based styling."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "priority-high" in response.text
assert "priority-urgent" in response.text
assert "priority-medium" in response.text
assert "priority-low" in response.text
def test_dashboard_has_auto_refresh(self, client):
"""Test that dashboard has auto-refresh functionality."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "REFRESH_INTERVAL" in response.text
assert "setInterval" in response.text
def test_dashboard_has_action_buttons(self, client):
"""Test that dashboard has quick action buttons."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "sendTestDirective" in response.text
assert "clearCompleted" in response.text
assert "forcePoll" in response.text
def test_dashboard_has_api_key_handling(self, client):
"""Test that dashboard handles API key from URL or localStorage."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "localStorage" in response.text
assert "URLSearchParams" in response.text
def test_dashboard_has_pwa_support(self, client):
"""Test that dashboard has PWA-ready meta tags."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "apple-mobile-web-app-capable" in response.text
assert "theme-color" in response.text
# ============================================================================
# INTEGRATION TESTS - Status Endpoints
# ============================================================================
class TestStatusEndpoints:
"""Tests for the /bridge/status endpoint."""
def test_status_returns_queue_metrics(self, client, auth_headers):
"""Test that status endpoint returns queue metrics."""
response = client.get("/bridge/status", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert "pending" in data
assert "processing" in data
assert "completed_today" in data
assert "failed_today" in data
def test_status_requires_auth(self, client):
"""Test that status endpoint requires authentication."""
response = client.get("/bridge/status")
assert response.status_code == 401
def test_status_returns_json(self, client, auth_headers):
"""Test that status returns JSON content."""
response = client.get("/bridge/status", headers=auth_headers)
assert response.status_code == 200
assert "application/json" in response.headers["content-type"]
class TestDirectivesEndpoints:
"""Tests for the /bridge/directives endpoint."""
def test_directives_returns_list(self, client, auth_headers):
"""Test that directives endpoint returns a list."""
response = client.get("/bridge/directives", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
def test_directives_contains_required_fields(self, client, auth_headers):
"""Test that directives contain required fields."""
response = client.get("/bridge/directives", headers=auth_headers)
assert response.status_code == 200
data = response.json()
if len(data) > 0:
directive = data[0]
assert "command" in directive
assert "priority" in directive
assert "status" in directive
def test_directives_requires_auth(self, client):
"""Test that directives endpoint requires authentication."""
response = client.get("/bridge/directives")
assert response.status_code == 401
class TestHistoryEndpoints:
"""Tests for the /bridge/history endpoint."""
def test_history_returns_list(self, client, auth_headers):
"""Test that history endpoint returns a list."""
response = client.get("/bridge/history", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
def test_history_supports_limit_parameter(self, client, auth_headers):
"""Test that history supports limit parameter."""
response = client.get("/bridge/history?limit=5", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert isinstance(data, list)
def test_history_requires_auth(self, client):
"""Test that history endpoint requires authentication."""
response = client.get("/bridge/history")
assert response.status_code == 401
class TestHealthEndpoints:
"""Tests for the /bridge/health endpoint."""
def test_health_returns_status(self, client):
"""Test that health endpoint returns status."""
response = client.get("/bridge/health")
assert response.status_code == 200
data = response.json()
assert "api_uptime_seconds" in data
assert "db_connected" in data
assert "poller_status" in data
assert "websocket_clients" in data
def test_health_no_auth_required(self, client):
"""Test that health endpoint doesn't require auth."""
response = client.get("/bridge/health")
assert response.status_code == 200
# ============================================================================
# INTEGRATION TESTS - Action Endpoints
# ============================================================================
class TestActionEndpoints:
"""Tests for action endpoints (POST methods)."""
def test_create_directive(self, client, auth_headers):
"""Test creating a new directive."""
payload = {
"command": "TEST_COMMAND",
"priority": 7,
"source": "test"
}
response = client.post(
"/bridge/directive",
json=payload,
headers=auth_headers
)
assert response.status_code == 200
data = response.json()
assert data["command"] == "TEST_COMMAND"
assert data["priority"] == 7
def test_create_directive_requires_auth(self, client):
"""Test that directive creation requires auth."""
payload = {"command": "TEST", "priority": 5}
response = client.post("/bridge/directive", json=payload)
assert response.status_code == 401
def test_clear_history(self, client, auth_headers):
"""Test clearing history."""
response = client.post("/bridge/history/clear", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["status"] == "cleared"
def test_clear_history_requires_auth(self, client):
"""Test that history clear requires auth."""
response = client.post("/bridge/history/clear")
assert response.status_code == 401
def test_force_poll(self, client, auth_headers):
"""Test forcing a poll."""
response = client.post("/bridge/poll", headers=auth_headers)
assert response.status_code == 200
data = response.json()
assert data["status"] == "poll_triggered"
def test_force_poll_requires_auth(self, client):
"""Test that poll requires auth."""
response = client.post("/bridge/poll")
assert response.status_code == 401
# ============================================================================
# EDGE CASE TESTS
# ============================================================================
class TestEdgeCases:
"""Edge case and error handling tests."""
def test_invalid_api_key(self, client):
"""Test that invalid API key is rejected."""
response = client.get(
"/bridge/status",
headers={"X-API-Key": "invalid-key"}
)
assert response.status_code == 401
def test_empty_api_key_header(self, client):
"""Test that empty API key is rejected."""
response = client.get(
"/bridge/status",
headers={"X-API-Key": ""}
)
assert response.status_code == 401
def test_api_key_in_query_string(self, client):
"""Test that API key can be passed in query string."""
response = client.get("/bridge/status?api_key=test-api-key-12345")
assert response.status_code == 200
def test_dashboard_without_api_key_shows_error_state(self, client):
"""Test that dashboard shows error state without API key."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
# Dashboard should handle missing API key gracefully
assert "setConnected" in response.text
assert "fetchWithAuth" in response.text
def test_health_endpoint_db_disconnected(self, app, client):
"""Test health endpoint when DB is disconnected."""
app.state.health_status["db_connected"] = False
response = client.get("/bridge/health")
assert response.status_code == 200
data = response.json()
assert data["db_connected"] == False
def test_health_endpoint_poller_stopped(self, app, client):
"""Test health endpoint when poller is stopped."""
app.state.health_status["poller_status"] = "stopped"
response = client.get("/bridge/health")
assert response.status_code == 200
data = response.json()
assert data["poller_status"] == "stopped"
def test_health_endpoint_zero_websocket_clients(self, app, client):
"""Test health endpoint with zero WebSocket clients."""
app.state.health_status["websocket_clients"] = 0
response = client.get("/bridge/health")
assert response.status_code == 200
data = response.json()
assert data["websocket_clients"] == 0
# ============================================================================
# PRIORITY INDICATOR TESTS
# ============================================================================
class TestPriorityIndicators:
"""Tests for priority-based visual indicators."""
def test_priority_10_red_border(self, client):
"""Test that priority 10 has red border styling."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
# Should have urgentPulse animation for priority 10
assert "urgentPulse" in response.text
def test_priority_8_9_orange_border(self, client):
"""Test that priority 8-9 has orange border styling."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "priority-urgent" in response.text
def test_priority_6_7_yellow_border(self, client):
"""Test that priority 6-7 has yellow border styling."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "priority-medium" in response.text
def test_priority_4_5_normal_border(self, client):
"""Test that priority 4-5 has normal border styling."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
# Normal priority uses default border
assert "directive-card" in response.text
def test_priority_1_3_muted_border(self, client):
"""Test that priority 1-3 has muted/grey styling."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "priority-low" in response.text
# ============================================================================
# ANIMATION TESTS
# ============================================================================
class TestAnimations:
"""Tests for dashboard animations."""
def test_fade_in_animation_exists(self, client):
"""Test that fadeIn animation exists."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "@keyframes fadeIn" in response.text
def test_pulse_animation_exists(self, client):
"""Test that pulse animation exists for status dot."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "@keyframes pulse" in response.text
def test_transition_on_directive_card(self, client):
"""Test that directive cards have smooth transitions."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert "transition: all 0.3s ease" in response.text
# ============================================================================
# MOBILE RESPONSIVENESS TESTS
# ============================================================================
class TestMobileResponsiveness:
"""Tests for mobile-first responsive design."""
def test_viewport_meta_tag(self, client):
"""Test that viewport meta tag is present."""
response = client.get("/bridge/dashboard")
assert response.status_code == 200
assert '