ss-tools/backend/tests/integration/test_translate_dictionary_integration.py

# #region DictionaryIntegrationTests [C:4] [TYPE Module] [SEMANTICS test, integration, dictionary, postgres, testcontainers]
# @BRIEF Integration tests for DictionaryCRUD and DictionaryEntryCRUD with real PostgreSQL.
# @RELATION BINDS_TO -> [DictionaryCRUD]
# @RELATION BINDS_TO -> [DictionaryEntryCRUD]
# @RELATION BINDS_TO -> [TerminologyDictionary]
# @RELATION BINDS_TO -> [DictionaryEntry]
# @TEST_CONTRACT DictionaryCRUD ->
# {
#   invariants: [
#     "create_dictionary persists with all fields",
#     "update_dictionary modifies only specified fields",
#     "delete_dictionary cascades to entries and job associations",
#     "delete_dictionary blocked when attached to active jobs",
#     "add_entry enforces uniqueness on (dict_id, source_term_norm, source_lang, target_lang)",
#     "same term with different language pairs allowed"
#   ]
# }
# @TEST_EDGE duplicate_entry -> ValueError on repeated normalized term
# @TEST_EDGE delete_active_job -> ValueError with active/scheduled message
# @TEST_EDGE same_term_different_lang_pair -> allowed (not duplicate)
# @TEST_EDGE invalid_regex -> ValueError on invalid regex pattern
# @TEST_INVARIANT unique_normalized -> VERIFIED_BY: [test_add_entry_duplicate_normalized, test_same_term_different_lang_pair_allowed]
# @TEST_INVARIANT cascade_integrity -> VERIFIED_BY: [test_delete_dictionary_cascades_to_entries]

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent.parent.parent / "src"))

import pytest
from sqlalchemy.orm import Session

from src.models.translate import (
    DictionaryEntry,
    TerminologyDictionary,
    TranslationJob,
    TranslationJobDictionary,
)
from src.plugins.translate.dictionary import DictionaryManager


