import json import unittest from unittest.mock import AsyncMock, patch from fastapi.testclient import TestClient from backend.auth import create_access_token from backend.llm_client import get_llm_configs from backend.main import _split_text, app from backend.models import TranslationSession from backend.sessions import SessionConflictError, SessionStore, session_store class ApiTests(unittest.TestCase): def setUp(self): session_store._store.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): response = TestClient(app).post( "/api/translate/create", json={"source_text": "Hello"} ) 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)): self.assertEqual( self.client.post(f"/api/translate/{session_id}/phase1").status_code, 200, ) self.assertEqual( self.client.post( f"/api/translate/{session_id}/phase2", json={"proper_nouns": [], "style": "중립적"}, ).status_code, 200, ) self.assertEqual( self.client.post(f"/api/translate/{session_id}/phase3").status_code, 200, ) self.assertEqual( self.client.post(f"/api/translate/{session_id}/phase4").status_code, 200, ) response = self.client.patch( f"/api/sessions/{session_id}", json={"source_text": "Changed"} ) 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_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)) self.client.post( f"/api/translate/{session_id}/phase2", json={"proper_nouns": [], "style": "중립적"}, ) 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 test_sessions_are_isolated_by_owner(self): store = SessionStore() 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 = SessionStore() 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 = SessionStore() 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, "부분 결과") if __name__ == "__main__": unittest.main()