From 5d4cb93f8b5e6bae79f4f8a2812822bc6e127365 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:37:28 +0800 Subject: [PATCH] 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 --- tests/test_pipeline/test_pipeline.py | 305 +++++++++++++++++++++++++++ 1 file changed, 305 insertions(+) create mode 100644 tests/test_pipeline/test_pipeline.py diff --git a/tests/test_pipeline/test_pipeline.py b/tests/test_pipeline/test_pipeline.py new file mode 100644 index 0000000..a6c74d9 --- /dev/null +++ b/tests/test_pipeline/test_pipeline.py @@ -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"