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
This commit is contained in:
2026-06-04 16:16:10 +03:00
parent a1a855e670
commit 4ca3caa86c
4 changed files with 486 additions and 0 deletions

View File

@@ -213,4 +213,59 @@ class TestDictionaryEntryCRUD:
entries, total = DictionaryManager.list_entries(db_session, d.id) entries, total = DictionaryManager.list_entries(db_session, d.id)
assert total == 2 assert total == 2
# endregion test_same_term_different_language_pair # 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 # #endregion TestDictionaryCRUD

View File

@@ -226,4 +226,116 @@ class TestMigration:
assert entry2.source_language == "en" assert entry2.source_language == "en"
assert entry2.target_language == "und" assert entry2.target_language == "und"
# endregion test_migrate_old_entries # 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 # #endregion TestDictionaryFilter

View File

@@ -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

View File

@@ -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