sepehr
b4e873ad2c
Some checks failed
Deploy to Production / Build and Deploy (push) Failing after 2m14s
Foundations:
- TranslationSegment model + migration f7e8d9c0b1a2 (segments, workspaces,
workspace_members, glossaries.workspace_id)
- SegmentRecorder injected into all 4 translators: unique (source,
translation) pairs captured per job and persisted (best-effort)
- set_segment_overrides: human-reviewed translations applied verbatim on
rebuild — top priority over TM and provider, zero API calls
Review API (routes/review_routes.py):
- GET /translations/{id}/segments (owner or job token)
- PATCH /segments/{id} edit/approve — feeds the per-user TM so approved
translations are reused in later jobs
- POST /translations/{id}/rebuild — rebuild document with reviewed text
- GET/POST /translations/{id}/xliff — XLIFF 1.2 export/import (edited
segments export their reviewed text)
Review editor (frontend /dashboard/reviews/[jobId]):
- side-by-side source/translation table, inline edit, approve (single or
all), rebuild & download (auth blob), XLIFF export/import, 13 locales
- 'Relire et corriger' link on the translation-complete screen
Team workspaces (routes/workspace_routes.py + /dashboard/teams):
- Workspace/WorkspaceMember models, roles owner/admin/member
- create (Business plan), list with seat usage, invite by email with
seat-limit enforcement (Business=5, Enterprise unlimited), removal
- shared glossaries: workspace members can use a glossary shared to their
workspace (access check extended)
Tests: 1184 passed / 0 failed (11 new: recorder, overrides, docx
capture->rebuild e2e, XLIFF structure/escaping, seats, workspace CRUD,
shared glossary access)
289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""Review foundation: segment recording → persistence → overrides → rebuild,
|
|
plus the XLIFF and workspace APIs."""
|
|
|
|
import pytest
|
|
from docx import Document
|
|
|
|
from database.models import TranslationSegment, Workspace, WorkspaceMember
|
|
from translators.segments import SegmentRecorder, apply_overrides
|
|
from translators.word_translator import WordTranslator
|
|
|
|
|
|
class _FakeProvider:
|
|
"""New-style provider; translates only on explicit call."""
|
|
|
|
def get_name(self):
|
|
return "fake"
|
|
|
|
def is_available(self):
|
|
return True
|
|
|
|
def translate_batch(self, texts, target_language, source_language="auto"):
|
|
return [f"FR:{t}" for t in texts]
|
|
|
|
|
|
# ===========================================================================
|
|
# Recorder + overrides (unit)
|
|
# ===========================================================================
|
|
class TestRecorder:
|
|
def test_records_unique_in_order(self):
|
|
rec = SegmentRecorder()
|
|
rec.record_pair("Hello", "Bonjour")
|
|
rec.record_pair("World", "Monde")
|
|
rec.record_pair("Hello", "Bonjour") # duplicate ignored
|
|
assert rec.get_pairs() == [("Hello", "Bonjour"), ("World", "Monde")]
|
|
|
|
def test_skips_empty_and_identity(self):
|
|
rec = SegmentRecorder()
|
|
rec.record_pair("", "x")
|
|
rec.record_pair(" ", "x")
|
|
rec.record_pair("same", "same")
|
|
assert rec.get_pairs() == []
|
|
|
|
|
|
class TestApplyOverrides:
|
|
def test_hits_and_misses(self):
|
|
hits, misses = apply_overrides(
|
|
["a", "b", "c"], {"a": "A!", "c": "C!"}
|
|
)
|
|
assert hits == {0: "A!", 2: "C!"}
|
|
assert misses == [1]
|
|
|
|
def test_identity_override_is_a_miss(self):
|
|
hits, misses = apply_overrides(["same"], {"same": "same"})
|
|
assert hits == {}
|
|
assert misses == [0]
|
|
|
|
def test_none_overrides_all_miss(self):
|
|
hits, misses = apply_overrides(["a", "b"], None)
|
|
assert hits == {} and misses == [0, 1]
|
|
|
|
|
|
# ===========================================================================
|
|
# End-to-end: translate → recorded segments → rebuild with overrides
|
|
# ===========================================================================
|
|
class TestSegmentCaptureAndRebuild:
|
|
def _make_doc(self, path):
|
|
doc = Document()
|
|
doc.add_paragraph("Hello world")
|
|
doc.add_paragraph("Second paragraph")
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
doc.save(str(path))
|
|
|
|
def test_capture_and_rebuild(self, tmp_path):
|
|
from database.models import Base
|
|
from database.connection import sync_engine
|
|
|
|
Base.metadata.create_all(bind=sync_engine)
|
|
|
|
self._make_doc(tmp_path / "in.docx")
|
|
|
|
# 1) Translate with recorder
|
|
t = WordTranslator(provider=_FakeProvider())
|
|
rec = SegmentRecorder()
|
|
t.set_segment_recorder(rec)
|
|
t.translate_file(tmp_path / "in.docx", tmp_path / "out.docx", "fr", "en")
|
|
|
|
pairs = t.get_recorded_segments()
|
|
assert ("Hello world", "FR:Hello world") in pairs
|
|
assert ("Second paragraph", "FR:Second paragraph") in pairs
|
|
|
|
# 2) Rebuild with a human-reviewed override
|
|
t2 = WordTranslator(provider=_FakeProvider())
|
|
t2.set_segment_overrides({"Hello world": "Bonjour le monde (relu)"})
|
|
t2.translate_file(tmp_path / "in.docx", tmp_path / "rebuilt.docx", "fr", "en")
|
|
|
|
result = Document(str(tmp_path / "rebuilt.docx"))
|
|
texts = [p.text for p in result.paragraphs]
|
|
assert "Bonjour le monde (relu)" in texts
|
|
# Non-overridden text still machine-translated
|
|
assert "FR:Second paragraph" in texts
|
|
|
|
|
|
# ===========================================================================
|
|
# XLIFF export format
|
|
# ===========================================================================
|
|
class TestXliff:
|
|
def test_export_structure(self):
|
|
from routes.review_routes import _xliff_export
|
|
|
|
job = {
|
|
"source_lang": "en",
|
|
"target_lang": "fr",
|
|
"file_name": "doc.docx",
|
|
}
|
|
segments = [
|
|
{
|
|
"id": "seg-1",
|
|
"segment_index": 0,
|
|
"source_text": "Hello",
|
|
"translated_text": "Bonjour",
|
|
"status": "pending",
|
|
"reviewed_text": None,
|
|
},
|
|
{
|
|
"id": "seg-2",
|
|
"segment_index": 1,
|
|
"source_text": "World",
|
|
"translated_text": "Monde",
|
|
"status": "edited",
|
|
"reviewed_text": "Le monde",
|
|
},
|
|
]
|
|
xml = _xliff_export(job, segments)
|
|
assert '<xliff xmlns="urn:oasis:names:tc:xliff:document:1.2"' in xml
|
|
assert 'source-language="en"' in xml
|
|
assert "<source>Hello</source>" in xml
|
|
# edited segments export their REVIEWED text
|
|
assert "<target>Le monde</target>" in xml
|
|
assert "status: edited" in xml
|
|
|
|
def test_export_escapes_xml(self):
|
|
from routes.review_routes import _xliff_export
|
|
|
|
xml = _xliff_export(
|
|
{"source_lang": "en", "target_lang": "fr", "file_name": "x"},
|
|
[
|
|
{
|
|
"id": "s1",
|
|
"segment_index": 0,
|
|
"source_text": "a < b & c",
|
|
"translated_text": "x < y",
|
|
"status": "pending",
|
|
"reviewed_text": None,
|
|
}
|
|
],
|
|
)
|
|
assert "a < b & c" in xml
|
|
|
|
|
|
# ===========================================================================
|
|
# Workspace seats
|
|
# ===========================================================================
|
|
class TestWorkspaceSeats:
|
|
def test_seat_limit_helpers(self):
|
|
from routes.workspace_routes import _seat_limit_for_owner
|
|
from models.subscription import PlanType
|
|
|
|
assert _seat_limit_for_owner(PlanType.BUSINESS) == 5
|
|
assert _seat_limit_for_owner(PlanType.FREE) == 0
|
|
assert _seat_limit_for_owner(PlanType.PRO) == 0
|
|
|
|
def test_workspace_crud_flow(self):
|
|
from database.connection import get_sync_session
|
|
from database.models import Base, User
|
|
from database.connection import sync_engine
|
|
import uuid
|
|
|
|
Base.metadata.create_all(bind=sync_engine)
|
|
|
|
with get_sync_session() as session:
|
|
owner = User(
|
|
id=str(uuid.uuid4()),
|
|
email=f"owner-{uuid.uuid4().hex[:6]}@t.local",
|
|
name="Owner",
|
|
password_hash="x",
|
|
plan="business",
|
|
)
|
|
member = User(
|
|
id=str(uuid.uuid4()),
|
|
email=f"member-{uuid.uuid4().hex[:6]}@t.local",
|
|
name="Member",
|
|
password_hash="x",
|
|
)
|
|
session.add_all([owner, member])
|
|
session.flush()
|
|
|
|
ws = Workspace(name="Équipe", owner_id=owner.id)
|
|
session.add(ws)
|
|
session.flush()
|
|
session.add(
|
|
WorkspaceMember(
|
|
workspace_id=ws.id, user_id=owner.id, role="owner"
|
|
)
|
|
)
|
|
session.add(
|
|
WorkspaceMember(
|
|
workspace_id=ws.id, user_id=member.id, role="member"
|
|
)
|
|
)
|
|
session.commit()
|
|
|
|
count = (
|
|
session.query(WorkspaceMember)
|
|
.filter(WorkspaceMember.workspace_id == ws.id)
|
|
.count()
|
|
)
|
|
assert count == 2
|
|
from services.glossary_service import _user_workspace_ids
|
|
|
|
ids = _user_workspace_ids(session, member.id)
|
|
assert ws.id in ids
|
|
|
|
session.query(WorkspaceMember).filter(
|
|
WorkspaceMember.workspace_id == ws.id
|
|
).delete()
|
|
session.query(Workspace).filter(Workspace.id == ws.id).delete()
|
|
session.query(User).filter(User.id.in_([owner.id, member.id])).delete()
|
|
session.commit()
|
|
|
|
|
|
# ===========================================================================
|
|
# Shared glossary access
|
|
# ===========================================================================
|
|
class TestSharedGlossaryAccess:
|
|
def test_workspace_member_can_access_shared_glossary(self):
|
|
from database.connection import get_sync_session, sync_engine
|
|
from database.models import Base, User, Glossary
|
|
from services.glossary_service import (
|
|
get_glossary_terms,
|
|
validate_glossary_access,
|
|
)
|
|
from utils.exceptions import GlossaryNotFoundError
|
|
import uuid
|
|
|
|
Base.metadata.create_all(bind=sync_engine)
|
|
uid = uuid.uuid4().hex[:8]
|
|
|
|
with get_sync_session() as session:
|
|
owner = User(id=str(uuid.uuid4()), email=f"go-{uid}@t.local",
|
|
name="O", password_hash="x")
|
|
member = User(id=str(uuid.uuid4()), email=f"gm-{uid}@t.local",
|
|
name="M", password_hash="x")
|
|
outsider = User(id=str(uuid.uuid4()), email=f"gx-{uid}@t.local",
|
|
name="X", password_hash="x")
|
|
session.add_all([owner, member, outsider])
|
|
session.flush()
|
|
|
|
ws = Workspace(name="W", owner_id=owner.id)
|
|
session.add(ws)
|
|
session.flush()
|
|
session.add(WorkspaceMember(workspace_id=ws.id, user_id=member.id, role="member"))
|
|
|
|
glossary = Glossary(
|
|
id=str(uuid.uuid4()), user_id=owner.id,
|
|
workspace_id=ws.id, name="Shared",
|
|
)
|
|
session.add(glossary)
|
|
session.commit()
|
|
gid = glossary.id
|
|
o_id, m_id, x_id = owner.id, member.id, outsider.id
|
|
ws_id = ws.id
|
|
|
|
try:
|
|
# owner OK
|
|
assert validate_glossary_access(gid, o_id) is True
|
|
# workspace member OK (shared)
|
|
assert validate_glossary_access(gid, m_id) is True
|
|
assert get_glossary_terms(gid, m_id)["source_language"] == "fr"
|
|
# outsider refused
|
|
with pytest.raises(GlossaryNotFoundError):
|
|
validate_glossary_access(gid, x_id)
|
|
finally:
|
|
with get_sync_session() as session:
|
|
session.query(Glossary).filter(Glossary.id == gid).delete()
|
|
session.query(WorkspaceMember).filter(WorkspaceMember.workspace_id == ws_id).delete()
|
|
session.query(Workspace).filter(Workspace.id == ws_id).delete()
|
|
session.query(User).filter(User.id.in_([o_id, m_id, x_id])).delete()
|
|
session.commit()
|