""" Tests pour POST /api/v1/translate Couvre les AC 1-10 de la story 2.10 : Endpoint POST /api/v1/translate (Core) """ import io import time from pathlib import Path from unittest.mock import patch, MagicMock, AsyncMock from zipfile import ZipFile import pytest from fastapi.testclient import TestClient TRANSLATE_URL = "/api/v1/translate" STATUS_URL = "/api/v1/translations" REGISTER_URL = "/api/v1/auth/register" LOGIN_URL = "/api/v1/auth/login" VALID_USER = { "email": "translate@example.com", "password": "Password123!", "name": "Translate User", } def create_valid_excel() -> bytes: """Create a minimal valid .xlsx file (ZIP with office content).""" buf = io.BytesIO() with ZipFile(buf, "w") as zf: zf.writestr( "[Content_Types].xml", '', ) zf.writestr( "_rels/.rels", '', ) zf.writestr( "xl/workbook.xml", '', ) buf.seek(0) return buf.read() def create_valid_docx() -> bytes: """Create a minimal valid .docx file.""" buf = io.BytesIO() with ZipFile(buf, "w") as zf: zf.writestr( "[Content_Types].xml", '', ) zf.writestr( "_rels/.rels", '', ) zf.writestr( "word/document.xml", '', ) buf.seek(0) return buf.read() def create_valid_pptx() -> bytes: """Create a minimal valid .pptx file.""" buf = io.BytesIO() with ZipFile(buf, "w") as zf: zf.writestr( "[Content_Types].xml", '', ) zf.writestr( "_rels/.rels", '', ) zf.writestr( "ppt/presentation.xml", '', ) buf.seek(0) return buf.read() def create_invalid_file() -> bytes: """Create an invalid file (not a ZIP/Office document).""" return b"This is not a valid office document" @pytest.fixture() def users_file(tmp_path: Path) -> Path: """Fichier de stockage JSON isole pour les tests.""" return tmp_path / "users.json" @pytest.fixture() def client(users_file: Path, monkeypatch): """TestClient avec stockage JSON isole et rate limiting desactive.""" import services.auth_service as auth_svc monkeypatch.setattr(auth_svc, "USERS_FILE", users_file) monkeypatch.setattr(auth_svc, "USE_DATABASE", False) monkeypatch.setattr(auth_svc, "DATABASE_AVAILABLE", False) from middleware.rate_limiting import RateLimitManager async def _check_request_allow(self, request): return True, "ok", "test" async def _check_translation_allow(self, request, file_size_mb=0): return True, "ok" monkeypatch.setattr(RateLimitManager, "check_request", _check_request_allow) monkeypatch.setattr(RateLimitManager, "check_translation", _check_translation_allow) from middleware.tier_quota import TierQuotaService async def _check_quota_allow(self, user_id, tier): from middleware.tier_quota import QuotaResult from datetime import datetime, timezone, timedelta now = datetime.now(timezone.utc) tomorrow = now.date() + timedelta(days=1) reset_at = datetime( tomorrow.year, tomorrow.month, tomorrow.day, tzinfo=timezone.utc ) return QuotaResult( allowed=True, remaining=5, reset_at_utc=reset_at, current_usage=0, limit=5 ) async def _increment_noop(self, user_id): pass monkeypatch.setattr(TierQuotaService, "check_quota", _check_quota_allow) monkeypatch.setattr(TierQuotaService, "increment_on_success", _increment_noop) from main import app return TestClient(app, raise_server_exceptions=True) @pytest.fixture() def authenticated_client(client): """Client avec un utilisateur enregistre et authentifie.""" client.post(REGISTER_URL, json=VALID_USER) response = client.post( LOGIN_URL, json={ "email": VALID_USER["email"], "password": VALID_USER["password"], }, ) token = response.json()["data"]["access_token"] client.headers["Authorization"] = f"Bearer {token}" return client # --------------------------------------------------------------------------- # AC2: File Upload # --------------------------------------------------------------------------- class TestFileUpload: """AC2: POST to /api/v1/translate accepts multipart/form-data""" def test_accepts_multipart_form_data(self, authenticated_client): """Endpoint accepts multipart/form-data with file, source_lang, target_lang""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 def test_requires_file_or_url(self, client): """Returns error if neither file nor file_url provided""" response = client.post( TRANSLATE_URL, data={"target_lang": "fr"}, ) assert response.status_code == 400 body = response.json() assert body["error"] == "MISSING_FILE" def test_accepts_source_and_target_lang(self, authenticated_client): """Accepts source_lang and target_lang parameters""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"source_lang": "en", "target_lang": "fr"}, ) assert response.status_code == 202 body = response.json() assert body["data"]["source_lang"] == "en" assert body["data"]["target_lang"] == "fr" # --------------------------------------------------------------------------- # AC3 & AC5: File Validation # --------------------------------------------------------------------------- class TestFileValidation: """AC3, AC5: System validates format (xlsx/docx/pptx only), max size 50MB, magic bytes""" def test_rejects_invalid_format_pdf(self, authenticated_client): """AC5: Unsupported formats return 400 with INVALID_FORMAT""" invalid_content = create_invalid_file() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ("test.pdf", io.BytesIO(invalid_content), "application/pdf") }, data={"target_lang": "fr"}, ) assert response.status_code == 400 body = response.json() assert body["error"] == "INVALID_FORMAT" def test_rejects_invalid_magic_bytes(self, authenticated_client): """AC3/AC4: Checks magic bytes, returns CORRUPTED_FILE for invalid content""" invalid_content = create_invalid_file() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "fake.xlsx", io.BytesIO(invalid_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 400 body = response.json() assert body["error"] == "CORRUPTED_FILE" def test_accepts_xlsx(self, authenticated_client): """Accepts .xlsx files""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 def test_accepts_docx(self, authenticated_client): """Accepts .docx files""" docx_content = create_valid_docx() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.docx", io.BytesIO(docx_content), "application/vnd.openxmlformats-officedocument.wordprocessingml.document", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 def test_accepts_pptx(self, authenticated_client): """Accepts .pptx files""" pptx_content = create_valid_pptx() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.pptx", io.BytesIO(pptx_content), "application/vnd.openxmlformats-officedocument.presentationml.presentation", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 def test_error_includes_accepted_formats(self, authenticated_client): """AC5: Error includes accepted formats list""" invalid_content = create_invalid_file() response = authenticated_client.post( TRANSLATE_URL, files={"file": ("test.txt", io.BytesIO(invalid_content), "text/plain")}, data={"target_lang": "fr"}, ) body = response.json() assert "accepted_formats" in body.get("details", {}) or ".xlsx" in str(body) # --------------------------------------------------------------------------- # AC7: File Too Large # --------------------------------------------------------------------------- class TestFileTooLarge: """AC7: Files > 50MB return 413 with FILE_TOO_LARGE""" def test_returns_413_for_large_file(self, authenticated_client, monkeypatch): """Files exceeding max size return 413""" from middleware.validation import FileValidator # Create a validator with very small limit for testing small_validator = FileValidator( max_size_mb=0.001, allowed_extensions={".xlsx", ".docx", ".pptx"} ) monkeypatch.setattr("routes.translate_routes.file_validator", small_validator) large_content = b"x" * 2000 # 2KB > 1KB limit response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "large.xlsx", io.BytesIO(large_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 413 body = response.json() assert body["error"] == "FILE_TOO_LARGE" # --------------------------------------------------------------------------- # AC4: Success Response # --------------------------------------------------------------------------- class TestSuccessResponse: """AC4: Valid requests return HTTP 202 with proper format""" def test_returns_202_on_success(self, authenticated_client): """Returns HTTP 202 Accepted""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 def test_response_has_data_with_id(self, authenticated_client): """Response contains data.id""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) body = response.json() assert "data" in body assert "id" in body["data"] assert body["data"]["id"].startswith("tr_") def test_response_has_status_processing(self, authenticated_client): """Response contains status 'processing'""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) body = response.json() assert body["data"]["status"] == "processing" def test_response_has_meta_with_rate_limit(self, authenticated_client): """Response contains meta.rate_limit_remaining""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) body = response.json() assert "meta" in body assert "rate_limit_remaining" in body["meta"] # --------------------------------------------------------------------------- # AC1: Authentication # --------------------------------------------------------------------------- class TestAuthentication: """AC1: Endpoint requires valid JWT token or X-API-Key""" def test_works_with_jwt_token(self, authenticated_client): """Accepts JWT Bearer token""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 def test_works_without_auth(self, client): """Allows unauthenticated requests (but with tier limits)""" excel_content = create_valid_excel() response = client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) # Should still work without auth assert response.status_code == 202 # --------------------------------------------------------------------------- # AC6: Quota Exceeded # --------------------------------------------------------------------------- class TestQuotaExceeded: """AC6: Users exceeding tier limit return 429""" def test_returns_429_when_quota_exceeded(self, client, monkeypatch): """Returns 429 with QUOTA_EXCEEDED when quota exceeded""" from middleware.tier_quota import TierQuotaService, QuotaResult from datetime import datetime, timezone, timedelta async def _check_quota_denied(self, user_id, tier): now = datetime.now(timezone.utc) tomorrow = now.date() + timedelta(days=1) reset_at = datetime( tomorrow.year, tomorrow.month, tomorrow.day, tzinfo=timezone.utc ) return QuotaResult( allowed=False, remaining=0, reset_at_utc=reset_at, current_usage=5, limit=5, ) monkeypatch.setattr(TierQuotaService, "check_quota", _check_quota_denied) # Register and login client.post(REGISTER_URL, json=VALID_USER) response = client.post( LOGIN_URL, json={ "email": VALID_USER["email"], "password": VALID_USER["password"], }, ) token = response.json()["data"]["access_token"] client.headers["Authorization"] = f"Bearer {token}" excel_content = create_valid_excel() response = client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 429 body = response.json() # HTTPException returns detail dict with error field assert ( body.get("error") == "QUOTA_EXCEEDED" or body.get("detail", {}).get("error") == "QUOTA_EXCEEDED" ) def test_includes_retry_after_header(self, client, monkeypatch): """Includes Retry-After header on 429""" from middleware.tier_quota import TierQuotaService, QuotaResult from datetime import datetime, timezone, timedelta async def _check_quota_denied(self, user_id, tier): now = datetime.now(timezone.utc) tomorrow = now.date() + timedelta(days=1) reset_at = datetime( tomorrow.year, tomorrow.month, tomorrow.day, tzinfo=timezone.utc ) return QuotaResult( allowed=False, remaining=0, reset_at_utc=reset_at, current_usage=5, limit=5, ) monkeypatch.setattr(TierQuotaService, "check_quota", _check_quota_denied) client.post(REGISTER_URL, json=VALID_USER) response = client.post( LOGIN_URL, json={ "email": VALID_USER["email"], "password": VALID_USER["password"], }, ) token = response.json()["data"]["access_token"] client.headers["Authorization"] = f"Bearer {token}" excel_content = create_valid_excel() response = client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert "retry-after" in response.headers # --------------------------------------------------------------------------- # AC10: Optional Parameters # --------------------------------------------------------------------------- class TestOptionalParameters: """AC10: Support mode, provider, webhook_url, glossary_id, custom_prompt""" def test_accepts_mode_classic(self, authenticated_client): """Accepts mode='classic'""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "mode": "classic"}, ) assert response.status_code == 202 def test_accepts_mode_llm(self, authenticated_client): """Accepts mode='llm'""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "mode": "llm"}, ) assert response.status_code == 202 def test_accepts_webhook_url(self, authenticated_client): """Accepts webhook_url parameter""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "webhook_url": "https://example.com/webhook"}, ) assert response.status_code == 202 # --------------------------------------------------------------------------- # AC8: Async Processing # --------------------------------------------------------------------------- class TestAsyncProcessing: """AC8: Translation is processed asynchronously""" def test_returns_immediately_with_job_id(self, authenticated_client): """Endpoint returns 202 immediately with job ID""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) assert response.status_code == 202 body = response.json() assert body["data"]["status"] == "processing" assert body["data"]["id"].startswith("tr_") def test_can_check_job_status(self, authenticated_client): """Can check job status via GET /api/v1/translations/{id}""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, ) job_id = response.json()["data"]["id"] status_response = authenticated_client.get(f"{STATUS_URL}/{job_id}") assert status_response.status_code == 200 body = status_response.json() assert "data" in body assert body["data"]["id"] == job_id def test_returns_404_for_unknown_job(self, authenticated_client): """Returns 404 for unknown job ID""" response = authenticated_client.get(f"{STATUS_URL}/tr_unknown123") assert response.status_code == 404 # --------------------------------------------------------------------------- # AC9: URL Ingestion (Pro) # --------------------------------------------------------------------------- class TestURLIngestion: """AC9: Pro users can provide file_url parameter instead of file upload""" def test_pro_feature_requires_pro_tier(self, authenticated_client, monkeypatch): """file_url is a Pro-only feature""" from models.subscription import User, PlanType from datetime import datetime # User is free tier by default, should get 403 excel_url = "https://example.com/test.xlsx" response = authenticated_client.post( TRANSLATE_URL, data={"target_lang": "fr", "file_url": excel_url}, ) assert response.status_code == 403 body = response.json() assert body["error"] == "PRO_FEATURE_REQUIRED" def test_glossary_requires_pro(self, authenticated_client): """glossary_id is a Pro-only feature""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "glossary_id": "some-glossary-id"}, ) assert response.status_code == 403 body = response.json() assert body["error"] == "PRO_FEATURE_REQUIRED" def test_custom_prompt_requires_pro(self, authenticated_client): """custom_prompt is a Pro-only feature""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "custom_prompt": "Translate formally"}, ) assert response.status_code == 403 body = response.json() assert body["error"] == "PRO_FEATURE_REQUIRED" # --------------------------------------------------------------------------- # Additional Tests for Code Review Issues # --------------------------------------------------------------------------- class TestProviderParameter: """AC10: Provider parameter support""" def test_accepts_provider_google(self, authenticated_client): """Accepts provider='google'""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "provider": "google"}, ) assert response.status_code == 202 def test_accepts_provider_ollama(self, authenticated_client): """Accepts provider='ollama'""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "provider": "ollama"}, ) assert response.status_code == 202 class TestSourceLangValidation: """Source language validation""" def test_invalid_source_lang_returns_400(self, authenticated_client): """Invalid source_lang returns 400""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "source_lang": "invalid_code_xyz"}, ) assert response.status_code == 400 body = response.json() assert body["error"] == "INVALID_FORMAT" class TestWebhookValidation: """Webhook URL validation""" def test_invalid_webhook_url_returns_400(self, authenticated_client): """Invalid webhook_url returns 400""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "webhook_url": "not-a-valid-url"}, ) assert response.status_code == 400 body = response.json() assert body["error"] == "INVALID_WEBHOOK_URL" def test_valid_webhook_url_accepted(self, authenticated_client): """Valid webhook_url is accepted""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "webhook_url": "https://example.com/webhook"}, ) assert response.status_code == 202 class TestNoHTTP500: """NFR12: Zero HTTP 500 - all errors should be 4xx""" def test_unexpected_error_returns_400_not_500( self, authenticated_client, monkeypatch ): """Unexpected errors return 400, not 500""" from routes import translate_routes async def _failing_validate(*args, **kwargs): raise RuntimeError("Unexpected error") monkeypatch.setattr( translate_routes.file_validator, "validate_async", _failing_validate ) response = authenticated_client.post( TRANSLATE_URL, files={"file": ("test.xlsx", io.BytesIO(b"fake"), "application/vnd...")}, data={"target_lang": "fr"}, ) assert response.status_code in [400, 413, 401, 403, 429] class TestAPIKeyAuth: """AC1: X-API-Key header authentication""" def test_api_key_auth_placeholder(self, client): """X-API-Key header is accepted (placeholder test)""" excel_content = create_valid_excel() response = client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr"}, headers={"X-API-Key": "test-api-key-placeholder"}, ) assert response.status_code in [202, 401] class TestTranslateImagesParameter: """Test translate_images parameter in POST /api/v1/translate""" def test_accepts_translate_images_parameter(self, authenticated_client): """Endpoint accepts translate_images form parameter""" excel_content = create_valid_excel() response = authenticated_client.post( TRANSLATE_URL, files={ "file": ( "test.xlsx", io.BytesIO(excel_content), "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet", ) }, data={"target_lang": "fr", "translate_images": "true"}, ) assert response.status_code == 202