# #region TestDictionaryCRUDIntegration [C:3] [TYPE Class]
# @BRIEF Integration tests for dictionary-level CRUD with real PostgreSQL.
class TestDictionaryCRUDIntegration:
    """Verify dictionary CRUD operations with real PostgreSQL."""

    # region test_create_dictionary_persists_all_fields [C:2] [TYPE Function]
    # @BRIEF Verify dictionary creation persists all fields.
    def test_create_dictionary_persists_all_fields(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(
            db_session,
            name="Finance Terms",
            description="Financial terminology mappings",
            created_by="test_user",
            is_active=True,
        )

        assert dictionary.id is not None
        assert dictionary.name == "Finance Terms"
        assert dictionary.description == "Financial terminology mappings"
        assert dictionary.created_by == "test_user"
        assert dictionary.is_active is True
        assert dictionary.created_at is not None

        # Verify persisted
        fetched = DictionaryManager.get_dictionary(db_session, dictionary.id)
        assert fetched.id == dictionary.id
        assert fetched.name == "Finance Terms"
    # endregion test_create_dictionary_persists_all_fields

    # region test_update_dictionary_partial_update [C:2] [TYPE Function]
    # @BRIEF Verify partial update only modifies specified fields.
    def test_update_dictionary_partial_update(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(
            db_session, name="Original", description="Original desc", created_by="user1",
        )
        original_id = dictionary.id
        original_created_by = dictionary.created_by

        # Update only name
        updated = DictionaryManager.update_dictionary(
            db_session, original_id, name="Updated Name",
        )

        assert updated.id == original_id
        assert updated.name == "Updated Name"
        assert updated.description == "Original desc"  # Unchanged
        assert updated.created_by == original_created_by  # Unchanged
    # endregion test_update_dictionary_partial_update

    # region test_delete_dictionary_cascades_to_entries [C:2] [TYPE Function]
    # @BRIEF Verify deleting dictionary removes all entries.
    def test_delete_dictionary_cascades_to_entries(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="To Delete")

        # Add entries
        DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "hola",
            source_language="en", target_language="es",
        )
        DictionaryManager.add_entry(
            db_session, dictionary.id, "world", "mundo",
            source_language="en", target_language="es",
        )

        # Verify entries exist
        entries, total = DictionaryManager.list_entries(db_session, dictionary.id)
        assert total == 2

        # Delete dictionary
        DictionaryManager.delete_dictionary(db_session, dictionary.id)

        # Verify dictionary gone
        with pytest.raises(ValueError, match="Dictionary not found"):
            DictionaryManager.get_dictionary(db_session, dictionary.id)

        # Verify entries gone (FK cascade)
        remaining_entries = (
            db_session.query(DictionaryEntry)
            .filter(DictionaryEntry.dictionary_id == dictionary.id)
            .all()
        )
        assert len(remaining_entries) == 0
    # endregion test_delete_dictionary_cascades_to_entries

    # region test_delete_dictionary_blocked_by_active_job [C:2] [TYPE Function]
    # @BRIEF Verify deletion blocked when attached to active/scheduled jobs.
    def test_delete_dictionary_blocked_by_active_job(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Protected")

        # Create active job with association
        job = TranslationJob(
            name="Active Job",
            source_dialect="postgresql",
            target_dialect="clickhouse",
            status="ACTIVE",
            created_by="test_user",
        )
        db_session.add(job)
        db_session.flush()

        link = TranslationJobDictionary(job_id=job.id, dictionary_id=dictionary.id)
        db_session.add(link)
        db_session.commit()

        # Attempt delete should fail
        with pytest.raises(ValueError, match="active/scheduled"):
            DictionaryManager.delete_dictionary(db_session, dictionary.id)

        # Verify dictionary still exists
        fetched = DictionaryManager.get_dictionary(db_session, dictionary.id)
        assert fetched is not None
    # endregion test_delete_dictionary_blocked_by_active_job

    # region test_delete_dictionary_blocked_by_scheduled_job [C:2] [TYPE Function]
    # @BRIEF Verify deletion blocked when attached to SCHEDULED jobs.
    def test_delete_dictionary_blocked_by_scheduled_job(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Scheduled Protected")

        # Create scheduled job with association
        job = TranslationJob(
            name="Scheduled Job",
            source_dialect="postgresql",
            target_dialect="clickhouse",
            status="SCHEDULED",
            created_by="test_user",
        )
        db_session.add(job)
        db_session.flush()

        link = TranslationJobDictionary(job_id=job.id, dictionary_id=dictionary.id)
        db_session.add(link)
        db_session.commit()

        with pytest.raises(ValueError, match="active/scheduled"):
            DictionaryManager.delete_dictionary(db_session, dictionary.id)
    # endregion test_delete_dictionary_blocked_by_scheduled_job

    # region test_delete_dictionary_allowed_with_completed_job [C:2] [TYPE Function]
    # @BRIEF Verify deletion allowed when only completed/failed jobs reference it.
    def test_delete_dictionary_allowed_with_completed_job(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Completed OK")

        # Create completed job with association
        job = TranslationJob(
            name="Completed Job",
            source_dialect="postgresql",
            target_dialect="clickhouse",
            status="COMPLETED",
            created_by="test_user",
        )
        db_session.add(job)
        db_session.flush()

        link = TranslationJobDictionary(job_id=job.id, dictionary_id=dictionary.id)
        db_session.add(link)
        db_session.commit()

        # Should succeed
        DictionaryManager.delete_dictionary(db_session, dictionary.id)

        with pytest.raises(ValueError, match="Dictionary not found"):
            DictionaryManager.get_dictionary(db_session, dictionary.id)
    # endregion test_delete_dictionary_allowed_with_completed_job

    # region test_list_dictionaries_pagination [C:2] [TYPE Function]
    # @BRIEF Verify paginated listing works correctly.
    def test_list_dictionaries_pagination(self, db_session: Session):
        # Create 7 dictionaries
        for i in range(7):
            DictionaryManager.create_dictionary(db_session, name=f"Dict {i}")

        # Page 1
        dicts, total = DictionaryManager.list_dictionaries(db_session, page=1, page_size=3)
        assert total == 7
        assert len(dicts) == 3

        # Page 2
        dicts, total = DictionaryManager.list_dictionaries(db_session, page=2, page_size=3)
        assert total == 7
        assert len(dicts) == 3

        # Page 3
        dicts, total = DictionaryManager.list_dictionaries(db_session, page=3, page_size=3)
        assert total == 7
        assert len(dicts) == 1
    # endregion test_list_dictionaries_pagination

# #endregion TestDictionaryCRUDIntegration


# #region TestDictionaryEntryCRUDIntegration [C:3] [TYPE Class]
# @BRIEF Integration tests for entry-level CRUD with real PostgreSQL.
class TestDictionaryEntryCRUDIntegration:
    """Verify entry CRUD operations with real PostgreSQL."""

    # region test_add_entry_persists_all_fields [C:2] [TYPE Function]
    # @BRIEF Verify entry creation persists all fields.
    def test_add_entry_persists_all_fields(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Test")

        entry = DictionaryManager.add_entry(
            db_session,
            dictionary.id,
            "hello world",
            "привет мир",
            source_language="en",
            target_language="ru",
            context_notes="greeting",
            is_regex=False,
        )

        assert entry.id is not None
        assert entry.dictionary_id == dictionary.id
        assert entry.source_term == "hello world"
        assert entry.target_term == "привет мир"
        assert entry.source_language == "en"
        assert entry.target_language == "ru"
        assert entry.context_notes == "greeting"
        assert entry.is_regex is False

        # Verify persisted
        entries, total = DictionaryManager.list_entries(db_session, dictionary.id)
        assert total == 1
        assert entries[0].source_term == "hello world"
    # endregion test_add_entry_persists_all_fields

    # region test_add_entry_duplicate_normalized [C:2] [TYPE Function]
    # @BRIEF Verify duplicate detection uses normalized (lowercased) term.
    def test_add_entry_duplicate_normalized(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Norm Test")

        DictionaryManager.add_entry(
            db_session, dictionary.id, "Hello", "Hola",
            source_language="en", target_language="es",
        )

        # Same term, different case — should be duplicate
        with pytest.raises(ValueError, match="already exists"):
            DictionaryManager.add_entry(
                db_session, dictionary.id, "HELLO", "Bonjour",
                source_language="en", target_language="es",
            )

        with pytest.raises(ValueError, match="already exists"):
            DictionaryManager.add_entry(
                db_session, dictionary.id, "hello", "Ciao",
                source_language="en", target_language="es",
            )
    # endregion test_add_entry_duplicate_normalized

    # region test_same_term_different_lang_pair_allowed [C:2] [TYPE Function]
    # @BRIEF Verify same term with different language pair is allowed.
    def test_same_term_different_lang_pair_allowed(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Multi Lang")

        entry1 = DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "привет",
            source_language="en", target_language="ru",
        )
        entry2 = DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "hallo",
            source_language="en", target_language="de",
        )
        entry3 = DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "bonjour",
            source_language="en", target_language="fr",
        )

        assert entry1.id != entry2.id != entry3.id
        assert entry1.target_language == "ru"
        assert entry2.target_language == "de"
        assert entry3.target_language == "fr"

        entries, total = DictionaryManager.list_entries(db_session, dictionary.id)
        assert total == 3
    # endregion test_same_term_different_lang_pair_allowed

    # region test_same_term_different_dictionary_allowed [C:2] [TYPE Function]
    # @BRIEF Verify same term in different dictionaries is allowed.
    def test_same_term_different_dictionary_allowed(self, db_session: Session):
        dict1 = DictionaryManager.create_dictionary(db_session, name="Dict 1")
        dict2 = DictionaryManager.create_dictionary(db_session, name="Dict 2")

        entry1 = DictionaryManager.add_entry(
            db_session, dict1.id, "hello", "hola",
            source_language="en", target_language="es",
        )
        entry2 = DictionaryManager.add_entry(
            db_session, dict2.id, "hello", "bonjour",
            source_language="en", target_language="fr",
        )

        assert entry1.id != entry2.id
        assert entry1.dictionary_id == dict1.id
        assert entry2.dictionary_id == dict2.id
    # endregion test_same_term_different_dictionary_allowed

    # region test_edit_entry_updates_fields [C:2] [TYPE Function]
    # @BRIEF Verify editing entry updates specified fields.
    def test_edit_entry_updates_fields(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Edit Test")

        entry = DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "hola",
            source_language="en", target_language="es",
            context_notes="greeting",
        )

        updated = DictionaryManager.edit_entry(
            db_session, entry.id,
            target_term="HOLA!",
            context_notes="formal greeting",
        )

        assert updated.target_term == "HOLA!"
        assert updated.context_notes == "formal greeting"
        assert updated.source_term == "hello"  # Unchanged
    # endregion test_edit_entry_updates_fields

    # region test_edit_entry_source_term_checks_uniqueness [C:2] [TYPE Function]
    # @BRIEF Verify editing source_term checks uniqueness constraint.
    def test_edit_entry_source_term_checks_uniqueness(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Uniq Test")

        DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "hola",
            source_language="en", target_language="es",
        )
        entry2 = DictionaryManager.add_entry(
            db_session, dictionary.id, "world", "mundo",
            source_language="en", target_language="es",
        )

        # Try to change entry2's source to "hello" — should fail
        with pytest.raises(ValueError, match="already exists"):
            DictionaryManager.edit_entry(db_session, entry2.id, source_term="HELLO")
    # endregion test_edit_entry_source_term_checks_uniqueness

    # region test_delete_entry [C:2] [TYPE Function]
    # @BRIEF Verify entry deletion.
    def test_delete_entry(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Delete Test")

        entry = DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "hola",
            source_language="en", target_language="es",
        )

        DictionaryManager.delete_entry(db_session, entry.id)

        entries, total = DictionaryManager.list_entries(db_session, dictionary.id)
        assert total == 0
    # endregion test_delete_entry

    # region test_clear_entries [C:2] [TYPE Function]
    # @BRIEF Verify clearing all entries for a dictionary.
    def test_clear_entries(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Clear Test")

        DictionaryManager.add_entry(
            db_session, dictionary.id, "hello", "hola",
            source_language="en", target_language="es",
        )
        DictionaryManager.add_entry(
            db_session, dictionary.id, "world", "mundo",
            source_language="en", target_language="es",
        )
        DictionaryManager.add_entry(
            db_session, dictionary.id, "goodbye", "adiós",
            source_language="en", target_language="es",
        )

        deleted = DictionaryManager.clear_entries(db_session, dictionary.id)
        assert deleted == 3

        entries, total = DictionaryManager.list_entries(db_session, dictionary.id)
        assert total == 0
    # endregion test_clear_entries

    # region test_add_entry_regex_valid_pattern [C:2] [TYPE Function]
    # @BRIEF Verify regex entry with valid pattern.
    def test_add_entry_regex_valid_pattern(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Regex Valid")

        entry = DictionaryManager.add_entry(
            db_session, dictionary.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, dictionary.id)
        assert total == 1
        assert entries[0].is_regex is True
    # endregion test_add_entry_regex_valid_pattern

    # region test_add_entry_regex_invalid_pattern_raises [C:2] [TYPE Function]
    # @BRIEF Verify invalid regex pattern raises ValueError.
    def test_add_entry_regex_invalid_pattern_raises(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Regex Invalid")

        with pytest.raises(ValueError, match="Invalid regex pattern"):
            DictionaryManager.add_entry(
                db_session, dictionary.id,
                r"[unclosed",
                "broken",
                source_language="en",
                target_language="ru",
                is_regex=True,
            )
    # endregion test_add_entry_regex_invalid_pattern_raises

    # region test_edit_entry_toggle_regex [C:2] [TYPE Function]
    # @BRIEF Verify toggling is_regex via edit_entry.
    def test_edit_entry_toggle_regex(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Regex Toggle")

        entry = DictionaryManager.add_entry(
            db_session, dictionary.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_edit_entry_toggle_regex

    # region test_edit_entry_regex_source_validates_pattern [C:2] [TYPE Function]
    # @BRIEF Verify editing source_term on regex entry validates pattern.
    def test_edit_entry_regex_source_validates_pattern(self, db_session: Session):
        dictionary = DictionaryManager.create_dictionary(db_session, name="Regex Edit Val")

        entry = DictionaryManager.add_entry(
            db_session, dictionary.id,
            r"\d+",
            "NUMBER",
            source_language="en",
            target_language="ru",
            is_regex=True,
        )

        # Try to set invalid regex
        with pytest.raises(ValueError, match="Invalid regex pattern"):
            DictionaryManager.edit_entry(db_session, entry.id, source_term=r"[unclosed")
    # endregion test_edit_entry_regex_source_validates_pattern

# #endregion TestDictionaryEntryCRUDIntegration

# #endregion DictionaryIntegrationTests