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()