kb-skills/tests/test_pipeline/test_idempotency.py

"""Tests for IdempotencyChecker module

Tests file existence checking for idempotent processing:
- File exists scenario
- File not exists scenario
- Partial files exist scenario
"""

import pytest
from pathlib import Path


class TestIdempotencyChecker:
    """Tests for IdempotencyChecker class"""

    def test_check_exists_returns_true_when_files_exist(self, tmp_path):
        """Test 1: 文件存在时 check_exists 返回 (True, [file_list])"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        # Create output directory with expected files
        output_dir = tmp_path / "bilibili"
        output_dir.mkdir()

        content_id = "BV1GJ411x7Qw"
        platform = "bilibili"

        # Create expected files
        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
        content_file = output_dir / f"{platform}_{content_id}_content.md"

        metadata_file.write_text('{"title": "test"}')
        content_file.write_text("# Test content")

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        assert exists is True
        assert len(files) == 2
        assert metadata_file in files
        assert content_file in files

    def test_check_exists_returns_false_when_files_not_exist(self, tmp_path):
        """Test 2: 文件不存在时 check_exists 返回 (False, [])"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "bilibili"
        output_dir.mkdir()

        content_id = "BV1GJ411x7Qw"
        platform = "bilibili"

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        assert exists is False
        assert len(files) == 0

    def test_check_exists_returns_partial_files_when_some_exist(self, tmp_path):
        """Test 3: 部分文件存在时返回正确的存在文件列表"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "zhihu"
        output_dir.mkdir()

        content_id = "12345678"
        platform = "zhihu"

        # Create only metadata file, not content file
        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
        metadata_file.write_text('{"title": "test"}')

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        # Should return False because not all files exist
        assert exists is False
        # But should list the existing file
        assert len(files) == 1
        assert metadata_file in files

    def test_check_exists_creates_directory_if_not_exists(self, tmp_path):
        """如果输出目录不存在，check_exists 应该返回 False"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "new_platform"  # Directory doesn't exist
        content_id = "BV1test123"
        platform = "bilibili"

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        assert exists is False
        assert len(files) == 0


class TestIdempotencyCheckerFileNaming:
    """Tests for file naming conventions"""

    def test_bilibili_file_naming(self, tmp_path):
        """测试B站文件的命名格式"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "bilibili"
        output_dir.mkdir()

        content_id = "BV1GJ411x7Qw"
        platform = "bilibili"

        # Create files with correct naming
        metadata_file = output_dir / "bilibili_BV1GJ411x7Qw_metadata.json"
        content_file = output_dir / "bilibili_BV1GJ411x7Qw_content.md"

        metadata_file.write_text('{}')
        content_file.write_text('# content')

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        assert exists is True
        assert len(files) == 2

    def test_zhihu_file_naming(self, tmp_path):
        """测试知乎文件的命名格式"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "zhihu"
        output_dir.mkdir()

        content_id = "87654321"
        platform = "zhihu"

        # Create files with correct naming
        metadata_file = output_dir / "zhihu_87654321_metadata.json"
        content_file = output_dir / "zhihu_87654321_content.md"

        metadata_file.write_text('{}')
        content_file.write_text('# content')

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        assert exists is True
        assert len(files) == 2


class TestIdempotencyCheckerEdgeCases:
    """Tests for edge cases"""

    def test_check_exists_with_empty_content_id(self, tmp_path):
        """空content_id应该返回False"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "bilibili"
        output_dir.mkdir()

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, "", "bilibili")

        assert exists is False
        assert len(files) == 0

    def test_check_exists_with_special_characters_in_id(self, tmp_path):
        """ID中包含特殊字符时的处理"""
        from kb_workflow.pipeline.idempotency import IdempotencyChecker

        output_dir = tmp_path / "bilibili"
        output_dir.mkdir()

        content_id = "BV1_test-123"
        platform = "bilibili"

        # Create files with special characters
        metadata_file = output_dir / f"{platform}_{content_id}_metadata.json"
        content_file = output_dir / f"{platform}_{content_id}_content.md"

        metadata_file.write_text('{}')
        content_file.write_text('# content')

        checker = IdempotencyChecker()
        exists, files = checker.check_exists(output_dir, content_id, platform)

        assert exists is True
        assert len(files) == 2