blank/tests/test_mcp_server.py

"""
Unit tests for the Novel Translator MCP Server.

This test suite validates all MCP tools and server functionality.
"""

import asyncio
import json
import tempfile
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, Mock, patch

import pytest

# Mock the heavy ML imports before importing server modules
sys_modules_mock = MagicMock()
sys_modules_mock.transformers = MagicMock()
sys_modules_mock.torch = MagicMock()

with patch.dict('sys.modules', {
    'transformers': sys_modules_mock.transformers,
    'torch': sys_modules_mock.torch
}):
    from src.mcp_server.server import (
        mcp,
        get_pipeline,
        get_glossary,
        get_cleaning_pipeline,
        get_fingerprint_service,
        create_task,
        update_progress,
        complete_task,
        notify_glossary_updated,
    )
    from src.glossary.models import Glossary, GlossaryEntry, TermCategory


# ============================================================================
# Fixtures
# ============================================================================

@pytest.fixture
def temp_file():
    """Create a temporary file for testing."""
    with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='.txt', encoding='utf-8') as f:
        f.write("## Chapter 1\n\nThis is test content.\n\n## Chapter 2\n\nMore content here.")
        temp_path = f.name
    yield temp_path
    Path(temp_path).unlink(missing_ok=True)


@pytest.fixture
def sample_glossary():
    """Create a sample glossary for testing."""
    glossary = Glossary()
    glossary.add(GlossaryEntry(
        source="林风",
        target="Lin Feng",
        category=TermCategory.CHARACTER,
        context="Main protagonist"
    ))
    glossary.add(GlossaryEntry(
        source="火球术",
        target="Fireball",
        category=TermCategory.SKILL,
        context="Magic spell"
    ))
    return glossary


@pytest.fixture
def mock_pipeline():
    """Create a mock translation pipeline."""
    pipeline = MagicMock()
    pipeline.translate = MagicMock(return_value=MagicMock(
        translated="Translated text",
        terms_used=["林风"]
    ))
    pipeline.translate_batch = MagicMock(return_value=[
        MagicMock(translated="Translation 1", terms_used=["term1"]),
        MagicMock(translated="Translation 2", terms_used=["term2"])
    ])
    pipeline.set_languages = MagicMock()
    pipeline.update_glossary = MagicMock()
    pipeline.src_lang = "zh"
    pipeline.tgt_lang = "en"
    return pipeline


@pytest.fixture
def mock_cleaning_pipeline():
    """Create a mock cleaning pipeline."""
    from src.cleaning.models import Chapter

    pipeline = MagicMock()
    pipeline.enable_cleaning = True
    pipeline.enable_splitting = True
    pipeline.process = MagicMock(return_value=[
        Chapter(index=0, title="Chapter 1", content="Content 1", char_count=100),
        Chapter(index=1, title="Chapter 2", content="Content 2", char_count=150),
    ])
    return pipeline


@pytest.fixture
def mock_fingerprint_service():
    """Create a mock fingerprint service."""
    service = MagicMock()
    service.check_before_import = MagicMock(return_value=(False, None))
    service.get_fingerprint = MagicMock(return_value="abc123def456")
    service.get_file_info = MagicMock(return_value={
        "fingerprint": "abc123",
        "metadata": {"size": 1000},
        "is_duplicate": False,
        "existing_work_id": None
    })
    return service


# ============================================================================
# Server State Tests
# ============================================================================

class TestServerState:
    """Tests for server state management."""

    def test_get_glossary_returns_singleton(self):
        """Test that get_glossary returns the same instance."""
        g1 = get_glossary()
        g2 = get_glossary()
        assert g1 is g2

    def test_create_task_generates_unique_id(self):
        """Test that create_task generates unique task IDs."""
        task1 = create_task("test_type", 10)
        task2 = create_task("test_type", 10)
        assert task1 != task2


# ============================================================================
# Translation Tool Tests
# ============================================================================

