File size: 7,154 Bytes
23d337e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 | """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)
|