test(04-04): add schema validation tests for PipelineInput/Output/Config

- Test PipelineInput validates url field (required) - Test PipelineOutput contains results dict - Test PipelineConfig has default steps list - Test PipelineConfig allows custom output_dir Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-13 23:37:28 +08:00
parent 93c4f20044
commit 5d4cb93f8b
1 changed files with 305 additions and 0 deletions
@@ -0,0 +1,305 @@
+"""Pipeline Tests
+
+Tests for Pipeline class and PipelineInput/Output/Config schemas.
+"""
+
+import pytest
+from pathlib import Path
+from datetime import datetime
+from pydantic import ValidationError
+from unittest.mock import MagicMock, patch, AsyncMock
+
+
+# =============================================================================
+# Pipeline Schema Tests (Task 1)
+# =============================================================================
+
+
+class TestPipelineInput:
+    """Test PipelineInput Schema"""
+
+    def test_schema_pipeline_input_validates_url_field(self):
+        """PipelineInput must include url field (required)"""
+        from kb_pipeline.schemas import PipelineInput
+
+        # Missing url
+        with pytest.raises(ValidationError) as exc_info:
+            PipelineInput()
+        assert "url" in str(exc_info.value)
+
+        # Empty url
+        with pytest.raises(ValidationError):
+            PipelineInput(url="")
+
+        # Valid url
+        input_data = PipelineInput(url="https://www.bilibili.com/video/BV1xx")
+        assert input_data.url == "https://www.bilibili.com/video/BV1xx"
+
+    def test_schema_pipeline_input_has_optional_output_dir(self):
+        """PipelineInput has optional output_dir field"""
+        from kb_pipeline.schemas import PipelineInput
+
+        # Without output_dir
+        input_data = PipelineInput(url="https://www.bilibili.com/video/BV1xx")
+        assert input_data.output_dir is None
+
+        # With output_dir
+        input_with_dir = PipelineInput(
+            url="https://www.bilibili.com/video/BV1xx",
+            output_dir=Path("/tmp/content")
+        )
+        assert input_with_dir.output_dir == Path("/tmp/content")
+
+    def test_schema_pipeline_input_has_skip_completed_default(self):
+        """PipelineInput skip_completed defaults to True"""
+        from kb_pipeline.schemas import PipelineInput
+
+        input_data = PipelineInput(url="https://www.bilibili.com/video/BV1xx")
+        assert input_data.skip_completed is True
+
+        # Can be set to False
+        input_no_skip = PipelineInput(
+            url="https://www.bilibili.com/video/BV1xx",
+            skip_completed=False
+        )
+        assert input_no_skip.skip_completed is False
+
+
+class TestPipelineOutput:
+    """Test PipelineOutput Schema"""
+
+    def test_schema_pipeline_output_contains_results_dict(self):
+        """PipelineOutput contains results dict"""
+        from kb_pipeline.schemas import PipelineOutput
+
+        output = PipelineOutput(success=True)
+        assert hasattr(output, "results")
+        assert isinstance(output.results, dict)
+        assert output.results == {}
+
+    def test_schema_pipeline_output_has_success_field(self):
+        """PipelineOutput requires success field"""
+        from kb_pipeline.schemas import PipelineOutput
+
+        with pytest.raises(ValidationError) as exc_info:
+            PipelineOutput()
+        assert "success" in str(exc_info.value)
+
+    def test_schema_pipeline_output_has_optional_error_message(self):
+        """PipelineOutput has optional error_message field"""
+        from kb_pipeline.schemas import PipelineOutput
+
+        output = PipelineOutput(success=True)
+        assert output.error_message is None
+
+        output_with_error = PipelineOutput(
+            success=False,
+            error_message="Pipeline failed"
+        )
+        assert output_with_error.error_message == "Pipeline failed"
+
+    def test_schema_pipeline_output_has_content_dir(self):
+        """PipelineOutput has optional content_dir field"""
+        from kb_pipeline.schemas import PipelineOutput
+
+        output = PipelineOutput(success=True)
+        assert output.content_dir is None
+
+        output_with_dir = PipelineOutput(
+            success=True,
+            content_dir=Path("/tmp/content/bilibili_BV1xx")
+        )
+        assert output_with_dir.content_dir == Path("/tmp/content/bilibili_BV1xx")
+
+
+class TestPipelineConfig:
+    """Test PipelineConfig Schema"""
+
+    def test_schema_pipeline_config_has_default_steps_list(self):
+        """PipelineConfig has default steps list"""
+        from kb_pipeline.schemas import PipelineConfig
+
+        config = PipelineConfig()
+        assert hasattr(config, "steps")
+        assert isinstance(config.steps, list)
+        assert config.steps == ["parse", "download", "asr", "analyze"]
+
+    def test_schema_pipeline_config_allows_custom_steps(self):
+        """PipelineConfig allows custom steps list"""
+        from kb_pipeline.schemas import PipelineConfig
+
+        custom_config = PipelineConfig(steps=["parse", "download"])
+        assert custom_config.steps == ["parse", "download"]
+
+    def test_schema_pipeline_config_allows_custom_output_dir(self):
+        """PipelineConfig allows custom output_dir"""
+        from kb_pipeline.schemas import PipelineConfig
+
+        config = PipelineConfig(output_dir=Path("/custom/output"))
+        assert config.output_dir == Path("/custom/output")
+
+        # Default is None
+        default_config = PipelineConfig()
+        assert default_config.output_dir is None
+
+    def test_schema_pipeline_config_has_skip_completed_default(self):
+        """PipelineConfig skip_completed defaults to True"""
+        from kb_pipeline.schemas import PipelineConfig
+
+        config = PipelineConfig()
+        assert config.skip_completed is True
+
+    def test_schema_pipeline_config_has_max_retries_default(self):
+        """PipelineConfig max_retries defaults to 3"""
+        from kb_pipeline.schemas import PipelineConfig
+
+        config = PipelineConfig()
+        assert config.max_retries == 3
+
+        # Can be customized
+        custom_config = PipelineConfig(max_retries=5)
+        assert custom_config.max_retries == 5
+
+
+# =============================================================================
+# Pipeline Class Tests (Task 2)
+# =============================================================================
+
+
+class TestPipelineInit:
+    """Test Pipeline initialization"""
+
+    def test_init_pipeline_with_processors(self):
+        """Pipeline initializes with all processors"""
+        from kb_pipeline.pipeline import Pipeline
+        from kb_pipeline.schemas import PipelineConfig
+
+        mock_processor1 = MagicMock()
+        mock_processor1.STEP_NAME = "parse"
+
+        mock_processor2 = MagicMock()
+        mock_processor2.STEP_NAME = "download"
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            pipeline = Pipeline(processors=[mock_processor1, mock_processor2])
+
+            assert "parse" in pipeline.processors
+            assert "download" in pipeline.processors
+            assert pipeline.processors["parse"] == mock_processor1
+            assert pipeline.processors["download"] == mock_processor2
+
+    def test_init_pipeline_validates_processor_step_name(self):
+        """Pipeline validates processor STEP_NAME matches config.steps"""
+        from kb_pipeline.pipeline import Pipeline
+        from kb_pipeline.schemas import PipelineConfig
+
+        mock_processor = MagicMock()
+        mock_processor.STEP_NAME = "unknown_step"
+
+        config = PipelineConfig(steps=["parse", "download"])
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            with pytest.raises(ValueError) as exc_info:
+                Pipeline(processors=[mock_processor], config=config)
+
+            assert "Unknown processor" in str(exc_info.value)
+
+    def test_init_pipeline_handles_missing_processor_gracefully(self):
+        """Pipeline handles missing processor gracefully"""
+        from kb_pipeline.pipeline import Pipeline
+        from kb_pipeline.schemas import PipelineConfig
+
+        # Empty processors list should work
+        config = PipelineConfig(steps=[])
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            pipeline = Pipeline(processors=[], config=config)
+            assert pipeline.processors == {}
+
+
+class TestPipelineParseUrl:
+    """Test Pipeline parse_url method"""
+
+    def test_parse_url_creates_correct_content_directory(self):
+        """Pipeline._get_content_dir creates correct content directory"""
+        from kb_pipeline.pipeline import Pipeline
+        from kb_pipeline.link_parser import ParsedLink
+        import tempfile
+
+        mock_processor = MagicMock()
+        mock_processor.STEP_NAME = "parse"
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            with tempfile.TemporaryDirectory() as tmpdir:
+                pipeline = Pipeline(
+                    processors=[mock_processor],
+                    output_dir=Path(tmpdir)
+                )
+
+                parsed = ParsedLink(
+                    platform="bilibili",
+                    content_type="video",
+                    content_id="BV1xx411c7mD",
+                    original_url="https://www.bilibili.com/video/BV1xx411c7mD"
+                )
+
+                content_dir = pipeline._get_content_dir(parsed)
+
+                assert content_dir.exists()
+                assert content_dir.name == "bilibili_BV1xx411c7mD"
+
+    def test_parse_url_returns_none_for_empty_url(self):
+        """Pipeline._parse_url returns None for empty URL"""
+        from kb_pipeline.pipeline import Pipeline
+
+        mock_processor = MagicMock()
+        mock_processor.STEP_NAME = "parse"
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            pipeline = Pipeline(processors=[mock_processor])
+            result = pipeline._parse_url("")
+
+            assert result is None
+
+    def test_parse_url_returns_none_for_invalid_url(self):
+        """Pipeline._parse_url returns None for unrecognized URL"""
+        from kb_pipeline.pipeline import Pipeline
+
+        mock_processor = MagicMock()
+        mock_processor.STEP_NAME = "parse"
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            pipeline = Pipeline(processors=[mock_processor])
+            result = pipeline._parse_url("https://www.unknown.com/123")
+
+            assert result is None
+
+    def test_parse_url_parses_bilibili_url(self):
+        """Pipeline._parse_url correctly parses Bilibili URL"""
+        from kb_pipeline.pipeline import Pipeline
+
+        mock_processor = MagicMock()
+        mock_processor.STEP_NAME = "parse"
+
+        with patch("kb_pipeline.pipeline.get_settings") as mock_settings:
+            mock_settings.return_value = MagicMock()
+
+            pipeline = Pipeline(processors=[mock_processor])
+            result = pipeline._parse_url("https://www.bilibili.com/video/BV1xx411c7mD")
+
+            assert result is not None
+            assert result.platform == "bilibili"
+            assert result.content_type == "video"
+            assert result.content_id == "BV1xx411c7mD"