class TestTranslationTools:
    """Tests for translation tools."""

    @pytest.mark.asyncio
    async def test_translate_text_with_mock(self, mock_pipeline):
        """Test translate_text with mocked pipeline."""
        from src.mcp_server.server import translate_text

        with patch('src.mcp_server.server.get_pipeline', return_value=mock_pipeline):
            result = await translate_text(
                text="你好世界",
                src_lang="zh",
                tgt_lang="en"
            )

            assert result["success"] is True
            assert result["translated"] == "Translated text"
            assert result["terms_used"] == ["林风"]

    @pytest.mark.asyncio
    async def test_translate_text_empty_input(self):
        """Test translate_text with empty input."""
        from src.mcp_server.server import translate_text

        result = await translate_text(text="", src_lang="zh", tgt_lang="en")
        assert result["success"] is False
        assert "empty" in result["error"].lower()

    @pytest.mark.asyncio
    async def test_translate_batch_with_mock(self, mock_pipeline):
        """Test translate_batch with mocked pipeline."""
        from src.mcp_server.server import translate_batch

        with patch('src.mcp_server.server.get_pipeline', return_value=mock_pipeline):
            result = await translate_batch(
                texts=["Text 1", "Text 2"],
                src_lang="zh",
                tgt_lang="en"
            )

            assert result["success"] is True
            assert result["translations"] == ["Translation 1", "Translation 2"]
            assert len(result["terms_used"]) == 2

    @pytest.mark.asyncio
    async def test_translate_batch_empty_list(self):
        """Test translate_batch with empty list."""
        from src.mcp_server.server import translate_batch

        result = await translate_batch(texts=[], src_lang="zh", tgt_lang="en")
        assert result["success"] is False
        assert "empty" in result["error"].lower()

    @pytest.mark.asyncio
    async def test_translate_file_with_mock(
        self,
        temp_file,
        mock_pipeline,
        mock_cleaning_pipeline
    ):
        """Test translate_file with mocked dependencies."""
        from src.mcp_server.server import translate_file

        with patch('src.mcp_server.server.get_pipeline', return_value=mock_pipeline), \
             patch('src.mcp_server.server.get_cleaning_pipeline', return_value=mock_cleaning_pipeline), \
             patch('src.mcp_server.server.create_task', return_value="test-task-id"), \
             patch('src.mcp_server.server.update_progress', new_callable=AsyncMock), \
             patch('src.mcp_server.server.complete_task', new_callable=AsyncMock):

            # Create a temp output file
            with tempfile.NamedTemporaryFile(mode='w', delete=False, suffix='_en.txt') as f:
                output_path = f.name

            try:
                result = await translate_file(
                    file_path=temp_file,
                    output_path=output_path
                )

                assert result["success"] is True
                assert "task_id" in result
                assert result["chapters_translated"] == 2

            finally:
                Path(output_path).unlink(missing_ok=True)

    @pytest.mark.asyncio
    async def test_translate_file_not_found(self):
        """Test translate_file with non-existent file."""
        from src.mcp_server.server import translate_file

        result = await translate_file(file_path="/nonexistent/file.txt")
        assert result["success"] is False
        assert "not found" in result["error"].lower()


# ============================================================================
# Cleaning Tool Tests
# ============================================================================

class TestCleaningTools:
    """Tests for cleaning tools."""

    @pytest.mark.asyncio
    async def test_clean_file_with_mock(self, temp_file, mock_cleaning_pipeline):
        """Test clean_file with mocked pipeline."""
        from src.mcp_server.server import clean_file

        with patch('src.mcp_server.server.get_cleaning_pipeline', return_value=mock_cleaning_pipeline):
            result = await clean_file(file_path=temp_file)

            assert result["success"] is True
            assert result["chapter_count"] == 2
            assert result["total_chars"] == 250
            assert len(result["chapters"]) == 2

    @pytest.mark.asyncio
    async def test_clean_file_not_found(self):
        """Test clean_file with non-existent file."""
        from src.mcp_server.server import clean_file

        result = await clean_file(file_path="/nonexistent/file.txt")
        assert result["success"] is False
        assert "not found" in result["error"].lower()

    @pytest.mark.asyncio
    async def test_split_chapters_with_mock(self, mock_cleaning_pipeline):
        """Test split_chapters with mocked splitter."""
        from src.mcp_server.server import split_chapters
        from src.cleaning.splitter import ChapterSplitter

        mock_splitter = MagicMock()
        mock_splitter.split = MagicMock(return_value=[
            MagicMock(index=0, title="Chapter 1", char_count=100, content="Content 1"),
            MagicMock(index=1, title="Chapter 2", char_count=150, content="Content 2"),
        ])

        with patch('src.cleaning.splitter.ChapterSplitter', return_value=mock_splitter):
            result = await split_chapters(text="## Chapter 1\n\nContent\n\n## Chapter 2\n\nMore content")

            assert result["success"] is True
            assert result["chapter_count"] == 2

    @pytest.mark.asyncio
    async def test_split_chapters_empty_text(self):
        """Test split_chapters with empty text."""
        from src.mcp_server.server import split_chapters

        result = await split_chapters(text="")
        assert result["success"] is False
        assert "empty" in result["error"].lower()


# ============================================================================
# Glossary Tool Tests
# ============================================================================

