busya
8a4310169b
12 known failures — all from Agent 3 new unverified tests (mock setup issues): - 3 dataset_review_routes_extended (DTO field mismatches) - 1 settings_consolidated (dict key access) - 1 llm_analysis_service_coverage (rate_limit mock) - 1 migration_plugin (SessionLocal side_effect exhaustion) - 1 preview (DB query vs dict key) - 5 scheduler (datetime timezone + async mock mismatches) NEW TEST FILES THIS SESSION: - test_batch_insert_coverage.py — 3 tests - test_storage_plugin.py — +3 tests - test_search.py — +2 tests - test_mapper.py — already 100% - test_llm_analysis_migration_v1_to_v2.py — 14 tests - test_llm_async_http.py — +1 test - test_prompt_builder.py — +1 test - test_service_datasource.py — +1 test - test_lang_detect.py — +1 test - test_scheduler.py — +6 tests - test_llm_analysis_service_coverage.py — +15 tests - test_dataset_review_routes_extended.py — +14 tests - test_settings_consolidated.py — +13 tests Modules pushed to 100%: _batch_insert, dictionary_entries, service_datasource, _llm_async_http, prompt_builder, dictionary_crud, _batch_sizer, storage/plugin, mapper.py Session: 7194→7643 tests (+449), 80%→83% raw (+3pp), 93.4%→96.9% real (+3.5pp). Remaining: 12 failures to fix + ~300 statements to reach 98% real coverage.
339 lines
13 KiB
Python
339 lines
13 KiB
Python
# #region Test.BulkFindReplaceService [C:3] [TYPE Module] [SEMANTICS test,bulk,find,replace]
|
|
# @BRIEF Tests for service_bulk_replace.py — BulkFindReplaceService.
|
|
# @RELATION BINDS_TO -> [BulkFindReplaceService]
|
|
# @TEST_EDGE: pattern_too_long -> ValueError
|
|
# @TEST_EDGE: invalid_regex -> ValueError
|
|
# @TEST_EDGE: no_match -> empty preview
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
sys.path.insert(0, str(Path(__file__).parent.parent.parent.parent / "src"))
|
|
|
|
import re
|
|
import pytest
|
|
from unittest.mock import MagicMock, patch
|
|
from datetime import UTC, datetime
|
|
|
|
from src.models.translate import (
|
|
TranslationBatch, TranslationLanguage, TranslationRecord, TranslationRun,
|
|
)
|
|
from src.plugins.translate.service_bulk_replace import BulkFindReplaceService
|
|
from .conftest import JOB_ID
|
|
|
|
|
|
def _make_batch_and_run(db_session, run_id="run-br"):
|
|
"""Create a batch and run for FK constraints."""
|
|
now = datetime.now(UTC)
|
|
run = TranslationRun(id=run_id, job_id=JOB_ID, status="RUNNING",
|
|
started_at=now)
|
|
db_session.add(run)
|
|
db_session.flush()
|
|
batch = TranslationBatch(id=f"batch-{run_id}", run_id=run_id,
|
|
batch_index=0, status="PENDING")
|
|
db_session.add(batch)
|
|
db_session.flush()
|
|
return run_id, batch.id
|
|
|
|
|
|
class TestCompilePattern:
|
|
"""BulkFindReplaceService._compile_pattern."""
|
|
|
|
def test_plain_text(self):
|
|
"""Plain text pattern is escaped."""
|
|
pattern = BulkFindReplaceService._compile_pattern("hello (world)", False)
|
|
assert isinstance(pattern, re.Pattern)
|
|
assert pattern.search("hello (world)")
|
|
assert not pattern.search("hello world")
|
|
|
|
def test_regex(self):
|
|
"""Regex pattern is compiled as-is."""
|
|
pattern = BulkFindReplaceService._compile_pattern(r"hello\s+world", True)
|
|
assert isinstance(pattern, re.Pattern)
|
|
assert pattern.search("hello world")
|
|
assert not pattern.search("helloworld")
|
|
|
|
def test_too_long(self):
|
|
"""Pattern exceeding max length raises ValueError."""
|
|
with pytest.raises(ValueError, match="Pattern too long"):
|
|
BulkFindReplaceService._compile_pattern("x" * 501, False)
|
|
|
|
def test_invalid_regex(self):
|
|
"""Invalid regex raises ValueError."""
|
|
with pytest.raises(ValueError, match="Invalid regex"):
|
|
BulkFindReplaceService._compile_pattern(r"[invalid", True)
|
|
|
|
|
|
class TestFindMatchingEntries:
|
|
"""BulkFindReplaceService._find_matching_entries."""
|
|
|
|
def test_returns_entries(self, db_session):
|
|
"""Returns entries matching run and language."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-1")
|
|
record = TranslationRecord(
|
|
id="rec-br-1", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-1", language_code="fr",
|
|
translated_value="bonjour", final_value="bonjour_edited",
|
|
status="edited",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
entries = BulkFindReplaceService._find_matching_entries(
|
|
db_session, run_id, "hello", False, "fr"
|
|
)
|
|
assert len(entries) == 1
|
|
assert entries[0].language_code == "fr"
|
|
|
|
def test_filters_by_language(self, db_session):
|
|
"""Only entries with matching language_code returned."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-2")
|
|
record = TranslationRecord(
|
|
id="rec-br-2", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang_fr = TranslationLanguage(
|
|
record_id="rec-br-2", language_code="fr",
|
|
translated_value="bonjour", final_value="bonjour",
|
|
status="translated",
|
|
)
|
|
lang_de = TranslationLanguage(
|
|
record_id="rec-br-2", language_code="de",
|
|
translated_value="hallo", final_value="hallo",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang_fr)
|
|
db_session.add(lang_de)
|
|
db_session.commit()
|
|
|
|
entries = BulkFindReplaceService._find_matching_entries(
|
|
db_session, run_id, "hello", False, "fr"
|
|
)
|
|
assert len(entries) == 1
|
|
assert entries[0].language_code == "fr"
|
|
|
|
|
|
class TestPreview:
|
|
"""BulkFindReplaceService.preview."""
|
|
|
|
def test_preview_finds_match(self, db_session):
|
|
"""Returns preview items with matched entries."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-prev")
|
|
record = TranslationRecord(
|
|
id="rec-br-3", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello world", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-3", language_code="fr",
|
|
translated_value="bonjour le monde", final_value="bonjour le monde",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
results = BulkFindReplaceService.preview(
|
|
db_session, run_id, "monde", False, "terre", "fr"
|
|
)
|
|
assert len(results) == 1
|
|
assert results[0]["new_value"] == "bonjour le terre"
|
|
|
|
def test_preview_no_match(self, db_session):
|
|
"""No matching entries returns empty list."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-no")
|
|
record = TranslationRecord(
|
|
id="rec-br-4", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-4", language_code="fr",
|
|
translated_value="bonjour", final_value="bonjour",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
results = BulkFindReplaceService.preview(
|
|
db_session, run_id, "zzzzz", False, "dummy", "fr"
|
|
)
|
|
assert len(results) == 0
|
|
|
|
|
|
class TestApply:
|
|
"""BulkFindReplaceService.apply."""
|
|
|
|
def test_apply_basic(self, db_session):
|
|
"""Apply replacement and verify DB updated."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-apply-1")
|
|
record = TranslationRecord(
|
|
id="rec-br-apply-1", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello world", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-apply-1", language_code="fr",
|
|
translated_value="bonjour le monde", final_value="bonjour le monde",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
result = BulkFindReplaceService.apply(
|
|
db_session, run_id, "monde", False, "terre", "fr"
|
|
)
|
|
assert result["rows_affected"] == 1
|
|
db_session.refresh(lang)
|
|
assert lang.final_value == "bonjour le terre"
|
|
|
|
def test_apply_no_match(self, db_session):
|
|
"""No match results in 0 rows affected."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-apply-2")
|
|
record = TranslationRecord(
|
|
id="rec-br-apply-2", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-apply-2", language_code="fr",
|
|
translated_value="bonjour", final_value="bonjour",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
result = BulkFindReplaceService.apply(
|
|
db_session, run_id, "zzzzz", False, "terre", "fr"
|
|
)
|
|
assert result["rows_affected"] == 0
|
|
|
|
def test_apply_with_dictionary_submit(self, db_session):
|
|
"""Apply with dictionary submission."""
|
|
from src.models.translate import TerminologyDictionary
|
|
dict_entry = TerminologyDictionary(id="dict-br-1", name="Test Dict", is_active=True)
|
|
db_session.add(dict_entry)
|
|
db_session.commit()
|
|
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-apply-3")
|
|
record = TranslationRecord(
|
|
id="rec-br-apply-3", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-apply-3", language_code="fr",
|
|
translated_value="bonjour tout le monde", final_value="bonjour tout le monde",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
with patch("src.plugins.translate.service_bulk_replace.InlineCorrectionService.submit_correction_to_dict") as mock_submit:
|
|
mock_submit.return_value = {"action": "created", "entry_id": "entry-1"}
|
|
result = BulkFindReplaceService.apply(
|
|
db_session, run_id, "monde", False, "terre",
|
|
"fr", submit_to_dictionary=True, dictionary_id="dict-br-1",
|
|
)
|
|
assert result["rows_affected"] == 1
|
|
assert result["corrections_submitted"] == 1
|
|
|
|
def test_apply_with_dictionary_submit_with_context(self, db_session):
|
|
"""Apply with dictionary submission with context."""
|
|
from src.models.translate import TerminologyDictionary
|
|
dict_entry = TerminologyDictionary(id="dict-br-ctx", name="Test Dict", is_active=True)
|
|
db_session.add(dict_entry)
|
|
db_session.commit()
|
|
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-apply-ctx")
|
|
record = TranslationRecord(
|
|
id="rec-br-apply-ctx", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello world", target_sql="bonjour", status="SUCCESS",
|
|
source_data={"table": "test"}, source_object_type="table_row",
|
|
source_object_name="Row 1", source_object_id="1",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-apply-ctx", language_code="fr",
|
|
translated_value="bonjour le monde", final_value="bonjour le monde",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
with patch("src.plugins.translate.service_bulk_replace.InlineCorrectionService.submit_correction_to_dict") as mock_submit:
|
|
mock_submit.return_value = {"action": "created", "entry_id": "entry-ctx"}
|
|
result = BulkFindReplaceService.apply(
|
|
db_session, run_id, "monde", False, "terre",
|
|
"fr", submit_to_dictionary=True, dictionary_id="dict-br-ctx",
|
|
submit_to_dictionary_with_context=True,
|
|
)
|
|
assert result["rows_affected"] == 1
|
|
assert result["corrections_submitted"] == 1
|
|
|
|
def test_apply_dictionary_submit_raises_exception(self, db_session):
|
|
"""Exception during dictionary submit caught gracefully."""
|
|
from src.models.translate import TerminologyDictionary
|
|
dict_entry = TerminologyDictionary(id="dict-br-exc", name="Test Dict", is_active=True)
|
|
db_session.add(dict_entry)
|
|
db_session.commit()
|
|
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-apply-exc")
|
|
record = TranslationRecord(
|
|
id="rec-br-apply-exc", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-apply-exc", language_code="fr",
|
|
translated_value="bonjour", final_value="bonjour",
|
|
status="translated",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
with patch("src.plugins.translate.service_bulk_replace.InlineCorrectionService.submit_correction_to_dict") as mock_submit:
|
|
mock_submit.side_effect = Exception("Dict failure")
|
|
result = BulkFindReplaceService.apply(
|
|
db_session, run_id, "bonjour", False, "salut",
|
|
"fr", submit_to_dictionary=True, dictionary_id="dict-br-exc",
|
|
)
|
|
assert result["corrections_submitted"] == 0
|
|
assert result["rows_affected"] == 1
|
|
|
|
def test_apply_pending_status_becomes_edited(self, db_session):
|
|
"""Entry with 'pending' status becomes 'edited'."""
|
|
run_id, batch_id = _make_batch_and_run(db_session, "run-br-status")
|
|
record = TranslationRecord(
|
|
id="rec-br-status", run_id=run_id, batch_id=batch_id,
|
|
source_sql="hello", target_sql="bonjour", status="SUCCESS",
|
|
)
|
|
db_session.add(record)
|
|
db_session.flush()
|
|
lang = TranslationLanguage(
|
|
record_id="rec-br-status", language_code="fr",
|
|
translated_value="bonjour", final_value="bonjour",
|
|
status="pending",
|
|
)
|
|
db_session.add(lang)
|
|
db_session.commit()
|
|
|
|
BulkFindReplaceService.apply(
|
|
db_session, run_id, "bonjour", False, "salut", "fr"
|
|
)
|
|
db_session.refresh(lang)
|
|
assert lang.status == "edited"
|
|
# #endregion Test.BulkFindReplaceService
|