blank/tests/cleaning/test_pipeline.py

"""
Integration tests for CleaningPipeline.
"""

import pytest
from pathlib import Path
from src.cleaning.pipeline import CleaningPipeline, CleaningPipelineError
from src.cleaning.reader import TxtReader
from src.cleaning.cleaner import TextCleaner
from src.cleaning.splitter import ChapterSplitter


@pytest.fixture
def sample_file(tmp_path):
    """Create a sample TXT file for testing."""
    file_path = tmp_path / "sample.txt"
    content = """第一章 开始

这是第一章的内容，包含一些文字。

林风站在山顶，看着远方。

第二章 继续

这是第二章的内容。

他开始了新的旅程。
"""
    file_path.write_text(content, encoding="utf-8")
    return file_path


@pytest.fixture
def pipeline():
    """Create a CleaningPipeline instance."""
    return CleaningPipeline()


class TestCleaningPipeline:
    """Test suite for CleaningPipeline."""

    def test_process_basic(self, pipeline, sample_file):
        """Test basic file processing."""
        chapters = pipeline.process(sample_file)
        assert len(chapters) >= 2
        assert "第一章" in chapters[0].title

    def test_process_with_info(self, pipeline, sample_file):
        """Test processing with info return."""
        chapters, info = pipeline.process(sample_file, return_info=True)
        assert len(chapters) >= 2
        assert 'encoding' in info
        assert 'original_size' in info
        assert 'cleaned_size' in info

    def test_process_to_result(self, pipeline, sample_file):
        """Test processing to CleaningResult."""
        result = pipeline.process_to_result(sample_file)
        assert result.chapter_count >= 2
        assert result.original_char_count > 0
        assert result.cleaned_char_count > 0

    def test_removal_rate_property(self, pipeline, sample_file):
        """Test removal rate calculation."""
        result = pipeline.process_to_result(sample_file)
        rate = result.removal_rate
        assert 0.0 <= rate <= 1.0

    def test_read_and_clean(self, pipeline, sample_file):
        """Test reading and cleaning without splitting."""
        content = pipeline.read_and_clean(sample_file)
        assert isinstance(content, str)
        assert len(content) > 0

    def test_get_file_info(self, pipeline, sample_file):
        """Test getting file information."""
        info = pipeline.get_file_info(sample_file)
        assert info['exists'] is True
        assert info['is_file'] is True
        assert 'size' in info
        assert 'encoding' in info

    def test_get_file_info_nonexistent(self, pipeline, tmp_path):
        """Test getting info for non-existent file."""
        info = pipeline.get_file_info(tmp_path / "nonexistent.txt")
        assert info['exists'] is False
        assert info['is_file'] is False

    def test_custom_components(self, sample_file):
        """Test pipeline with custom components."""
        custom_reader = TxtReader(default_encoding="utf-8")
        custom_cleaner = TextCleaner(remove_extra_whitespace=True)
        custom_splitter = ChapterSplitter(min_chapter_length=10)

        pipeline = CleaningPipeline(
            reader=custom_reader,
            cleaner=custom_cleaner,
            splitter=custom_splitter
        )

        chapters = pipeline.process(sample_file)
        assert len(chapters) >= 2

    def test_disable_cleaning(self, sample_file):
        """Test pipeline with cleaning disabled."""
        pipeline = CleaningPipeline(enable_cleaning=False)
        chapters, info = pipeline.process(sample_file, return_info=True)
        assert len(chapters) >= 2
        assert info.get('removed_chars', 0) == 0

    def test_disable_splitting(self, sample_file):
        """Test pipeline with splitting disabled."""
        pipeline = CleaningPipeline(enable_splitting=False)
        chapters = pipeline.process(sample_file)
        assert len(chapters) == 1
        assert chapters[0].title == "全文"

    def test_create_custom_splitter(self, pipeline):
        """Test creating custom splitter."""
        pipeline.create_custom_splitter(
            min_chapter_length=50,
            merge_short_chapters=False
        )
        assert pipeline.splitter.min_chapter_length == 50

    def test_create_custom_cleaner(self, pipeline):
        """Test creating custom cleaner."""
        pipeline.create_custom_cleaner(
            remove_extra_whitespace=True,
            fix_punctuation=True
        )
        assert pipeline.cleaner.remove_extra_whitespace is True

    def test_batch_process(self, pipeline, tmp_path):
        """Test batch processing multiple files."""
        # Create multiple files
        files = []
        for i in range(3):
            file_path = tmp_path / f"file_{i}.txt"
            content = f"第{i+1}章\n内容{i}\n"
            file_path.write_text(content, encoding="utf-8")
            files.append(file_path)

        results = pipeline.batch_process(files)
        assert len(results) == 3

        for path, chapters in results:
            assert isinstance(chapters, list)

    def test_batch_process_with_errors(self, pipeline, tmp_path):
        """Test batch processing with some errors."""
        files = [
            tmp_path / "exists.txt",
            tmp_path / "nonexistent.txt"
        ]
        files[0].write_text("内容", encoding="utf-8")

        results = pipeline.batch_process(files, raise_on_error=False)
        assert len(results) == 2
        assert isinstance(results[0][1], list)  # Success
        assert isinstance(results[1][1], Exception)  # Error

    def test_batch_process_raise_on_error(self, pipeline, tmp_path):
        """Test batch processing raises on error."""
        files = [tmp_path / "nonexistent.txt"]

        with pytest.raises(CleaningPipelineError):
            pipeline.batch_process(files, raise_on_error=True)

    def test_process_nonexistent_file(self, pipeline):
        """Test processing non-existent file raises error."""
        with pytest.raises(CleaningPipelineError):
            pipeline.process("/nonexistent/file.txt")

    def test_process_empty_file(self, pipeline, tmp_path):
        """Test processing empty file."""
        empty_file = tmp_path / "empty.txt"
        empty_file.write_text("", encoding="utf-8")

        chapters = pipeline.process(empty_file)
        # Should handle gracefully - either empty list or single empty chapter
        assert isinstance(chapters, list)

    def test_result_properties(self, pipeline, sample_file):
        """Test CleaningResult properties."""
        result = pipeline.process_to_result(sample_file)

        # Test chapter_count property
        assert result.chapter_count == len(result.chapters)

        # Test chapters have content
        for chapter in result.chapters:
            assert hasattr(chapter, 'content')
            assert hasattr(chapter, 'char_count')

    def test_chapter_word_count_property(self, pipeline, sample_file):
        """Test chapter word_count property."""
        chapters = pipeline.process(sample_file)
        for chapter in chapters:
            assert chapter.word_count >= 0

    def test_chapter_len_operator(self, pipeline, sample_file):
        """Test len() operator on chapters."""
        chapters = pipeline.process(sample_file)
        for chapter in chapters:
            assert len(chapter) == chapter.char_count

    def test_full_pipeline_integration(self, pipeline, sample_file):
        """Test full integration of all components."""
        # This test verifies the entire pipeline works together
        result = pipeline.process_to_result(sample_file)

        # Verify all stages completed
        assert result.chapter_count > 0
        assert result.original_char_count > 0
        assert result.cleaned_char_count >= 0

        # Verify chapter structure
        for chapter in result.chapters:
            assert hasattr(chapter, 'index')
            assert hasattr(chapter, 'title')
            assert hasattr(chapter, 'content')
            assert chapter.index >= 0

    def test_chinese_encoding_detection(self, pipeline, tmp_path):
        """Test processing files with different Chinese encodings."""
        # GBK encoded file
        gbk_file = tmp_path / "gbk.txt"
        content = "第一章 测试\n内容"
        gbk_file.write_bytes(content.encode("gbk"))

        chapters = pipeline.process(gbk_file)
        assert len(chapters) >= 1

    def test_large_file_handling(self, pipeline, tmp_path):
        """Test handling larger files."""
        large_file = tmp_path / "large.txt"
        # Create a file with many chapters
        lines = []
        for i in range(50):
            lines.append(f"第{i+1}章")
            lines.append("这是测试内容。" * 10)

        large_file.write_text("\n".join(lines), encoding="utf-8")

        chapters = pipeline.process(large_file)
        assert len(chapters) == 50

    def test_no_chapters_detected(self, pipeline, tmp_path):
        """Test file without chapter titles."""
        no_chapter_file = tmp_path / "no_chapter.txt"
        no_chapter_file.write_text("这是一段没有章节标题的文本。\n第二行内容。", encoding="utf-8")

        chapters = pipeline.process(no_chapter_file)
        # Should return single chapter with "全文" title
        assert len(chapters) == 1
        assert chapters[0].title == "全文"

    def test_special_characters_in_file(self, pipeline, tmp_path):
        """Test handling files with special characters."""
        special_file = tmp_path / "special.txt"
        content = "第一章：测试！\n\"引号\"内容\n\t制表符\n多种标点：；，。！？"
        special_file.write_text(content, encoding="utf-8")

        chapters = pipeline.process(special_file)
        assert len(chapters) >= 1

    def test_cleaning_statistics(self, pipeline, sample_file):
        """Test that cleaning statistics are accurate."""
        result = pipeline.process_to_result(sample_file)

        # Verify statistics are consistent
        if result.original_char_count > result.cleaned_char_count:
            assert result.removed_char_count > 0
            assert result.removed_char_count == result.original_char_count - result.cleaned_char_count

    def test_pipeline_with_custom_patterns(self, tmp_path):
        """Test pipeline with custom chapter patterns."""
        custom_file = tmp_path / "custom.txt"
        # Make content longer to avoid merging
        content = """EPISODE 1 Start

This is episode one with enough content to avoid merging.

EPISODE 2 Middle

This is episode two with enough content to avoid merging as well.
"""
        custom_file.write_text(content, encoding="utf-8")

        pipeline = CleaningPipeline()
        pipeline.create_custom_splitter(
            min_chapter_length=10,
            merge_short_chapters=False,
            custom_patterns=[(r'^EPISODE\s+\d+', 1)]
        )

        chapters = pipeline.process(custom_file)
        assert len(chapters) >= 2

    def test_is_binary_detection(self, pipeline, tmp_path):
        """Test binary file detection."""
        text_file = tmp_path / "text.txt"
        text_file.write_text("文本内容", encoding="utf-8")

        binary_file = tmp_path / "binary.bin"
        binary_file.write_bytes(b"\x00\x01\x02\x03" * 100)

        text_info = pipeline.get_file_info(text_file)
        binary_info = pipeline.get_file_info(binary_file)

        assert text_info['is_binary'] is False
        assert binary_info['is_binary'] is True