462 lines
18 KiB
Python
462 lines
18 KiB
Python
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import AsyncMock, patch
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from backend.auth import create_access_token, get_current_user, require_admin
|
|
from backend.database import Database, LastAdminError
|
|
from backend.llm_client import get_llm_configs
|
|
from backend.main import _split_text, app
|
|
from backend.models import TranslationSession
|
|
from backend.sessions import (
|
|
SessionArchivedError,
|
|
SessionConflictError,
|
|
SessionStore,
|
|
session_store,
|
|
)
|
|
|
|
|
|
class ApiTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.temp_dir = tempfile.TemporaryDirectory()
|
|
cls.db = Database(Path(cls.temp_dir.name) / "api.db")
|
|
cls.db.create_user("admin", "unused", is_admin=True)
|
|
session_store.db = cls.db
|
|
app.dependency_overrides[get_current_user] = lambda: "admin"
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
app.dependency_overrides.clear()
|
|
cls.temp_dir.cleanup()
|
|
|
|
def setUp(self):
|
|
with self.db.connect() as connection:
|
|
connection.execute("DELETE FROM session_revisions")
|
|
connection.execute("DELETE FROM translation_sessions")
|
|
session_store._progress.clear()
|
|
token = create_access_token({"user_id": "admin"})
|
|
self.client = TestClient(app, headers={"Authorization": f"Bearer {token}"})
|
|
self.model = get_llm_configs()[0].alias
|
|
|
|
def create_session(self, source_text: str = "Hello") -> str:
|
|
response = self.client.post(
|
|
"/api/translate/create",
|
|
json={
|
|
"source_text": source_text,
|
|
"source_language": "영어",
|
|
"target_language": "한국어",
|
|
"model_phase1": self.model,
|
|
"model_phase2": "",
|
|
"model_phase3": self.model,
|
|
"model_phase4": self.model,
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
return response.json()["session_id"]
|
|
|
|
def test_translation_routes_require_authentication(self):
|
|
override = app.dependency_overrides.pop(get_current_user)
|
|
try:
|
|
response = TestClient(app).post(
|
|
"/api/translate/create", json={"source_text": "Hello"}
|
|
)
|
|
finally:
|
|
app.dependency_overrides[get_current_user] = override
|
|
self.assertEqual(response.status_code, 401)
|
|
|
|
def test_progress_endpoint_returns_only_transient_progress(self):
|
|
session_id = self.create_session()
|
|
session_store.set_progress(
|
|
session_id,
|
|
{
|
|
"phase": 1,
|
|
"chunk": 2,
|
|
"total_chunks": 3,
|
|
"status": "수신 중",
|
|
"preview": "부분 결과",
|
|
},
|
|
"admin",
|
|
)
|
|
|
|
response = self.client.get(f"/api/sessions/{session_id}/progress")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(
|
|
response.json(),
|
|
{
|
|
"progress": {
|
|
"phase": 1,
|
|
"chunk": 2,
|
|
"total_chunks": 3,
|
|
"status": "수신 중",
|
|
"preview": "부분 결과",
|
|
}
|
|
},
|
|
)
|
|
|
|
def test_unknown_model_is_rejected(self):
|
|
response = self.client.post(
|
|
"/api/translate/create",
|
|
json={"source_text": "Hello", "model_phase1": "missing-model"},
|
|
)
|
|
self.assertEqual(response.status_code, 400)
|
|
|
|
def test_source_update_invalidates_all_results(self):
|
|
session_id = self.create_session()
|
|
phase1 = json.dumps(
|
|
{
|
|
"translated": "안녕하세요",
|
|
"proper_nouns": [],
|
|
"summary": "인사",
|
|
"style": "중립적",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
responses = [phase1, "다시 번역", "최종 번역"]
|
|
with patch("backend.main.chat_complete", new=AsyncMock(side_effect=responses)):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase1")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.post(
|
|
f"/api/translate/{session_id}/phase2",
|
|
json={
|
|
"proper_nouns": [],
|
|
"style": "중립적",
|
|
"expected_version": response.json()["version"],
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.post(f"/api/translate/{session_id}/phase3")
|
|
self.assertEqual(response.status_code, 200)
|
|
response = self.client.post(f"/api/translate/{session_id}/phase4")
|
|
self.assertEqual(response.status_code, 200)
|
|
|
|
response = self.client.patch(
|
|
f"/api/sessions/{session_id}",
|
|
json={"source_text": "Changed", "expected_version": response.json()["version"]},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
data = response.json()["data"]
|
|
self.assertIsNone(data["phase1_result"])
|
|
self.assertEqual(data["phase1_chunks"], [])
|
|
self.assertEqual(data["phase2_proper_nouns"], [])
|
|
self.assertEqual(data["phase2_style"], "")
|
|
self.assertEqual(data["phase3_result"], "")
|
|
self.assertEqual(data["phase3_chunks"], [])
|
|
self.assertEqual(data["phase4_result"], "")
|
|
|
|
def test_invalid_phase1_json_falls_back_to_plain_translation(self):
|
|
session_id = self.create_session()
|
|
mocked = AsyncMock(side_effect=["not json", "still not json", "일반 번역"])
|
|
with patch("backend.main.chat_complete", new=mocked):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase1")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"]["phase1_result"]["translated"], "일반 번역")
|
|
self.assertTrue(response.json()["data"]["phase1_result"]["warnings"])
|
|
self.assertEqual(mocked.await_count, 3)
|
|
|
|
def test_phase1_accepts_fenced_json_with_trailing_comma(self):
|
|
session_id = self.create_session()
|
|
malformed = """응답입니다.
|
|
```json
|
|
{"translated":"안녕하세요","proper_nouns":[],"summary":"인사","style":"중립적",}
|
|
```
|
|
"""
|
|
mocked = AsyncMock(return_value=malformed)
|
|
with patch("backend.main.chat_complete", new=mocked):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase1")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(response.json()["data"]["phase1_result"]["translated"], "안녕하세요")
|
|
self.assertEqual(mocked.await_count, 1)
|
|
|
|
def test_phase2_rejects_stale_version(self):
|
|
session_id = self.create_session()
|
|
phase1 = json.dumps(
|
|
{
|
|
"translated": "안녕하세요",
|
|
"proper_nouns": [],
|
|
"summary": "",
|
|
"style": "",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
with patch("backend.main.chat_complete", new=AsyncMock(return_value=phase1)):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase1")
|
|
stale_version = response.json()["version"]
|
|
session_store.update(session_id, {"phase2_style": "newer"}, "admin")
|
|
|
|
response = self.client.post(
|
|
f"/api/translate/{session_id}/phase2",
|
|
json={
|
|
"proper_nouns": [],
|
|
"style": "stale",
|
|
"expected_version": stale_version,
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 409)
|
|
|
|
def test_archived_session_rejects_api_patch(self):
|
|
session_id = self.create_session()
|
|
response = self.client.get(f"/api/sessions/{session_id}")
|
|
version = response.json()["version"]
|
|
self.assertEqual(
|
|
self.client.post(f"/api/sessions/{session_id}/archive").status_code,
|
|
200,
|
|
)
|
|
|
|
response = self.client.patch(
|
|
f"/api/sessions/{session_id}",
|
|
json={"source_text": "changed", "expected_version": version + 1},
|
|
)
|
|
self.assertEqual(response.status_code, 409)
|
|
|
|
def test_long_phase1_is_split_into_bounded_chunks(self):
|
|
source_text = "这是用于测试长文本分割的句子。" * 120
|
|
expected_chunks = _split_text(source_text)
|
|
self.assertGreater(len(expected_chunks), 1)
|
|
self.assertTrue(all(len(chunk) <= 1500 for chunk in expected_chunks))
|
|
session_id = self.create_session(source_text)
|
|
phase1 = json.dumps(
|
|
{
|
|
"translated": "긴 문장 번역",
|
|
"proper_nouns": [],
|
|
"summary": "요약",
|
|
"style": "중립적",
|
|
},
|
|
ensure_ascii=False,
|
|
)
|
|
mocked = AsyncMock(side_effect=[phase1] * len(expected_chunks))
|
|
with patch("backend.main.chat_complete", new=mocked):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase1")
|
|
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(mocked.await_count, len(expected_chunks))
|
|
self.assertEqual(len(response.json()["data"]["phase1_chunks"]), len(expected_chunks))
|
|
|
|
response = self.client.post(
|
|
f"/api/translate/{session_id}/phase2",
|
|
json={
|
|
"proper_nouns": [],
|
|
"style": "중립적",
|
|
"expected_version": response.json()["version"],
|
|
},
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
phase3_mock = AsyncMock(side_effect=["재번역 조각"] * len(expected_chunks))
|
|
with patch("backend.main.chat_complete", new=phase3_mock):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase3")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(phase3_mock.await_count, len(expected_chunks))
|
|
self.assertEqual(len(response.json()["data"]["phase3_chunks"]), len(expected_chunks))
|
|
|
|
phase4_mock = AsyncMock(side_effect=["완성 조각"] * len(expected_chunks))
|
|
with patch("backend.main.chat_complete", new=phase4_mock):
|
|
response = self.client.post(f"/api/translate/{session_id}/phase4")
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertEqual(phase4_mock.await_count, len(expected_chunks))
|
|
self.assertEqual(
|
|
response.json()["data"]["phase4_result"].count("완성 조각"),
|
|
len(expected_chunks),
|
|
)
|
|
|
|
def test_split_text_prefers_english_sentence_boundary(self):
|
|
sentence = "This is an English sentence with several words. "
|
|
chunks = _split_text(sentence * 50)
|
|
|
|
self.assertGreater(len(chunks), 1)
|
|
self.assertTrue(all(chunk.endswith(".") for chunk in chunks[:-1]))
|
|
self.assertTrue(all(len(chunk) <= 1500 for chunk in chunks))
|
|
|
|
|
|
class SessionStoreTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.db = Database(Path(self.temp_dir.name) / "store.db")
|
|
self.db.create_user("alice", "unused")
|
|
self.db.create_user("bob", "unused")
|
|
self.store = SessionStore(self.db)
|
|
|
|
def tearDown(self):
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_sessions_are_isolated_by_owner(self):
|
|
store = self.store
|
|
session_id = store.create(TranslationSession(source_text="secret"), "alice")
|
|
|
|
self.assertIsNotNone(store.get(session_id, "alice"))
|
|
self.assertIsNone(store.get(session_id, "bob"))
|
|
self.assertIsNone(store.update(session_id, {"source_text": "stolen"}, "bob"))
|
|
self.assertFalse(store.delete(session_id, "bob"))
|
|
self.assertEqual(store.get(session_id, "alice").source_text, "secret")
|
|
|
|
def test_stale_phase_update_is_rejected(self):
|
|
store = self.store
|
|
session_id = store.create(TranslationSession(source_text="first"), "alice")
|
|
_, version = store.get_with_version(session_id, "alice")
|
|
store.update(session_id, {"source_text": "second"}, "alice")
|
|
|
|
with self.assertRaises(SessionConflictError):
|
|
store.update(
|
|
session_id,
|
|
{"phase3_result": "stale"},
|
|
"alice",
|
|
expected_version=version,
|
|
)
|
|
|
|
def test_progress_does_not_change_session_version(self):
|
|
store = self.store
|
|
session_id = store.create(TranslationSession(source_text="first"), "alice")
|
|
_, version = store.get_with_version(session_id, "alice")
|
|
|
|
self.assertTrue(
|
|
store.set_progress(
|
|
session_id,
|
|
{
|
|
"phase": 1,
|
|
"chunk": 1,
|
|
"total_chunks": 2,
|
|
"status": "수신 중",
|
|
"preview": "부분 결과",
|
|
},
|
|
"alice",
|
|
)
|
|
)
|
|
session, current_version = store.get_with_version(session_id, "alice")
|
|
self.assertEqual(current_version, version)
|
|
self.assertEqual(session.progress.preview, "부분 결과")
|
|
|
|
def test_session_persists_across_store_instances(self):
|
|
session_id = self.store.create(
|
|
TranslationSession(source_text="persistent text"), "alice"
|
|
)
|
|
|
|
reopened_store = SessionStore(Database(self.db.path))
|
|
reopened = reopened_store.get(session_id, "alice")
|
|
self.assertIsNotNone(reopened)
|
|
self.assertEqual(reopened.source_text, "persistent text")
|
|
|
|
def test_previous_revision_can_be_restored(self):
|
|
session_id = self.store.create(TranslationSession(source_text="first"), "alice")
|
|
self.store.update(
|
|
session_id,
|
|
{
|
|
"phase1_result": {
|
|
"translated": "첫 번역",
|
|
"proper_nouns": [],
|
|
"summary": "",
|
|
"style": "",
|
|
}
|
|
},
|
|
"alice",
|
|
current_phase=1,
|
|
)
|
|
self.store.update(
|
|
session_id,
|
|
{"source_text": "second", "phase1_result": None},
|
|
"alice",
|
|
snapshot_reason="session_edit",
|
|
current_phase=0,
|
|
)
|
|
|
|
revision = self.store.get_revision(session_id, "alice")
|
|
self.assertEqual(revision["data"]["source_text"], "first")
|
|
self.assertTrue(self.store.restore_revision(session_id, "alice"))
|
|
restored = self.store.get(session_id, "alice")
|
|
self.assertEqual(restored.source_text, "first")
|
|
self.assertEqual(restored.phase1_result.translated, "첫 번역")
|
|
|
|
def test_archive_clone_and_list_are_owner_scoped(self):
|
|
session_id = self.store.create(TranslationSession(source_text="Alice text"), "alice")
|
|
self.store.create(TranslationSession(source_text="Bob text"), "bob")
|
|
|
|
self.assertEqual(len(self.store.list("alice")), 1)
|
|
self.assertTrue(self.store.archive(session_id, "alice"))
|
|
self.assertEqual(self.store.list("alice"), [])
|
|
self.assertEqual(len(self.store.list("alice", include_archived=True)), 1)
|
|
clone_id = self.store.clone(session_id, "alice")
|
|
self.assertIsNotNone(clone_id)
|
|
self.assertIsNone(self.store.get(clone_id, "bob"))
|
|
|
|
def test_archived_session_rejects_content_updates(self):
|
|
session_id = self.store.create(TranslationSession(source_text="locked"), "alice")
|
|
_, version = self.store.get_with_version(session_id, "alice")
|
|
self.assertTrue(self.store.archive(session_id, "alice"))
|
|
|
|
with self.assertRaises(SessionArchivedError):
|
|
self.store.update(
|
|
session_id,
|
|
{"source_text": "changed"},
|
|
"alice",
|
|
expected_version=version + 1,
|
|
)
|
|
|
|
def test_clone_preserves_completed_phase_metadata(self):
|
|
session_id = self.store.create(TranslationSession(source_text="complete"), "alice")
|
|
self.store.update(
|
|
session_id,
|
|
{"phase4_result": "final"},
|
|
"alice",
|
|
current_phase=4,
|
|
)
|
|
|
|
clone_id = self.store.clone(session_id, "alice")
|
|
summary = next(item for item in self.store.list("alice") if item["session_id"] == clone_id)
|
|
self.assertEqual(summary["current_phase"], 4)
|
|
self.assertEqual(summary["status"], "completed")
|
|
|
|
|
|
class UserDatabaseTests(unittest.TestCase):
|
|
def test_password_change_increments_token_version(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
db = Database(Path(directory) / "users.db")
|
|
db.create_user("admin", "first", is_admin=True)
|
|
before = db.get_user("admin")
|
|
self.assertTrue(db.update_user("admin", password_hash="second"))
|
|
after = db.get_user("admin")
|
|
self.assertEqual(after["token_version"], before["token_version"] + 1)
|
|
|
|
def test_last_active_admin_cannot_be_removed(self):
|
|
with tempfile.TemporaryDirectory() as directory:
|
|
db = Database(Path(directory) / "users.db")
|
|
db.create_user("admin", "first", is_admin=True)
|
|
|
|
with self.assertRaises(LastAdminError):
|
|
db.update_user("admin", is_active=False)
|
|
|
|
|
|
class AdminApiTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp_dir = tempfile.TemporaryDirectory()
|
|
self.db = Database(Path(self.temp_dir.name) / "admin.db")
|
|
self.db.create_user("admin", "unused", is_admin=True)
|
|
self.database_patch = patch("backend.main.database", self.db)
|
|
self.database_patch.start()
|
|
app.dependency_overrides[require_admin] = lambda: "admin"
|
|
self.client = TestClient(app)
|
|
|
|
def tearDown(self):
|
|
app.dependency_overrides.pop(require_admin, None)
|
|
self.database_patch.stop()
|
|
self.temp_dir.cleanup()
|
|
|
|
def test_admin_can_create_and_deactivate_user(self):
|
|
response = self.client.post(
|
|
"/api/admin/users",
|
|
json={"id": "translator", "password": "secure-pass", "is_admin": False},
|
|
)
|
|
self.assertEqual(response.status_code, 201)
|
|
users = self.client.get("/api/admin/users").json()
|
|
self.assertEqual({user["id"] for user in users}, {"admin", "translator"})
|
|
|
|
response = self.client.patch(
|
|
"/api/admin/users/translator", json={"is_active": False}
|
|
)
|
|
self.assertEqual(response.status_code, 200)
|
|
self.assertFalse(self.db.get_user("translator")["is_active"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|