class TestGlossaryTools:
    """Tests for glossary tools."""

    @pytest.mark.asyncio
    async def test_glossary_add(self):
        """Test adding a term to the glossary."""
        from src.mcp_server.server import glossary_add, get_glossary

        # Clear the glossary first
        get_glossary()._terms.clear()

        result = await glossary_add(
            source="林风",
            target="Lin Feng",
            category="character",
            context="Main protagonist"
        )

        assert result["success"] is True
        assert "entry" in result
        assert result["entry"]["source"] == "林风"
        assert result["entry"]["target"] == "Lin Feng"

    @pytest.mark.asyncio
    async def test_glossary_add_empty_source(self):
        """Test adding a term with empty source."""
        from src.mcp_server.server import glossary_add

        result = await glossary_add(source="", target="Lin Feng")
        assert result["success"] is False
        assert "empty" in result["error"].lower()

    @pytest.mark.asyncio
    async def test_glossary_add_empty_target(self):
        """Test adding a term with empty target."""
        from src.mcp_server.server import glossary_add

        result = await glossary_add(source="林风", target="")
        assert result["success"] is False
        assert "empty" in result["error"].lower()

    @pytest.mark.asyncio
    async def test_glossary_list(self, sample_glossary):
        """Test listing glossary entries."""
        from src.mcp_server.server import glossary_list, get_glossary

        # Replace with sample glossary
        with patch('src.mcp_server.server.get_glossary', return_value=sample_glossary):
            result = await glossary_list()

            assert result["success"] is True
            assert result["count"] == 2
            assert len(result["entries"]) == 2

            # Check entries
            sources = [e["source"] for e in result["entries"]]
            assert "林风" in sources
            assert "火球术" in sources

    @pytest.mark.asyncio
    async def test_glossary_clear(self):
        """Test clearing the glossary."""
        from src.mcp_server.server import glossary_clear, get_glossary

        # Add some entries first
        get_glossary().add(GlossaryEntry(
            source="Test",
            target="Test EN",
            category=TermCategory.OTHER
        ))

        result = await glossary_clear()

        assert result["success"] is True
        assert len(get_glossary()._terms) == 0


# ============================================================================
# Fingerprint Tool Tests
# ============================================================================

class TestFingerprintTools:
    """Tests for fingerprint tools."""

    @pytest.mark.asyncio
    async def test_check_duplicate(self, temp_file, mock_fingerprint_service):
        """Test checking for duplicate files."""
        from src.mcp_server.server import check_duplicate

        with patch('src.mcp_server.server.get_fingerprint_service', return_value=mock_fingerprint_service):
            result = await check_duplicate(file_path=temp_file)

            assert result["success"] is True
            assert result["is_duplicate"] is False
            assert result["fingerprint"] == "abc123def456"

    @pytest.mark.asyncio
    async def test_check_duplicate_not_found(self):
        """Test check_duplicate with non-existent file."""
        from src.mcp_server.server import check_duplicate

        result = await check_duplicate(file_path="/nonexistent/file.txt")
        assert result["success"] is False
        assert "not found" in result["error"].lower()

    @pytest.mark.asyncio
    async def test_get_fingerprint(self, temp_file, mock_fingerprint_service):
        """Test getting file fingerprint."""
        from src.mcp_server.server import get_fingerprint

        with patch('src.mcp_server.server.get_fingerprint_service', return_value=mock_fingerprint_service):
            result = await get_fingerprint(file_path=temp_file)

            assert result["success"] is True
            assert result["fingerprint"] == "abc123def456"
            assert "file_name" in result

    @pytest.mark.asyncio
    async def test_get_fingerprint_not_found(self):
        """Test get_fingerprint with non-existent file."""
        from src.mcp_server.server import get_fingerprint

        result = await get_fingerprint(file_path="/nonexistent/file.txt")
        assert result["success"] is False
        assert "not found" in result["error"].lower()


# ============================================================================
# Progress Resource Tests
# ============================================================================

class TestProgressResources:
    """Tests for progress resources."""

    @pytest.mark.asyncio
    async def test_progress_resource_flow(self):
        """Test the full progress resource flow."""
        from src.mcp_server.server import get_progress_resource, list_all_progress

        # Create a task
        task_id = create_task("test_type", 10)

        # Update progress
        await update_progress(task_id, {"current": 5, "percent": 50.0})

        # Get progress
        progress_json = await get_progress_resource(task_id)
        progress = json.loads(progress_json)

        assert progress["task_id"] == task_id
        assert progress["current"] == 5
        assert progress["percent"] == 50.0

        # List all progress
        list_json = await list_all_progress()
        task_list = json.loads(list_json)

        assert task_list["count"] >= 1

        # Complete task
        await complete_task(task_id, success=True)

        # Verify completion
        final_progress = json.loads(await get_progress_resource(task_id))
        assert final_progress["status"] == "completed"

    @pytest.mark.asyncio
    async def test_progress_resource_not_found(self):
        """Test getting progress for non-existent task."""
        from src.mcp_server.server import get_progress_resource

        progress_json = await get_progress_resource("non-existent-task-id")
        progress = json.loads(progress_json)

        assert "error" in progress
        assert "not found" in progress["error"].lower()


# ============================================================================
# Integration Tests
# ============================================================================

class TestIntegration:
    """Integration tests for complete workflows."""

    @pytest.mark.asyncio
    async def test_glossary_translation_workflow(self):
        """Test adding terms and using them in translation."""
        from src.mcp_server.server import glossary_add, glossary_list

        # Clear glossary
        get_glossary()._terms.clear()

        # Add terms
        await glossary_add(source="林风", target="Lin Feng", category="character")
        await glossary_add(source="青云宗", target="Qingyun Sect", category="organization")

        # List terms
        result = await glossary_list()

        assert result["success"] is True
        assert result["count"] == 2

        # Verify terms were added
        sources = {e["source"] for e in result["entries"]}
        assert "林风" in sources
        assert "青云宗" in sources


if __name__ == "__main__":
    pytest.main([__file__, "-v", "--cov=src/mcp_server", "--cov-report=term"])