feat(translation): quality pipeline overhaul + new features (audit 2026-08-29)
All checks were successful
Deploy to Production / Build and Deploy (push) Successful in 2m20s

Translation quality & format preservation:
- Word: merge adjacent same-format runs into one unit (sentence-level
  coherence like inline-tag handling); translate comments/balloons;
  dedupe textbox collection (was translated twice); RTL no longer
  overrides center/justify alignment; CJK/Arabic font hints (eastAsia/cs)
- PPTX: chart translations now actually reach the output file
  (ChartPart.blob is read-only — rewrite chart XML in the saved ZIP);
  CJK typeface hints (a:ea)
- Excel: sheet renames no longer break references — rewrite cell
  formulas (3D/quoted), defined names, data validations, cond. formats
- PDF: bold/italic honored (hebo/heit/hebi); table cells never merge;
  unchanged blocks left untouched (typography preserved, fixes duplicate
  hyperlinks); attempted/changed stats + route gate now cover PDF;
  CJK font paths; scanned PDFs via Mistral OCR (detection + admin settings)

Features:
- formality param (formal/informal) + automatic regional-variant prompts
- output_mode=bilingual docx (source above translation)
- per-user translation memory on Redis (falls back to LRU), context-hashed
- QA report + 0-100 confidence score in job status; L0 on by default
- OpenAI-compatible providers: whole chunk in ONE numbered-JSON request
  (~15x fewer calls) with per-item fallback; base prompt always present
  (custom prompt no longer replaces translation instructions)

Infra & marketing alignment:
- plan-based engine gating + vision gating (closes paid-engine leak);
  /providers/available filtered per plan; 107 languages exposed
- zh-CN/zh-TW validation fixed; libmagic disabled on Windows (native crash)
- admin: Mistral OCR settings + engine status dashboard; httpx<0.28 pin
  (TestClient breakage); Prometheus test fixture fixed
- marketing docs aligned with code (PDF+OCR, retention, engines, pricing)
- security: .env.ionos/.env.production/provider_settings.json removed

Tests: 1173 passed / 0 failed (6 network tests deselected: free Google
endpoint temporarily blocked from this machine)
This commit is contained in:
2026-08-29 18:38:09 +02:00
parent 992f13d53c
commit 526c87348f
87 changed files with 6996 additions and 1024 deletions

View File

