338 lines
12 KiB
Python
338 lines
12 KiB
Python
"""
|
|
Integration tests for WiFi-DensePose API endpoints.
|
|
|
|
Tests all REST API endpoints with real service dependencies.
|
|
"""
|
|
|
|
import pytest
|
|
import asyncio
|
|
from datetime import datetime, timedelta
|
|
from typing import Dict, Any
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
from fastapi.testclient import TestClient
|
|
from fastapi import FastAPI
|
|
import httpx
|
|
|
|
from src.api.dependencies import (
|
|
get_pose_service,
|
|
get_stream_service,
|
|
get_hardware_service,
|
|
get_current_user
|
|
)
|
|
from src.api.routers.health import router as health_router
|
|
from src.api.routers.pose import router as pose_router
|
|
from src.api.routers.stream import router as stream_router
|
|
|
|
|
|
class TestAPIEndpoints:
|
|
"""Integration tests for API endpoints."""
|
|
|
|
@pytest.fixture
|
|
def app(self):
|
|
"""Create FastAPI app with test dependencies."""
|
|
app = FastAPI()
|
|
app.include_router(health_router, prefix="/health", tags=["health"])
|
|
app.include_router(pose_router, prefix="/pose", tags=["pose"])
|
|
app.include_router(stream_router, prefix="/stream", tags=["stream"])
|
|
return app
|
|
|
|
@pytest.fixture
|
|
def mock_pose_service(self):
|
|
"""Mock pose service."""
|
|
service = AsyncMock()
|
|
service.health_check.return_value = {
|
|
"status": "healthy",
|
|
"message": "Service operational",
|
|
"uptime_seconds": 3600.0,
|
|
"metrics": {"processed_frames": 1000}
|
|
}
|
|
service.is_ready.return_value = True
|
|
service.estimate_poses.return_value = {
|
|
"timestamp": datetime.utcnow(),
|
|
"frame_id": "test-frame-001",
|
|
"persons": [],
|
|
"zone_summary": {"zone1": 0},
|
|
"processing_time_ms": 50.0,
|
|
"metadata": {}
|
|
}
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def mock_stream_service(self):
|
|
"""Mock stream service."""
|
|
service = AsyncMock()
|
|
service.health_check.return_value = {
|
|
"status": "healthy",
|
|
"message": "Stream service operational",
|
|
"uptime_seconds": 1800.0
|
|
}
|
|
service.is_ready.return_value = True
|
|
service.get_status.return_value = {
|
|
"is_active": True,
|
|
"active_streams": [],
|
|
"uptime_seconds": 1800.0
|
|
}
|
|
service.is_active.return_value = True
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def mock_hardware_service(self):
|
|
"""Mock hardware service."""
|
|
service = AsyncMock()
|
|
service.health_check.return_value = {
|
|
"status": "healthy",
|
|
"message": "Hardware connected",
|
|
"uptime_seconds": 7200.0,
|
|
"metrics": {"connected_routers": 3}
|
|
}
|
|
service.is_ready.return_value = True
|
|
return service
|
|
|
|
@pytest.fixture
|
|
def mock_user(self):
|
|
"""Mock authenticated user."""
|
|
return {
|
|
"id": "test-user-001",
|
|
"username": "testuser",
|
|
"email": "test@example.com",
|
|
"is_admin": False,
|
|
"is_active": True,
|
|
"permissions": ["read", "write"]
|
|
}
|
|
|
|
@pytest.fixture
|
|
def client(self, app, mock_pose_service, mock_stream_service, mock_hardware_service, mock_user):
|
|
"""Create test client with mocked dependencies."""
|
|
app.dependency_overrides[get_pose_service] = lambda: mock_pose_service
|
|
app.dependency_overrides[get_stream_service] = lambda: mock_stream_service
|
|
app.dependency_overrides[get_hardware_service] = lambda: mock_hardware_service
|
|
app.dependency_overrides[get_current_user] = lambda: mock_user
|
|
|
|
with TestClient(app) as client:
|
|
yield client
|
|
|
|
def test_health_check_endpoint_should_fail_initially(self, client):
|
|
"""Test health check endpoint - should fail initially."""
|
|
# This test should fail because we haven't implemented the endpoint properly
|
|
response = client.get("/health/health")
|
|
|
|
# This assertion will fail initially, driving us to implement the endpoint
|
|
assert response.status_code == 200
|
|
assert "status" in response.json()
|
|
assert "components" in response.json()
|
|
assert "system_metrics" in response.json()
|
|
|
|
def test_readiness_check_endpoint_should_fail_initially(self, client):
|
|
"""Test readiness check endpoint - should fail initially."""
|
|
response = client.get("/health/ready")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "ready" in data
|
|
assert "checks" in data
|
|
assert isinstance(data["checks"], dict)
|
|
|
|
def test_liveness_check_endpoint_should_fail_initially(self, client):
|
|
"""Test liveness check endpoint - should fail initially."""
|
|
response = client.get("/health/live")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "status" in data
|
|
assert data["status"] == "alive"
|
|
|
|
def test_version_info_endpoint_should_fail_initially(self, client):
|
|
"""Test version info endpoint - should fail initially."""
|
|
response = client.get("/health/version")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "name" in data
|
|
assert "version" in data
|
|
assert "environment" in data
|
|
|
|
def test_pose_current_endpoint_should_fail_initially(self, client):
|
|
"""Test current pose estimation endpoint - should fail initially."""
|
|
response = client.get("/pose/current")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "timestamp" in data
|
|
assert "frame_id" in data
|
|
assert "persons" in data
|
|
assert "zone_summary" in data
|
|
|
|
def test_pose_analyze_endpoint_should_fail_initially(self, client):
|
|
"""Test pose analysis endpoint - should fail initially."""
|
|
request_data = {
|
|
"zone_ids": ["zone1", "zone2"],
|
|
"confidence_threshold": 0.7,
|
|
"max_persons": 10,
|
|
"include_keypoints": True,
|
|
"include_segmentation": False
|
|
}
|
|
|
|
response = client.post("/pose/analyze", json=request_data)
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "timestamp" in data
|
|
assert "persons" in data
|
|
|
|
def test_zone_occupancy_endpoint_should_fail_initially(self, client):
|
|
"""Test zone occupancy endpoint - should fail initially."""
|
|
response = client.get("/pose/zones/zone1/occupancy")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "zone_id" in data
|
|
assert "current_occupancy" in data
|
|
|
|
def test_zones_summary_endpoint_should_fail_initially(self, client):
|
|
"""Test zones summary endpoint - should fail initially."""
|
|
response = client.get("/pose/zones/summary")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "total_persons" in data
|
|
assert "zones" in data
|
|
|
|
def test_stream_status_endpoint_should_fail_initially(self, client):
|
|
"""Test stream status endpoint - should fail initially."""
|
|
response = client.get("/stream/status")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "is_active" in data
|
|
assert "connected_clients" in data
|
|
|
|
def test_stream_start_endpoint_should_fail_initially(self, client):
|
|
"""Test stream start endpoint - should fail initially."""
|
|
response = client.post("/stream/start")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "message" in data
|
|
|
|
def test_stream_stop_endpoint_should_fail_initially(self, client):
|
|
"""Test stream stop endpoint - should fail initially."""
|
|
response = client.post("/stream/stop")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "message" in data
|
|
|
|
|
|
class TestAPIErrorHandling:
|
|
"""Test API error handling scenarios."""
|
|
|
|
@pytest.fixture
|
|
def app_with_failing_services(self):
|
|
"""Create app with failing service dependencies."""
|
|
app = FastAPI()
|
|
app.include_router(health_router, prefix="/health", tags=["health"])
|
|
app.include_router(pose_router, prefix="/pose", tags=["pose"])
|
|
|
|
# Mock failing services
|
|
failing_pose_service = AsyncMock()
|
|
failing_pose_service.health_check.side_effect = Exception("Service unavailable")
|
|
|
|
app.dependency_overrides[get_pose_service] = lambda: failing_pose_service
|
|
|
|
return app
|
|
|
|
def test_health_check_with_failing_service_should_fail_initially(self, app_with_failing_services):
|
|
"""Test health check with failing service - should fail initially."""
|
|
with TestClient(app_with_failing_services) as client:
|
|
response = client.get("/health/health")
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "unhealthy"
|
|
assert "hardware" in data["components"]
|
|
assert data["components"]["pose"]["status"] == "unhealthy"
|
|
|
|
|
|
class TestAPIAuthentication:
|
|
"""Test API authentication scenarios."""
|
|
|
|
@pytest.fixture
|
|
def app_with_auth(self):
|
|
"""Create app with authentication enabled."""
|
|
app = FastAPI()
|
|
app.include_router(pose_router, prefix="/pose", tags=["pose"])
|
|
|
|
# Mock authenticated user dependency
|
|
def get_authenticated_user():
|
|
return {
|
|
"id": "auth-user-001",
|
|
"username": "authuser",
|
|
"is_admin": True,
|
|
"permissions": ["read", "write", "admin"]
|
|
}
|
|
|
|
app.dependency_overrides[get_current_user] = get_authenticated_user
|
|
|
|
return app
|
|
|
|
def test_authenticated_endpoint_access_should_fail_initially(self, app_with_auth):
|
|
"""Test authenticated endpoint access - should fail initially."""
|
|
with TestClient(app_with_auth) as client:
|
|
response = client.post("/pose/analyze", json={
|
|
"confidence_threshold": 0.8,
|
|
"include_keypoints": True
|
|
})
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 200
|
|
|
|
|
|
class TestAPIValidation:
|
|
"""Test API request validation."""
|
|
|
|
@pytest.fixture
|
|
def validation_app(self):
|
|
"""Create app for validation testing."""
|
|
app = FastAPI()
|
|
app.include_router(pose_router, prefix="/pose", tags=["pose"])
|
|
|
|
# Mock service
|
|
mock_service = AsyncMock()
|
|
app.dependency_overrides[get_pose_service] = lambda: mock_service
|
|
|
|
return app
|
|
|
|
def test_invalid_confidence_threshold_should_fail_initially(self, validation_app):
|
|
"""Test invalid confidence threshold validation - should fail initially."""
|
|
with TestClient(validation_app) as client:
|
|
response = client.post("/pose/analyze", json={
|
|
"confidence_threshold": 1.5, # Invalid: > 1.0
|
|
"include_keypoints": True
|
|
})
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 422
|
|
assert "validation error" in response.json()["detail"][0]["msg"].lower()
|
|
|
|
def test_invalid_max_persons_should_fail_initially(self, validation_app):
|
|
"""Test invalid max_persons validation - should fail initially."""
|
|
with TestClient(validation_app) as client:
|
|
response = client.post("/pose/analyze", json={
|
|
"max_persons": 0, # Invalid: < 1
|
|
"include_keypoints": True
|
|
})
|
|
|
|
# This will fail initially
|
|
assert response.status_code == 422 |