From 4ca3caa86ccfb26a7b3e09f842be8149f3b9aeff Mon Sep 17 00:00:00 2001 From: busya Date: Thu, 4 Jun 2026 16:16:10 +0300 Subject: [PATCH] test(backend): add is_regex dictionary enforcement and metrics tests test_enforce_dictionary.py (new): - Verify regex patterns are matched correctly in translation enforcement - Verify invalid regex patterns are gracefully skipped test_metrics_cumulative.py (new): - Verify metrics calculations produce correct cumulative statistics test_dictionary_crud.py (extended): - test_add_entry_regex: verify creating entry with is_regex=True - test_add_entry_regex_validation: verify invalid regex raises ValueError test_dictionary_filter.py (extended): - Add test coverage for regex-based dictionary entry filtering --- .../__tests__/test_dictionary_crud.py | 55 ++++++ .../__tests__/test_dictionary_filter.py | 112 +++++++++++ .../__tests__/test_enforce_dictionary.py | 142 ++++++++++++++ .../__tests__/test_metrics_cumulative.py | 177 ++++++++++++++++++ 4 files changed, 486 insertions(+) create mode 100644 backend/src/plugins/translate/__tests__/test_enforce_dictionary.py create mode 100644 backend/src/plugins/translate/__tests__/test_metrics_cumulative.py diff --git a/backend/src/plugins/translate/__tests__/test_dictionary_crud.py b/backend/src/plugins/translate/__tests__/test_dictionary_crud.py index ea007fd8..4ca0a713 100644 --- a/backend/src/plugins/translate/__tests__/test_dictionary_crud.py +++ b/backend/src/plugins/translate/__tests__/test_dictionary_crud.py @@ -213,4 +213,59 @@ class TestDictionaryEntryCRUD: entries, total = DictionaryManager.list_entries(db_session, d.id) assert total == 2 # endregion test_same_term_different_language_pair + + # region test_add_entry_regex [C:2] [TYPE Function] + # @BRIEF Verify creating entry with is_regex=True stores and reads back correctly. + def test_add_entry_regex(self, db_session: Session): + d = DictionaryManager.create_dictionary(db_session, name="Regex Test") + entry = DictionaryManager.add_entry( + db_session, d.id, r"\d{3}-\d{2}-\d{4}", "ID-MASKED", + source_language="en", target_language="ru", is_regex=True, + ) + assert entry.is_regex is True + assert entry.source_term == r"\d{3}-\d{2}-\d{4}" + entries, total = DictionaryManager.list_entries(db_session, d.id) + assert total == 1 + assert entries[0].is_regex is True + # endregion test_add_entry_regex + + # region test_add_entry_regex_validation [C:2] [TYPE Function] + # @BRIEF Verify invalid regex pattern raises ValueError. + def test_add_entry_regex_validation(self, db_session: Session): + d = DictionaryManager.create_dictionary(db_session, name="Regex Val") + with pytest.raises(ValueError, match="Invalid regex pattern"): + DictionaryManager.add_entry( + db_session, d.id, r"[unclosed", "broken", + source_language="en", target_language="ru", is_regex=True, + ) + # endregion test_add_entry_regex_validation + + # region test_add_entry_regex_edit_toggle [C:2] [TYPE Function] + # @BRIEF Verify toggling is_regex via edit_entry works. + def test_add_entry_regex_edit_toggle(self, db_session: Session): + d = DictionaryManager.create_dictionary(db_session, name="Regex Toggle") + entry = DictionaryManager.add_entry( + db_session, d.id, "hello", "hola", + source_language="en", target_language="es", is_regex=False, + ) + assert entry.is_regex is False + + updated = DictionaryManager.edit_entry(db_session, entry.id, is_regex=True) + assert updated.is_regex is True + + updated2 = DictionaryManager.edit_entry(db_session, entry.id, is_regex=False) + assert updated2.is_regex is False + # endregion test_add_entry_regex_edit_toggle + + # region test_edit_entry_regex_validates_pattern [C:2] [TYPE Function] + # @BRIEF Verify editing source_term on a regex entry validates the pattern. + def test_edit_entry_regex_validates_pattern(self, db_session: Session): + d = DictionaryManager.create_dictionary(db_session, name="Regex Edit Val") + entry = DictionaryManager.add_entry( + db_session, d.id, r"\d+", "NUMBER", + source_language="en", target_language="ru", is_regex=True, + ) + with pytest.raises(ValueError, match="Invalid regex pattern"): + DictionaryManager.edit_entry(db_session, entry.id, source_term=r"[unclosed") + # endregion test_edit_entry_regex_validates_pattern # #endregion TestDictionaryCRUD diff --git a/backend/src/plugins/translate/__tests__/test_dictionary_filter.py b/backend/src/plugins/translate/__tests__/test_dictionary_filter.py index b5452eab..2fb31236 100644 --- a/backend/src/plugins/translate/__tests__/test_dictionary_filter.py +++ b/backend/src/plugins/translate/__tests__/test_dictionary_filter.py @@ -226,4 +226,116 @@ class TestMigration: assert entry2.source_language == "en" assert entry2.target_language == "und" # endregion test_migrate_old_entries + + +class TestFilterRegex: + """Verify regex matching in filter_for_batch.""" + + # region test_filter_regex_match [C:2] [TYPE Function] + def test_filter_regex_match(self, db_session: Session): + d = DictionaryManager.create_dictionary(db_session, name="Regex Dict") + DictionaryManager.add_entry( + db_session, d.id, r"\d{3}-\d{2}-\d{4}", "SSN-MASKED", + source_language="en", target_language="es", is_regex=True, + ) + job = TranslationJob(name="Regex Job", source_dialect="a", target_dialect="b", status="DRAFT") + db_session.add(job) + db_session.flush() + link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id) + db_session.add(link) + db_session.commit() + + result = DictionaryManager.filter_for_batch(db_session, ["My SSN is 123-45-6789"], job.id) + assert len(result) == 1 + assert result[0]["source_term"] == r"\d{3}-\d{2}-\d{4}" + assert result[0]["is_regex"] is True + assert result[0]["target_term"] == "SSN-MASKED" + # endregion test_filter_regex_match + + # region test_filter_regex_no_match [C:2] [TYPE Function] + def test_filter_regex_no_match(self, db_session: Session): + d = DictionaryManager.create_dictionary(db_session, name="Regex Dict") + DictionaryManager.add_entry( + db_session, d.id, r"\d{3}-\d{2}-\d{4}", "SSN-MASKED", + source_language="en", target_language="es", is_regex=True, + ) + job = TranslationJob(name="Regex Job", source_dialect="a", target_dialect="b", status="DRAFT") + db_session.add(job) + db_session.flush() + link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id) + db_session.add(link) + db_session.commit() + + result = DictionaryManager.filter_for_batch(db_session, ["just plain text no digits"], job.id) + assert len(result) == 0 + # endregion test_filter_regex_no_match + + # region test_filter_regex_invalid_pattern_skipped [C:2] [TYPE Function] + def test_filter_regex_invalid_pattern_skipped(self, db_session: Session): + """Invalid regex patterns should be silently skipped (no crash).""" + d = DictionaryManager.create_dictionary(db_session, name="Regex Dict") + # Add a bad regex directly to the DB to bypass CRUD validation + from src.models.translate import DictionaryEntry + entry = DictionaryEntry( + dictionary_id=d.id, source_term=r"[unclosed", source_term_normalized=r"[unclosed", + target_term="BAD", source_language="en", target_language="es", + is_regex=True, + ) + db_session.add(entry) + job = TranslationJob(name="Regex Job", source_dialect="a", target_dialect="b", status="DRAFT") + db_session.add(job) + db_session.flush() + link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id) + db_session.add(link) + db_session.commit() + + result = DictionaryManager.filter_for_batch(db_session, ["anything here"], job.id) + assert len(result) == 0 + # endregion test_filter_regex_invalid_pattern_skipped + + # region test_filter_regex_bypasses_word_boundary [C:2] [TYPE Function] + def test_filter_regex_bypasses_word_boundary(self, db_session: Session): + """Regex entries are NOT subject to word-boundary matching.""" + d = DictionaryManager.create_dictionary(db_session, name="Regex Dict") + DictionaryManager.add_entry( + db_session, d.id, r"cat", "gato", + source_language="en", target_language="es", is_regex=True, + ) + job = TranslationJob(name="Regex Job", source_dialect="a", target_dialect="b", status="DRAFT") + db_session.add(job) + db_session.flush() + link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id) + db_session.add(link) + db_session.commit() + + # "catalog" contains "cat" as substring — regex finds it, word-boundary wouldn't + result = DictionaryManager.filter_for_batch(db_session, ["catalog"], job.id) + assert len(result) == 1 + # endregion test_filter_regex_bypasses_word_boundary + + # region test_filter_regex_non_regex_still_word_boundary [C:2] [TYPE Function] + def test_filter_regex_non_regex_still_word_boundary(self, db_session: Session): + """Non-regex entries still use word-boundary matching even alongside regex entries.""" + d = DictionaryManager.create_dictionary(db_session, name="Mixed Dict") + DictionaryManager.add_entry( + db_session, d.id, "cat", "gato", + source_language="en", target_language="es", is_regex=False, + ) + DictionaryManager.add_entry( + db_session, d.id, r"\d+", "NUM", + source_language="en", target_language="es", is_regex=True, + ) + job = TranslationJob(name="Mixed Job", source_dialect="a", target_dialect="b", status="DRAFT") + db_session.add(job) + db_session.flush() + link = TranslationJobDictionary(job_id=job.id, dictionary_id=d.id) + db_session.add(link) + db_session.commit() + + result = DictionaryManager.filter_for_batch(db_session, ["catalog has 42 items"], job.id) + # regex matches \d+, non-regex should NOT match 'cat' inside 'catalog' + source_terms = [m["source_term"] for m in result] + assert r"\d+" in source_terms + assert "cat" not in source_terms + # endregion test_filter_regex_non_regex_still_word_boundary # #endregion TestDictionaryFilter diff --git a/backend/src/plugins/translate/__tests__/test_enforce_dictionary.py b/backend/src/plugins/translate/__tests__/test_enforce_dictionary.py new file mode 100644 index 00000000..2db7647a --- /dev/null +++ b/backend/src/plugins/translate/__tests__/test_enforce_dictionary.py @@ -0,0 +1,142 @@ +# #region TestEnforceDictionary [C:3] [TYPE Module] [SEMANTICS test, dictionary, enforce, regex] +# @BRIEF Validate _enforce_dictionary regex and non-regex enforcement logic. +# @RELATION BINDS_TO -> [TranslationUtils] + +from src.plugins.translate._utils import _enforce_dictionary + + +class TestEnforceDictionaryNonRegex: + """Verify non-regex enforcement still works.""" + + # region test_enforce_basic_replacement [C:2] [TYPE Function] + def test_enforce_basic_replacement(self): + """LLM left source term untranslated — enforcement should fix it.""" + source_text = "hello world" + per_lang = {"es": "hello mundo"} # "hello" wasn't translated to "hola" + matches = [ + {"source_term": "hello", "target_term": "hola", "is_regex": False}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert "hola" in per_lang["es"] + assert "hello" not in per_lang["es"] + # endregion test_enforce_basic_replacement + + # region test_enforce_target_already_present [C:2] [TYPE Function] + def test_enforce_target_already_present(self): + """If target term is already in the translation, skip replacement.""" + source_text = "hello world" + per_lang = {"es": "hola mundo"} + matches = [ + {"source_term": "hello", "target_term": "hola", "is_regex": False}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert per_lang["es"] == "hola mundo" # unchanged + # endregion test_enforce_target_already_present + + # region test_enforce_source_not_in_text [C:2] [TYPE Function] + def test_enforce_source_not_in_text(self): + """If source term is not in source text, skip.""" + source_text = "goodbye world" + per_lang = {"es": "adiós mundo"} + matches = [ + {"source_term": "hello", "target_term": "hola", "is_regex": False}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert per_lang["es"] == "adiós mundo" # unchanged + # endregion test_enforce_source_not_in_text + + # region test_enforce_empty_values_skipped [C:2] [TYPE Function] + def test_enforce_empty_values_skipped(self): + """Empty source_text or dict_matches should be no-ops.""" + per_lang = {"es": "hola mundo"} + _enforce_dictionary("", per_lang, [{"source_term": "hello", "target_term": "hola", "is_regex": False}], "b", "r") + assert per_lang["es"] == "hola mundo" + + _enforce_dictionary("hello world", per_lang, [], "b", "r") + assert per_lang["es"] == "hola mundo" + # endregion test_enforce_empty_values_skipped + + +class TestEnforceDictionaryRegex: + """Verify regex-based enforcement.""" + + # region test_enforce_regex_replacement [C:2] [TYPE Function] + def test_enforce_regex_replacement(self): + """Regex pattern matches and replaces in translation.""" + source_text = "Contact: 123-45-6789" + per_lang = {"es": "Contacto: 123-45-6789"} # SSN not masked + matches = [ + {"source_term": r"\d{3}-\d{2}-\d{4}", "target_term": "XXX-XX-XXXX", "is_regex": True}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert "XXX-XX-XXXX" in per_lang["es"] + assert "123-45-6789" not in per_lang["es"] + # endregion test_enforce_regex_replacement + + # region test_enforce_regex_no_match [C:2] [TYPE Function] + def test_enforce_regex_no_match(self): + """Regex doesn't match source — nothing changes.""" + source_text = "just text no digits" + per_lang = {"es": "solo texto sin dígitos"} + matches = [ + {"source_term": r"\d{3}-\d{2}-\d{4}", "target_term": "XXX-XX-XXXX", "is_regex": True}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert per_lang["es"] == "solo texto sin dígitos" # unchanged + # endregion test_enforce_regex_no_match + + # region test_enforce_regex_invalid_pattern_skipped [C:2] [TYPE Function] + def test_enforce_regex_invalid_pattern_skipped(self): + """Invalid regex pattern in dict match should be silently skipped.""" + source_text = "some text" + per_lang = {"es": "algún texto"} + matches = [ + {"source_term": r"[invalid", "target_term": "FIXED", "is_regex": True}, + ] + # Should not raise + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert per_lang["es"] == "algún texto" # unchanged + # endregion test_enforce_regex_invalid_pattern_skipped + + # region test_enforce_regex_target_already_present [C:2] [TYPE Function] + def test_enforce_regex_target_already_present(self): + """If target term is already in translation, regex enforcement skips.""" + source_text = "Call 555-1234" + per_lang = {"es": "Llama al NUM"} + matches = [ + {"source_term": r"\d{3}-\d{4}", "target_term": "NUM", "is_regex": True}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert per_lang["es"] == "Llama al NUM" # unchanged + # endregion test_enforce_regex_target_already_present + + # region test_enforce_regex_multi_language [C:2] [TYPE Function] + def test_enforce_regex_multi_language(self): + """Regex enforcement applies to all languages.""" + source_text = "Price: 99.99" + per_lang = { + "es": "Precio: 99.99 euros", + "fr": "Prix: 99.99 euros", + } + matches = [ + {"source_term": r"\d+\.\d{2}", "target_term": "XX.YY", "is_regex": True}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert "XX.YY" in per_lang["es"] + assert "XX.YY" in per_lang["fr"] + assert "99.99" not in per_lang["es"] + assert "99.99" not in per_lang["fr"] + # endregion test_enforce_regex_multi_language + + # region test_enforce_regex_case_sensitive_matching [C:2] [TYPE Function] + def test_enforce_regex_case_insensitive(self): + """Regex enforcement uses IGNORECASE flag.""" + source_text = "HELLO World" + per_lang = {"es": "HELLO Mundo"} # "HELLO" not translated + matches = [ + {"source_term": "hello", "target_term": "hola", "is_regex": True}, + ] + _enforce_dictionary(source_text, per_lang, matches, "batch_1", "row_1") + assert "hola" in per_lang["es"] + # endregion test_enforce_regex_case_insensitive +# #endregion TestEnforceDictionary diff --git a/backend/src/plugins/translate/__tests__/test_metrics_cumulative.py b/backend/src/plugins/translate/__tests__/test_metrics_cumulative.py new file mode 100644 index 00000000..da0d187a --- /dev/null +++ b/backend/src/plugins/translate/__tests__/test_metrics_cumulative.py @@ -0,0 +1,177 @@ +# #region TestMetricsCumulative [C:3] [TYPE Module] [SEMANTICS test, metrics, cumulative, tokens] +# @BRIEF Validate cumulative_tokens and cumulative_cost aggregation in TranslationMetrics. +# @RELATION BINDS_TO -> [TranslationMetrics] + +import pytest +from datetime import datetime, timezone + +from sqlalchemy import create_engine, event +from sqlalchemy.orm import Session, sessionmaker + +from src.models.translate import ( + Base, + TranslationJob, + TranslationRun, + TranslationRunLanguageStats, +) +from src.plugins.translate.metrics import TranslationMetrics +import uuid + + +# region db_session [TYPE Fixture] +@pytest.fixture +def db_session(): + engine = create_engine("sqlite:///:memory", echo=False) + + @event.listens_for(engine, "connect") + def _set_sqlite_pragma(dbapi_connection, connection_record): + cursor = dbapi_connection.cursor() + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + + Base.metadata.create_all(bind=engine) + session = sessionmaker(bind=engine)() + try: + yield session + finally: + session.close() + Base.metadata.drop_all(bind=engine) +# endregion db_session + + +class TestCumulativeTokens: + """Verify cumulative_tokens aggregation across runs.""" + + # region test_cumulative_tokens_single_run [C:2] [TYPE Function] + def test_cumulative_tokens_single_run(self, db_session: Session): + job = TranslationJob(id=str(uuid.uuid4()), name="Test Job", source_dialect="a", target_dialect="b", status="ACTIVE", created_by="test") + db_session.add(job) + db_session.flush() + + run = TranslationRun( + id=str(uuid.uuid4()), job_id=job.id, + status="COMPLETED", + started_at=datetime.now(timezone.utc), + created_at=datetime.now(timezone.utc), + ) + db_session.add(run) + db_session.flush() + + stats = TranslationRunLanguageStats( + run_id=run.id, language_code="en", + token_count=1500, estimated_cost=0.003, + translated_rows=50, + ) + db_session.add(stats) + db_session.commit() + + metrics = TranslationMetrics(db_session).get_job_metrics(job.id) + assert metrics["cumulative_tokens"] == 1500 + assert metrics["cumulative_cost"] == 0.003 + assert metrics["per_language_metrics"]["en"]["tokens"] == 1500 + # endregion test_cumulative_tokens_single_run + + # region test_cumulative_tokens_multi_run [C:2] [TYPE Function] + def test_cumulative_tokens_multi_run(self, db_session: Session): + job = TranslationJob(id=str(uuid.uuid4()), name="Multi Run Job", source_dialect="a", target_dialect="b", status="ACTIVE", created_by="test") + db_session.add(job) + db_session.flush() + + for i in range(3): + run = TranslationRun( + id=str(uuid.uuid4()), job_id=job.id, + status="COMPLETED", + started_at=datetime.now(timezone.utc), + created_at=datetime.now(timezone.utc), + ) + db_session.add(run) + db_session.flush() + + stats = TranslationRunLanguageStats( + run_id=run.id, language_code="en", + token_count=1000 * (i + 1), + estimated_cost=0.001 * (i + 1), + translated_rows=10 * (i + 1), + ) + db_session.add(stats) + + db_session.commit() + + metrics = TranslationMetrics(db_session).get_job_metrics(job.id) + # tokens: 1000 + 2000 + 3000 = 6000 + assert metrics["cumulative_tokens"] == 6000 + # cost: 0.001 + 0.002 + 0.003 = 0.006 + assert abs(metrics["cumulative_cost"] - 0.006) < 1e-10 + # endregion test_cumulative_tokens_multi_run + + # region test_cumulative_tokens_multi_language [C:2] [TYPE Function] + def test_cumulative_tokens_multi_language(self, db_session: Session): + job = TranslationJob(id=str(uuid.uuid4()), name="Multi Lang Job", source_dialect="a", target_dialect="b", status="ACTIVE", created_by="test") + db_session.add(job) + db_session.flush() + + run = TranslationRun( + id=str(uuid.uuid4()), job_id=job.id, + status="COMPLETED", + started_at=datetime.now(timezone.utc), + created_at=datetime.now(timezone.utc), + ) + db_session.add(run) + db_session.flush() + + db_session.add(TranslationRunLanguageStats( + run_id=run.id, language_code="en", + token_count=5000, estimated_cost=0.01, translated_rows=100, + )) + db_session.add(TranslationRunLanguageStats( + run_id=run.id, language_code="ru", + token_count=3000, estimated_cost=0.006, translated_rows=60, + )) + db_session.commit() + + metrics = TranslationMetrics(db_session).get_job_metrics(job.id) + assert metrics["cumulative_tokens"] == 8000 # 5000 + 3000 + assert abs(metrics["cumulative_cost"] - 0.016) < 1e-10 + # endregion test_cumulative_tokens_multi_language + + # region test_cumulative_tokens_no_runs [C:2] [TYPE Function] + def test_cumulative_tokens_no_runs(self, db_session: Session): + job = TranslationJob(id=str(uuid.uuid4()), name="No Runs Job", source_dialect="a", target_dialect="b", status="DRAFT", created_by="test") + db_session.add(job) + db_session.commit() + + metrics = TranslationMetrics(db_session).get_job_metrics(job.id) + assert metrics["cumulative_tokens"] == 0 + assert metrics["cumulative_cost"] == 0.0 + assert metrics["per_language_metrics"] == {} + # endregion test_cumulative_tokens_no_runs + + # region test_cumulative_tokens_with_null_values [C:2] [TYPE Function] + def test_cumulative_tokens_with_null_values(self, db_session: Session): + job = TranslationJob(id=str(uuid.uuid4()), name="Null Tokens", source_dialect="a", target_dialect="b", status="ACTIVE", created_by="test") + db_session.add(job) + db_session.flush() + + run = TranslationRun( + id=str(uuid.uuid4()), job_id=job.id, + status="COMPLETED", + started_at=datetime.now(timezone.utc), + created_at=datetime.now(timezone.utc), + ) + db_session.add(run) + db_session.commit() + + # Add stats with null token_count and estimated_cost + stats = TranslationRunLanguageStats( + run_id=run.id, language_code="en", + token_count=None, estimated_cost=None, + translated_rows=10, + ) + db_session.add(stats) + db_session.commit() + + metrics = TranslationMetrics(db_session).get_job_metrics(job.id) + assert metrics["cumulative_tokens"] == 0 # None treated as 0 + assert metrics["cumulative_cost"] == 0.0 + # endregion test_cumulative_tokens_with_null_values +# #endregion TestMetricsCumulative