@@ -0,0 +1,228 @@
"""Tests for security fixes C1C4 (audit 2026-08-26)."""
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from routes import translate_routes as tr
from middleware.cleanup import FileCleanupManager
class TestSanitizeUrlFilename:
"""AC1 — path traversal in URL-downloaded filenames is neutralized."""
@pytest.mark.parametrize(
"raw",
[
"../../evil.xlsx",
"..\\..\\evil.docx",
"normal_file.pptx",
"..\t..evil.pdf",
"...",
"",
],
)
def test_traversal_stripped(self, raw):
result = tr._sanitize_url_filename(raw)
# Core guarantee: no traversal or path separators survive
assert ".." not in result
assert "/" not in result
assert "\\" not in result
assert result != ""
def test_long_filename_truncated(self):
raw = "a" * 300 + ".xlsx"
result = tr._sanitize_url_filename(raw)
assert len(result) <= 255
assert result.endswith(".xlsx")
class TestRedirectSsrf:
"""AC2 — redirect to internal address is blocked."""
@pytest.mark.asyncio
async def test_redirect_to_metadata_blocked(self):
def fake_response(status, location=None):
resp = MagicMock()
resp.status_code = status
resp.headers = {"location": location} if location else {}
return resp
class FakeClient:
def build_request(self, method, url):
return (method, url)
async def send(self, req, stream=False):
url = req[1]
if "public.example" in url:
return fake_response(302, "http://169.254.169.254/latest/meta-data")
return fake_response(200)
async def __aenter__(self):
return self
async def __aexit__(self, *a):
return False
with patch.object(tr.httpx, "AsyncClient", return_value=FakeClient()):
with pytest.raises(tr.TranslateEndpointError) as exc:
await tr.download_from_url("http://public.example/file.xlsx")
assert exc.value.details.get("reason") == "ssrf_blocked"
class TestCleanupOrphanGrace:
"""AC3 — young orphans are not deleted."""
def _manager(self, tmp_path):
m = FileCleanupManager(
upload_dir=tmp_path / "uploads",
output_dir=tmp_path / "outputs",
temp_dir=tmp_path / "temp",
)
for d in (m.upload_dir, m.output_dir, m.temp_dir):
d.mkdir(parents=True, exist_ok=True)
return m
@pytest.mark.asyncio
async def test_young_orphan_kept(self, tmp_path):
"""Orphan younger than the grace period is NOT deleted."""
m = self._manager(tmp_path)
f = m.upload_dir / "recent_orphan.xlsx"
f.write_bytes(b"x")
fake_redis = MagicMock()
fake_redis.keys = AsyncMock(return_value=[])
fake_redis.get = AsyncMock(return_value=None)
with patch(
"middleware.cleanup._get_async_redis", return_value=fake_redis
):
stats = await m.cleanup()
assert f.exists()
@pytest.mark.asyncio
async def test_input_path_key_recognized(self, tmp_path):
"""Files tracked under 'input_path' are not orphans (key mismatch fix)."""
import json as _json
m = self._manager(tmp_path)
f = m.upload_dir / "tracked.xlsx"
f.write_bytes(b"x")
fake_redis = MagicMock()
fake_redis.keys = AsyncMock(return_value=["translation:file:tr_1"])
fake_redis.get = AsyncMock(
return_value=_json.dumps({"input_path": str(f), "user_id": "u1"})
)
with patch(
"middleware.cleanup._get_async_redis", return_value=fake_redis
):
stats = await m.cleanup()
assert f.exists()
assert stats["orphaned_deleted"] == 0
@pytest.mark.asyncio
async def test_old_orphan_deleted(self, tmp_path):
import json as _json
import os
import time
m = self._manager(tmp_path)
f = m.upload_dir / "old_orphan.xlsx"
f.write_bytes(b"x")
old = time.time() - (m.orphan_grace_seconds + 600)
os.utime(f, (old, old))
fake_redis = MagicMock()
fake_redis.keys = AsyncMock(return_value=[])
fake_redis.get = AsyncMock(return_value=None)
with patch(
"middleware.cleanup._get_async_redis", return_value=fake_redis
):
stats = await m.cleanup()
assert not f.exists()
assert stats["orphaned_deleted"] == 1
class TestJobAccessControl:
"""H2 — ownership / token checks on status and download."""
def _job(self, user_id=None, token="tok123"):
return {"id": "tr_abc", "user_id": user_id, "access_token": token}
def _user(self, uid):
u = MagicMock()
u.id = uid
return u
def test_owner_allowed(self):
job = self._job(user_id=7)
assert tr._check_job_access(job, self._user(7), None) is None
def test_other_user_denied(self):
job = self._job(user_id=7)
resp = tr._check_job_access(job, self._user(8), None)
assert resp is not None and resp.status_code == 403
def test_anonymous_caller_on_owned_job_denied(self):
job = self._job(user_id=7)
resp = tr._check_job_access(job, None, "tok123")
assert resp is not None and resp.status_code == 401
def test_anonymous_job_requires_token(self):
job = self._job(user_id=None)
assert tr._check_job_access(job, None, "wrong") is not None
assert tr._check_job_access(job, None, None) is not None
assert tr._check_job_access(job, None, "tok123") is None
def test_old_anonymous_job_without_token_denied(self):
job = {"id": "tr_old", "user_id": None} # job created before the fix
assert tr._check_job_access(job, None, "anything") is not None
class TestZipBomb:
"""H1 — dangerous archives are rejected."""
def _make_zip(self, tmp_path, entries):
import zipfile
p = tmp_path / "bomb.xlsx"
with zipfile.ZipFile(p, "w", zipfile.ZIP_DEFLATED) as zf:
for name, data in entries:
zf.writestr(name, data)
return p
def test_normal_file_accepted(self, tmp_path):
from utils.file_handler import validate_zip_safety
p = self._make_zip(tmp_path, [("sheet1.xml", b"<xml/>ok" * 100)])
validate_zip_safety(p) # no exception
def test_not_a_zip_rejected(self, tmp_path):
from utils.file_handler import validate_zip_safety
p = tmp_path / "fake.xlsx"
p.write_bytes(b"this is not a zip file")
with pytest.raises(ValueError):
validate_zip_safety(p)
def test_high_ratio_rejected(self, tmp_path):
from utils.file_handler import validate_zip_safety
# 50 MB of zeros compresses far beyond the 100:1 ratio cap
p = self._make_zip(tmp_path, [("huge.xml", b"\0" * (50 * 1024 * 1024))])
with pytest.raises(ValueError):
validate_zip_safety(p)
def test_declared_total_too_big_rejected(self, tmp_path):
import zipfile
from unittest.mock import patch as _patch
from utils.file_handler import validate_zip_safety
p = self._make_zip(tmp_path, [("a.xml", b"<xml/>")])
fake_info = MagicMock()
fake_info.is_dir = lambda: False
fake_info.file_size = 5 * 1024 * 1024 * 1024 # 5 GB declared
fake_info.compress_size = 50 * 1024 * 1024
with _patch.object(zipfile.ZipFile, "infolist", return_value=[fake_info]):
with pytest.raises(ValueError):
validate_zip_safety(p)