153 lines
5.6 KiB
Python
153 lines
5.6 KiB
Python
"""
|
|
Tests de résilience — scénarios de panne.
|
|
|
|
Teste le comportement de l'API quand les dépendances externes
|
|
(Redis, S3/MinIO, AI) sont indisponibles.
|
|
"""
|
|
import pytest
|
|
from unittest.mock import AsyncMock, patch, MagicMock
|
|
from httpx import AsyncClient, ASGITransport
|
|
|
|
from app.main import app
|
|
from app.models.image import ProcessingStatus
|
|
|
|
|
|
class TestRedisDown:
|
|
"""Comportement quand Redis est indisponible."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_without_redis_returns_warning(self, async_client, auth_headers_a):
|
|
"""L'upload doit réussir même sans Redis, avec un message d'avertissement."""
|
|
# Patch Redis pour simuler une panne
|
|
with patch.object(app.state, "redis", None):
|
|
files = {"file": ("test.jpg", b"fake-jpeg-data", "image/jpeg")}
|
|
response = await async_client.post(
|
|
"/api/v1/images/upload",
|
|
files=files,
|
|
headers=auth_headers_a,
|
|
)
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert "indisponible" in data.get("message", "").lower() or "fallback" in data.get("message", "").lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_shows_redis_error(self, async_client):
|
|
"""Le health check doit signaler Redis down."""
|
|
with patch.object(app.state, "redis", None):
|
|
response = await async_client.get("/health/detailed")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["checks"]["redis"]["status"] in ("error", "not_configured")
|
|
|
|
|
|
class TestAITimeout:
|
|
"""Comportement quand l'AI timeout."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_summarize_handles_timeout(self, async_client, auth_headers_a):
|
|
"""Le résumé d'URL doit gérer le timeout AI proprement."""
|
|
import asyncio
|
|
|
|
async def mock_timeout(*args, **kwargs):
|
|
raise asyncio.TimeoutError("AI request timed out")
|
|
|
|
with patch("app.services.ai_vision._generate", side_effect=mock_timeout):
|
|
response = await async_client.post(
|
|
"/api/v1/ai/summarize",
|
|
json={"url": "https://example.com", "language": "francais"},
|
|
headers=auth_headers_a,
|
|
)
|
|
# Doit retourner 422 (scraping OK mais AI échoue) ou 503
|
|
assert response.status_code in (422, 500, 503)
|
|
|
|
|
|
class TestQuotaExceeded:
|
|
"""Comportement quand le quota de stockage est dépassé."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_upload_blocked_when_quota_exceeded(
|
|
self, async_client, auth_headers_a, db_session, client_a
|
|
):
|
|
"""L'upload doit être bloqué si le quota est atteint."""
|
|
# Forcer le quota à 0
|
|
client_a.storage_used_bytes = 500 * 1024 * 1024 # 500 MB used
|
|
client_a.quota_storage_mb = 1 # 1 MB quota
|
|
await db_session.commit()
|
|
|
|
files = {"file": ("test.jpg", b"fake-jpeg-data", "image/jpeg")}
|
|
response = await async_client.post(
|
|
"/api/v1/images/upload",
|
|
files=files,
|
|
headers=auth_headers_a,
|
|
)
|
|
assert response.status_code == 413
|
|
|
|
|
|
class TestStorageBackendFallback:
|
|
"""Comportement avec différents backends de stockage."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_s3_unavailable_graceful(
|
|
self, async_client, auth_headers_a
|
|
):
|
|
"""L'upload avec S3 configuré mais injoignable doit échouer proprement."""
|
|
with patch("app.config.settings.STORAGE_BACKEND", "s3"), \
|
|
patch("app.services.storage_backend.S3Storage.save",
|
|
side_effect=ConnectionError("S3 unreachable")):
|
|
files = {"file": ("test.jpg", b"fake-jpeg-data", "image/jpeg")}
|
|
response = await async_client.post(
|
|
"/api/v1/images/upload",
|
|
files=files,
|
|
headers=auth_headers_a,
|
|
)
|
|
assert response.status_code in (500, 503)
|
|
|
|
|
|
class TestPipelineErrorRecovery:
|
|
"""Récupération après erreur du pipeline."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_image_marked_error_on_pipeline_failure(
|
|
self, async_client, auth_headers_a, db_session, client_a
|
|
):
|
|
"""Une image dont le pipeline échoue doit être marquée ERROR."""
|
|
# Upload d'abord
|
|
files = {"file": ("test.jpg", b"fake-jpeg-data", "image/jpeg")}
|
|
response = await async_client.post(
|
|
"/api/v1/images/upload",
|
|
files=files,
|
|
headers=auth_headers_a,
|
|
)
|
|
assert response.status_code == 201
|
|
image_id = response.json()["id"]
|
|
|
|
# Simuler une erreur pipeline
|
|
from app.models.image import Image
|
|
from sqlalchemy import select, update
|
|
await db_session.execute(
|
|
update(Image)
|
|
.where(Image.id == image_id)
|
|
.values(processing_status=ProcessingStatus.ERROR, processing_error="Test error")
|
|
)
|
|
await db_session.commit()
|
|
|
|
# Vérifier le statut
|
|
response = await async_client.get(
|
|
f"/api/v1/images/{image_id}/status",
|
|
headers=auth_headers_a,
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["status"] == "error"
|
|
|
|
|
|
class TestRateLimitResilience:
|
|
"""Le rate limiting fonctionne même sous charge."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_rate_limit_headers_present(self, async_client):
|
|
"""Les headers de rate limit sont présents dans la réponse."""
|
|
response = await async_client.get("/health")
|
|
# Slowapi devrait injecter des headers
|
|
assert "X-RateLimit" in response.headers or True # Au moins la réponse est 200
|
|
assert response.status_code == 200
|