feat(04-01): add ParseInput and ParseOutput schemas

- ParseInput extends StepInput with url field - ParseOutput extends StepOutput with platform, content_type, content_id, resolved_url, original_url - Tests for schema validation and inheritance Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-13 23:11:44 +08:00
parent 3ad2576aed
commit efddef98d3
2 changed files with 137 additions and 0 deletions
@@ -472,3 +472,100 @@ class TestJSONSerialization:
        restored = WorkflowState.model_validate_json(json_str)
        assert "fetch_metadata" in restored.steps
        assert restored.steps["fetch_metadata"].status == StepStatus.COMPLETED
+
+
+# =============================================================================
+# Parse Schemas Tests (Phase 04-01)
+# =============================================================================
+
+
+class TestParseInput:
+    """测试 ParseInput Schema"""
+
+    def test_parse_input_validates_url_field(self):
+        """ParseInput 必须包含 url 字段，且 min_length=1"""
+        from kb_pipeline.schemas import ParseInput
+
+        # 缺少 url
+        with pytest.raises(ValidationError) as exc_info:
+            ParseInput()
+        assert "url" in str(exc_info.value)
+
+        # 空 url
+        with pytest.raises(ValidationError):
+            ParseInput(url="")
+
+        # 有效 url
+        input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx")
+        assert input_data.url == "https://www.bilibili.com/video/BV1xx"
+
+    def test_parse_input_content_id_default_empty(self):
+        """ParseInput content_id 默认为空字符串"""
+        from kb_pipeline.schemas import ParseInput
+
+        input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx")
+        assert input_data.content_id == ""
+
+    def test_parse_input_platform_default_empty(self):
+        """ParseInput platform 默认为空字符串"""
+        from kb_pipeline.schemas import ParseInput
+
+        input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx")
+        assert input_data.platform == ""
+
+
+class TestParseOutput:
+    """测试 ParseOutput Schema"""
+
+    def test_parse_output_inherits_step_output(self):
+        """ParseOutput 继承 StepOutput 的 success 和 error_message"""
+        from kb_pipeline.schemas import ParseOutput, StepOutput
+
+        output = ParseOutput(
+            success=True,
+            platform="bilibili",
+            content_type="video",
+            content_id="BV1xx",
+            resolved_url="https://www.bilibili.com/video/BV1xx",
+            original_url="https://b23.tv/abc"
+        )
+
+        assert isinstance(output, StepOutput)
+        assert output.success is True
+        assert output.error_message is None
+
+    def test_parse_output_has_required_fields(self):
+        """ParseOutput 包含所有必需字段"""
+        from kb_pipeline.schemas import ParseOutput
+
+        output = ParseOutput(
+            success=True,
+            platform="bilibili",
+            content_type="video",
+            content_id="BV1xx",
+            resolved_url="https://www.bilibili.com/video/BV1xx",
+            original_url="https://b23.tv/abc"
+        )
+
+        assert output.platform == "bilibili"
+        assert output.content_type == "video"
+        assert output.content_id == "BV1xx"
+        assert output.resolved_url == "https://www.bilibili.com/video/BV1xx"
+        assert output.original_url == "https://b23.tv/abc"
+
+    def test_parse_output_with_error(self):
+        """ParseOutput 可以表示失败状态"""
+        from kb_pipeline.schemas import ParseOutput
+
+        output = ParseOutput(
+            success=False,
+            error_message="Unsupported URL format",
+            platform="",
+            content_type="",
+            content_id="",
+            resolved_url="",
+            original_url="https://unknown.com/123"
+        )
+
+        assert output.success is False
+        assert output.error_message == "Unsupported URL format"