"""Integration tests — end-to-end through the FastAPI TestClient.""" from __future__ import annotations import base64 import pytest from fastapi.testclient import TestClient from api.main import create_app from config.settings import Settings @pytest.fixture def client(test_settings): app = create_app(test_settings) with TestClient(app) as c: yield c class TestHealthEndpoints: def test_health_root(self, client): r = client.get("/health") assert r.status_code == 200 assert r.json()["status"] == "ok" def test_health_live(self, client): r = client.get("/health/live") assert r.status_code == 200 assert r.json()["status"] == "alive" def test_health_ready(self, client): r = client.get("/health/ready") assert r.status_code == 200 def test_health_providers(self, client): r = client.get("/health/providers") assert r.status_code == 200 data = r.json() assert "status" in data assert "providers" in data assert isinstance(data["providers"], list) class TestStatsEndpoint: def test_stats_returns_metrics(self, client): r = client.get("/stats") assert r.status_code == 200 data = r.json() assert "providers" in data assert "timings" in data assert "counters" in data assert "health" in data class TestProvidersEndpoint: def test_list_providers(self, client): r = client.get("/providers") assert r.status_code == 200 data = r.json() assert "providers" in data assert isinstance(data["providers"], list) # At minimum, haar + image_quality + image_properties + exif + # image_integrity + duplicate_detector should be available names = [p["name"] for p in data["providers"] if p["available"]] assert "haar" in names assert "image_quality" in names assert "exif" in names assert "image_integrity" in names def test_get_single_provider(self, client): r = client.get("/providers/haar") assert r.status_code == 200 data = r.json() assert data["name"] == "haar" def test_get_unknown_provider_returns_404(self, client): r = client.get("/providers/nonexistent") assert r.status_code == 404 class TestCacheEndpoint: def test_cache_stats(self, client): r = client.get("/cache") assert r.status_code == 200 data = r.json() assert "entries" in data assert "hits" in data assert "misses" in data def test_cache_clear(self, client): r = client.delete("/cache") assert r.status_code == 200 assert "cleared" in r.json() class TestFacesDetect: def test_detect_with_base64(self, client, sample_image_b64): r = client.post("/faces/detect", json={"image_base64": sample_image_b64}) assert r.status_code == 200 data = r.json() assert data["success"] is True assert "report" in data assert "elapsed_ms" in data def test_detect_with_no_input_returns_error(self, client): r = client.post("/faces/detect", json={}) assert r.status_code == 200 # service returns 200 with success=False data = r.json() assert data["success"] is False assert "error" in data def test_detect_with_invalid_base64(self, client): r = client.post("/faces/detect", json={"image_base64": "!!!invalid!!!"}) assert r.status_code == 200 data = r.json() assert data["success"] is False class TestAnalysisEndpoints: def test_image_analysis(self, client, sample_image_b64): r = client.post("/analysis/image", json={"image_base64": sample_image_b64}) assert r.status_code == 200 data = r.json() assert data["success"] is True assert "image_analyses" in data["report"] assert len(data["report"]["image_analyses"]) > 0 def test_metadata_extraction(self, client, sample_image_b64): r = client.post("/analysis/metadata", json={"image_base64": sample_image_b64}) assert r.status_code == 200 data = r.json() assert data["success"] is True assert "metadata_extractions" in data["report"] def test_forensics_analysis(self, client, sample_image_b64): r = client.post("/analysis/forensics", json={"image_base64": sample_image_b64}) assert r.status_code == 200 data = r.json() assert data["success"] is True assert "forensics" in data["report"] assert len(data["report"]["forensics"]) > 0 class TestJobsEndpoint: def test_create_and_get_job(self, client, sample_image_b64): # Create r = client.post("/jobs", json={ "kind": "detection", "image_base64": sample_image_b64, }) assert r.status_code == 200 data = r.json() assert data["status"] == "completed" job_id = data["job_id"] # Fetch r2 = client.get(f"/jobs/{job_id}") assert r2.status_code == 200 assert r2.json()["id"] == job_id def test_list_jobs(self, client, sample_image_b64): # Create one job client.post("/jobs", json={"kind": "detection", "image_base64": sample_image_b64}) # List r = client.get("/jobs") assert r.status_code == 200 data = r.json() assert "jobs" in data assert len(data["jobs"]) > 0 def test_get_unknown_job_404(self, client): r = client.get("/jobs/nonexistent") assert r.status_code == 404 def test_export_job(self, client, sample_image_b64): # Create r = client.post("/jobs", json={ "kind": "detection", "image_base64": sample_image_b64, }) job_id = r.json()["job_id"] # Export r2 = client.get(f"/export/{job_id}") assert r2.status_code == 200 assert "job" in r2.text class TestErrorHandling: def test_unknown_route_returns_404(self, client): r = client.get("/nonexistent-route") assert r.status_code == 404 def test_rate_limit_returns_429(self, test_settings): """If we send >rate_limit_per_minute requests, we should get 429.""" test_settings.rate_limit_per_minute = 2 app = create_app(test_settings) with TestClient(app) as c: # Send 3 requests (limit is 2) c.get("/stats") c.get("/stats") r = c.get("/stats") assert r.status_code == 429 def test_oversized_body_rejected(self, client): """Bodies over max_request_body_bytes should be rejected.""" # Create a body that's clearly over the limit # Default is 25 MB, so we need a much larger payload — but we can # also test the content-length header path big_b64 = "A" * 30 * 1024 * 1024 # 30 MB of base64 r = client.post("/faces/detect", json={"image_base64": big_b64}) # Either rejected by size limit OR by validation assert r.status_code in (413, 200)