"""Tests for Translation Memory models and manager.""" import json import pytest from datetime import datetime from pathlib import Path from src.context.translation_memory import TranslationMemoryManager from src.models.translation_memory import ( MatchType, SegmentLevel, TMMatch, TMSegment, TMStatistics, calculate_similarity, normalize_text, _levenshtein_distance, ) from src.models.verse import VerseReference class TestNormalizeText: """Tests for text normalization.""" def test_lowercase(self) -> None: """Test lowercasing.""" assert normalize_text("ABC") == "abc" def test_remove_punctuation(self) -> None: """Test punctuation removal.""" result = normalize_text("Hello, World!") assert "," not in result assert "!" not in result def test_collapse_whitespace(self) -> None: """Test whitespace collapsing.""" result = normalize_text("hello world") assert result == "hello world" def test_unicode_normalization(self) -> None: """Test Unicode normalization.""" # Greek text should be normalized text = "λόγος" result = normalize_text(text) assert result == "λόγος" def test_preserve_greek_characters(self) -> None: """Test Greek characters are preserved.""" text = "Ἐν ἀρχῇ ἦν ὁ λόγος" result = normalize_text(text) assert "λόγος" in result class TestCalculateSimilarity: """Tests for similarity calculation.""" def test_exact_match(self) -> None: """Test exact match returns 1.0.""" assert calculate_similarity("hello", "hello") == 1.0 def test_completely_different(self) -> None: """Test completely different returns low score.""" score = calculate_similarity("abc", "xyz") assert score < 0.5 def test_similar_strings(self) -> None: """Test similar strings return high score.""" score = calculate_similarity("hello", "hallo") assert score > 0.7 def test_empty_strings(self) -> None: """Test empty string handling.""" assert calculate_similarity("", "") == 1.0 assert calculate_similarity("hello", "") == 0.0 assert calculate_similarity("", "hello") == 0.0 def test_symmetric(self) -> None: """Test similarity is symmetric.""" s1 = calculate_similarity("hello", "hallo") s2 = calculate_similarity("hallo", "hello") assert s1 == s2 class TestLevenshteinDistance: """Tests for Levenshtein distance.""" def test_identical(self) -> None: """Test identical strings have distance 0.""" assert _levenshtein_distance("hello", "hello") == 0 def test_single_insertion(self) -> None: """Test single insertion.""" assert _levenshtein_distance("hello", "hellos") == 1 def test_single_deletion(self) -> None: """Test single deletion.""" assert _levenshtein_distance("hello", "hell") == 1 def test_single_substitution(self) -> None: """Test single substitution.""" assert _levenshtein_distance("hello", "hallo") == 1 def test_empty_to_string(self) -> None: """Test empty to string.""" assert _levenshtein_distance("", "hello") == 5 def test_string_to_empty(self) -> None: """Test string to empty.""" assert _levenshtein_distance("hello", "") == 5 class TestSegmentLevel: """Tests for SegmentLevel enum.""" def test_values(self) -> None: """Test enum values.""" assert SegmentLevel.VERSE.value == "verse" assert SegmentLevel.PHRASE.value == "phrase" class TestMatchType: """Tests for MatchType enum.""" def test_values(self) -> None: """Test enum values.""" assert MatchType.EXACT.value == "exact" assert MatchType.FUZZY.value == "fuzzy" assert MatchType.PHRASE.value == "phrase" class TestTMSegment: """Tests for TMSegment model.""" def test_create_basic(self) -> None: """Test basic segment creation.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment( source_text="Ἐν ἀρχῇ ἦν ὁ λόγος", source_normalized="ἐν ἀρχῇ ἦν ὁ λόγος", target_text="初めに言葉があった", reference=ref, ) assert segment.source_text == "Ἐν ἀρχῇ ἦν ὁ λόγος" assert segment.level == SegmentLevel.VERSE assert segment.reuse_count == 0 def test_create_factory(self) -> None: """Test factory method with auto-normalization.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="Ἐν ἀρχῇ ἦν ὁ λόγος", target_text="初めに言葉があった", reference=ref, ) # Normalized should be lowercase without extra punctuation assert segment.source_normalized == normalize_text("Ἐν ἀρχῇ ἦν ὁ λόγος") def test_create_with_metadata(self) -> None: """Test creation with metadata.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="Test", target_text="テスト", reference=ref, metadata={"model": "claude-sonnet", "version": "1.0"}, ) assert segment.metadata["model"] == "claude-sonnet" def test_auto_generated_id(self) -> None: """Test ID is auto-generated.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="Test", target_text="テスト", reference=ref, ) assert segment.id is not None assert len(segment.id) > 0 def test_increment_reuse(self) -> None: """Test reuse counter increment.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="Test", target_text="テスト", reference=ref, ) assert segment.reuse_count == 0 segment.increment_reuse() assert segment.reuse_count == 1 segment.increment_reuse() assert segment.reuse_count == 2 class TestTMMatch: """Tests for TMMatch model.""" @pytest.fixture def sample_segment(self) -> TMSegment: """Create a sample segment.""" ref = VerseReference(book="john", chapter=1, verse=1) return TMSegment.create( source_text="Test", target_text="テスト", reference=ref, ) def test_create_match(self, sample_segment: TMSegment) -> None: """Test creating a match.""" match = TMMatch( segment=sample_segment, similarity=0.95, match_type=MatchType.FUZZY, ) assert match.similarity == 0.95 assert match.match_type == MatchType.FUZZY def test_is_exact(self, sample_segment: TMSegment) -> None: """Test is_exact property.""" exact = TMMatch( segment=sample_segment, similarity=1.0, match_type=MatchType.EXACT, ) fuzzy = TMMatch( segment=sample_segment, similarity=0.9, match_type=MatchType.FUZZY, ) assert exact.is_exact is True assert fuzzy.is_exact is False def test_is_usable(self, sample_segment: TMSegment) -> None: """Test is_usable property.""" high = TMMatch( segment=sample_segment, similarity=0.8, match_type=MatchType.FUZZY, ) low = TMMatch( segment=sample_segment, similarity=0.5, match_type=MatchType.FUZZY, ) assert high.is_usable is True assert low.is_usable is False class TestTMStatistics: """Tests for TMStatistics model.""" def test_from_empty_segments(self) -> None: """Test statistics from empty list.""" stats = TMStatistics.from_segments([]) assert stats.total_segments == 0 assert stats.average_reuse == 0.0 def test_from_segments(self) -> None: """Test statistics calculation.""" segments = [ TMSegment.create( source_text=f"Test {i}", target_text=f"テスト {i}", reference=VerseReference(book="john", chapter=1, verse=i), level=SegmentLevel.VERSE if i % 2 == 0 else SegmentLevel.PHRASE, ) for i in range(1, 6) ] # Set some reuse counts segments[0].reuse_count = 5 segments[1].reuse_count = 3 stats = TMStatistics.from_segments(segments) assert stats.total_segments == 5 # range(1,6) gives [1,2,3,4,5]. i%2==0 for i=2,4. So verse_segments = 2 assert stats.verse_segments == 2 assert stats.phrase_segments == 3 assert stats.total_reuses == 8 assert "john" in stats.books_covered class TestTranslationMemoryManager: """Tests for TranslationMemoryManager.""" @pytest.fixture def manager(self, tmp_path: Path) -> TranslationMemoryManager: """Create a manager with temporary storage.""" return TranslationMemoryManager(tmp_path / "segments.json") @pytest.fixture def sample_segment(self) -> TMSegment: """Create a sample segment.""" ref = VerseReference(book="john", chapter=1, verse=1) return TMSegment.create( source_text="Ἐν ἀρχῇ ἦν ὁ λόγος", target_text="初めに言葉があった", reference=ref, ) def test_init_default_path(self) -> None: """The default path is taken from the module constant. The constant is redirected by an autouse fixture during tests, so this asserts the wiring rather than a literal production path — which is the point: no test may reach the real store. """ import src.context.translation_memory as tm_module manager = TranslationMemoryManager() assert manager.storage_path == tm_module.DEFAULT_STORAGE_PATH assert "data/memory/segments.json" not in str(manager.storage_path) def test_init_custom_path(self, tmp_path: Path) -> None: """Test custom storage path.""" path = tmp_path / "custom.json" manager = TranslationMemoryManager(path) assert manager.storage_path == path def test_add_segment( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test adding a segment.""" manager.add_segment(sample_segment) assert len(manager) == 1 def test_add_from_translation(self, manager: TranslationMemoryManager) -> None: """Test creating segment from translation.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = manager.add_from_translation( source_text="Test source", target_text="Test target", reference=ref, metadata={"model": "test"}, ) assert len(manager) == 1 assert segment.metadata["model"] == "test" def test_get_segment( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test getting a segment by ID.""" manager.add_segment(sample_segment) retrieved = manager.get_segment(sample_segment.id) assert retrieved is not None assert retrieved.id == sample_segment.id def test_get_segment_not_found(self, manager: TranslationMemoryManager) -> None: """Test getting nonexistent segment.""" result = manager.get_segment("nonexistent") assert result is None def test_get_by_reference( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test getting segments by reference.""" manager.add_segment(sample_segment) results = manager.get_by_reference(sample_segment.reference) assert len(results) == 1 assert results[0].id == sample_segment.id def test_get_by_reference_empty(self, manager: TranslationMemoryManager) -> None: """Test getting segments for unknown reference.""" ref = VerseReference(book="john", chapter=99, verse=99) results = manager.get_by_reference(ref) assert len(results) == 0 def test_find_matches_exact( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test finding exact match.""" manager.add_segment(sample_segment) matches = manager.find_matches(sample_segment.source_text) assert len(matches) == 1 assert matches[0].is_exact def test_find_matches_fuzzy(self, manager: TranslationMemoryManager) -> None: """Test finding fuzzy matches.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="hello world", target_text="こんにちは世界", reference=ref, ) manager.add_segment(segment) matches = manager.find_matches("hello worlds", threshold=0.8) assert len(matches) == 1 assert matches[0].match_type == MatchType.FUZZY def test_find_matches_threshold(self, manager: TranslationMemoryManager) -> None: """Test threshold filtering.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="hello world", target_text="こんにちは世界", reference=ref, ) manager.add_segment(segment) # High threshold should filter out matches = manager.find_matches("completely different text", threshold=0.9) assert len(matches) == 0 def test_find_matches_max_results(self, manager: TranslationMemoryManager) -> None: """Test max results limit.""" ref = VerseReference(book="john", chapter=1, verse=1) for i in range(10): manager.add_segment( TMSegment.create( source_text=f"test {i}", target_text=f"テスト {i}", reference=ref, ) ) matches = manager.find_matches("test 5", threshold=0.5, max_results=3) assert len(matches) <= 3 def test_find_exact_match( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test find_exact_match helper.""" manager.add_segment(sample_segment) result = manager.find_exact_match(sample_segment.source_text) assert result is not None assert result.id == sample_segment.id def test_find_exact_match_not_found( self, manager: TranslationMemoryManager ) -> None: """Test find_exact_match when no exact match.""" result = manager.find_exact_match("nonexistent text") assert result is None def test_increment_reuse( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test incrementing reuse count.""" manager.add_segment(sample_segment) assert sample_segment.reuse_count == 0 result = manager.increment_reuse(sample_segment.id) assert result is True assert manager.get_segment(sample_segment.id).reuse_count == 1 def test_increment_reuse_not_found( self, manager: TranslationMemoryManager ) -> None: """Test increment for nonexistent segment.""" result = manager.increment_reuse("nonexistent") assert result is False def test_get_statistics( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test getting statistics.""" manager.add_segment(sample_segment) stats = manager.get_statistics() assert stats.total_segments == 1 assert "john" in stats.books_covered def test_get_all_segments( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test getting all segments.""" manager.add_segment(sample_segment) all_segs = manager.get_all_segments() assert len(all_segs) == 1 def test_remove_segment( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test removing a segment.""" manager.add_segment(sample_segment) assert len(manager) == 1 result = manager.remove_segment(sample_segment.id) assert result is True assert len(manager) == 0 def test_remove_segment_not_found( self, manager: TranslationMemoryManager ) -> None: """Test removing nonexistent segment.""" result = manager.remove_segment("nonexistent") assert result is False def test_clear( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test clearing all segments.""" manager.add_segment(sample_segment) manager.clear() assert len(manager) == 0 def test_save_and_load( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test save and load roundtrip.""" manager.add_segment(sample_segment) manager.save() # Create new manager and load manager2 = TranslationMemoryManager(manager.storage_path) manager2.load() assert len(manager2) == 1 loaded = manager2.get_segment(sample_segment.id) assert loaded is not None assert loaded.source_text == sample_segment.source_text assert loaded.target_text == sample_segment.target_text def test_save_creates_directory(self, tmp_path: Path) -> None: """Test save creates parent directories.""" path = tmp_path / "nested" / "dir" / "segments.json" manager = TranslationMemoryManager(path) manager.add_from_translation( source_text="Test", target_text="テスト", reference=VerseReference(book="john", chapter=1, verse=1), ) manager.save() assert path.exists() def test_load_nonexistent(self, tmp_path: Path) -> None: """Test loading from nonexistent file.""" path = tmp_path / "nonexistent.json" manager = TranslationMemoryManager(path) manager.load() assert len(manager) == 0 def test_reload( self, manager: TranslationMemoryManager, sample_segment: TMSegment ) -> None: """Test reload discards in-memory changes.""" # Save initial state manager.add_segment(sample_segment) manager.save() # Add more in memory manager.add_from_translation( source_text="New", target_text="新規", reference=VerseReference(book="john", chapter=1, verse=2), ) assert len(manager) == 2 # Reload should restore to saved state manager.reload() assert len(manager) == 1 def test_format_matches_for_prompt(self, manager: TranslationMemoryManager) -> None: """Test formatting matches for prompt.""" ref = VerseReference(book="john", chapter=1, verse=1) segment = TMSegment.create( source_text="Test", target_text="テスト", reference=ref, ) match = TMMatch(segment=segment, similarity=0.9, match_type=MatchType.FUZZY) formatted = manager.format_matches_for_prompt([match]) assert "90%" in formatted assert "john:1:1" in formatted assert "テスト" in formatted def test_format_matches_empty(self, manager: TranslationMemoryManager) -> None: """Test formatting empty matches.""" formatted = manager.format_matches_for_prompt([]) assert "No similar translations" in formatted def test_format_matches_limit(self, manager: TranslationMemoryManager) -> None: """Test match display limit.""" matches = [] for i in range(5): ref = VerseReference(book="john", chapter=1, verse=i + 1) segment = TMSegment.create( source_text=f"Test {i}", target_text=f"テスト {i}", reference=ref, ) matches.append( TMMatch(segment=segment, similarity=0.9 - i * 0.05, match_type=MatchType.FUZZY) ) formatted = manager.format_matches_for_prompt(matches) # Should show "and X more" assert "more" in formatted class TestPromptBuilderWithTM: """Tests for PromptBuilder with Translation Memory integration.""" def test_build_prompt_with_tm_matches(self) -> None: """Test prompt includes TM matches.""" from src.context.glossary import GlossaryManager from src.pipeline.prompt import PromptBuilder from src.models.verse import ( TranslationContext, Verse, VerseReference, ) glossary = GlossaryManager() builder = PromptBuilder(glossary) ref = VerseReference(book="john", chapter=1, verse=1) verse = Verse( reference=ref, greek_text="Ἐν ἀρχῇ ἦν ὁ λόγος", ) context = TranslationContext(target_verse=verse) # Create TM match segment = TMSegment.create( source_text="Ἐν ἀρχῇ", target_text="初めに", reference=VerseReference(book="john", chapter=1, verse=1), ) match = TMMatch(segment=segment, similarity=0.8, match_type=MatchType.FUZZY) prompt = builder.build_prompt(context, tm_matches=[match]) assert "TRANSLATION MEMORY" in prompt assert "80%" in prompt assert "初めに" in prompt def test_build_prompt_without_tm(self) -> None: """Test prompt without TM matches.""" from src.context.glossary import GlossaryManager from src.pipeline.prompt import PromptBuilder from src.models.verse import ( TranslationContext, Verse, VerseReference, ) glossary = GlossaryManager() builder = PromptBuilder(glossary) ref = VerseReference(book="john", chapter=1, verse=1) verse = Verse(reference=ref, greek_text="Ἐν ἀρχῇ") context = TranslationContext(target_verse=verse) prompt = builder.build_prompt(context) assert "TRANSLATION MEMORY" not in prompt class TestConcurrentSavesNeverLoseWork: """Parallel book workers must not roll back each other's translations. Each worker holds its own manager over one shared file, loaded when its book started and saved hours later. A plain overwrite meant the last writer discarded every other book's segments — days of paid translation gone, silently, with the run still reporting success. """ @staticmethod def _ref(book: str, chapter: int, verse: int) -> VerseReference: """Build a verse reference. Args: book: Book name. chapter: Chapter number. verse: Verse number. Returns: The VerseReference. """ return VerseReference(book=book, chapter=chapter, verse=verse) def test_a_later_save_keeps_an_earlier_workers_segments(self, tmp_path): path = tmp_path / "segments.json" first = TranslationMemoryManager(path) second = TranslationMemoryManager(path) first.load() second.load() first.add_from_translation("source A", "訳A", self._ref("john", 1, 1)) second.add_from_translation("source B", "訳B", self._ref("mark", 2, 2)) first.save() second.save() final = TranslationMemoryManager(path) final.load() assert sorted(s.source_text for s in final.get_all_segments()) == [ "source A", "source B", ] def test_many_concurrent_saves_lose_nothing(self, tmp_path): import threading path = tmp_path / "segments.json" managers = [] for index in range(8): manager = TranslationMemoryManager(path) manager.load() manager.add_from_translation( f"source {index}", f"訳 {index}", self._ref("john", 1, index + 1) ) managers.append(manager) threads = [threading.Thread(target=m.save) for m in managers] for thread in threads: thread.start() for thread in threads: thread.join() final = TranslationMemoryManager(path) final.load() assert len(final) == 8 def test_a_retranslation_is_added_rather_than_replacing_the_earlier_one(self, tmp_path): # Segment ids are random UUIDs here, so re-translating the same source # yields a second segment rather than overwriting the first. That is # duplication, not loss — and loss is the property under test. Merging # must never be the thing that drops either rendering. path = tmp_path / "segments.json" first = TranslationMemoryManager(path) first.add_from_translation("same source", "古い訳", self._ref("john", 1, 1)) first.save() second = TranslationMemoryManager(path) second.add_from_translation("same source", "新しい訳", self._ref("john", 1, 1)) second.save() final = TranslationMemoryManager(path) final.load() assert sorted(s.target_text for s in final.get_all_segments()) == ["古い訳", "新しい訳"] def test_saving_the_same_manager_twice_does_not_duplicate(self, tmp_path): path = tmp_path / "segments.json" manager = TranslationMemoryManager(path) manager.add_from_translation("source", "訳", self._ref("john", 1, 1)) manager.save() manager.save() final = TranslationMemoryManager(path) final.load() assert len(final) == 1 def test_the_reference_index_survives_a_merge(self, tmp_path): path = tmp_path / "segments.json" first = TranslationMemoryManager(path) first.add_from_translation("source A", "訳A", self._ref("john", 1, 1)) first.save() second = TranslationMemoryManager(path) second.load() second.add_from_translation("source B", "訳B", self._ref("mark", 2, 2)) second.save() assert second.get_by_reference(self._ref("john", 1, 1)) assert second.get_by_reference(self._ref("mark", 2, 2)) class TestCorruptStoreIsPreserved: def test_an_unreadable_store_is_quarantined_not_overwritten(self, tmp_path, caplog): # Silently starting fresh meant the next save replaced days of # accumulated translation with whatever the process happened to hold. path = tmp_path / "segments.json" path.write_text("{ this is not json", encoding="utf-8") manager = TranslationMemoryManager(path) manager.load() manager.add_from_translation( "after corruption", "訳", VerseReference(book="luke", chapter=1, verse=1) ) manager.save() quarantined = list(tmp_path.glob("segments.corrupt.*.json")) assert len(quarantined) == 1 assert quarantined[0].read_text(encoding="utf-8") == "{ this is not json" final = TranslationMemoryManager(path) final.load() assert len(final) == 1 def test_a_missing_store_is_not_an_error(self, tmp_path): manager = TranslationMemoryManager(tmp_path / "absent.json") manager.load() assert len(manager) == 0 def test_no_stray_temp_file_survives_a_save(self, tmp_path): path = tmp_path / "segments.json" manager = TranslationMemoryManager(path) manager.add_from_translation( "source", "訳", VerseReference(book="john", chapter=1, verse=1) ) manager.save() assert not list(tmp_path.glob("*.tmp"))