26bd096a06
Major changes across backend, frontend, infrastructure: - Provider system with model selection (Google, DeepL, OpenAI, Ollama, Google Cloud) - Admin panel: user management, pricing, settings - Glossary system with CSV import/export - Subscription and tier quota management - Security hardening (rate limiting, API key auth, path traversal fixes) - Docker compose for dev, prod, and IONOS deployment - Alembic migrations for new tables - Frontend: dashboard, pricing page, landing page, i18n (en/fr) - Test suite and verification scripts Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>
894 lines
31 KiB
Python
894 lines
31 KiB
Python
"""
|
|
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",
|
|
'<?xml version="1.0"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"></Types>',
|
|
)
|
|
zf.writestr(
|
|
"_rels/.rels",
|
|
'<?xml version="1.0"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"></Relationships>',
|
|
)
|
|
zf.writestr(
|
|
"xl/workbook.xml",
|
|
'<?xml version="1.0"?><workbook xmlns="http://schemas.openxmlformats.org/spreadsheetml/2006/main"></workbook>',
|
|
)
|
|
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",
|
|
'<?xml version="1.0"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"></Types>',
|
|
)
|
|
zf.writestr(
|
|
"_rels/.rels",
|
|
'<?xml version="1.0"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"></Relationships>',
|
|
)
|
|
zf.writestr(
|
|
"word/document.xml",
|
|
'<?xml version="1.0"?><w:document xmlns:w="http://schemas.openxmlformats.org/wordprocessingml/2006/main"></w:document>',
|
|
)
|
|
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",
|
|
'<?xml version="1.0"?><Types xmlns="http://schemas.openxmlformats.org/package/2006/content-types"></Types>',
|
|
)
|
|
zf.writestr(
|
|
"_rels/.rels",
|
|
'<?xml version="1.0"?><Relationships xmlns="http://schemas.openxmlformats.org/package/2006/relationships"></Relationships>',
|
|
)
|
|
zf.writestr(
|
|
"ppt/presentation.xml",
|
|
'<?xml version="1.0"?><p:presentation xmlns:p="http://schemas.openxmlformats.org/presentationml/2006/main"></p:presentation>',
|
|
)
|
|
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]
|