From 3ad2576aed0867479747a7eab6fb0d344c73bb78 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Fri, 13 Mar 2026 23:10:52 +0800 Subject: [PATCH] test(04-01): add failing tests for BaseProcessor abstract class - Test BaseProcessor cannot be instantiated directly - Test subclass must implement process() method - Test state_manager attribute is provided - Test STEP_NAME class attribute requirement Co-Authored-By: Claude Opus 4.6 --- kb_pipeline/processors/base.py | 55 ++++++++++++ tests/test_processors/test_base.py | 130 +++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+) create mode 100644 kb_pipeline/processors/base.py create mode 100644 tests/test_processors/test_base.py diff --git a/kb_pipeline/processors/base.py b/kb_pipeline/processors/base.py new file mode 100644 index 0000000..8adc259 --- /dev/null +++ b/kb_pipeline/processors/base.py @@ -0,0 +1,55 @@ +"""BaseProcessor - Abstract base class for all pipeline processors + +Provides a unified interface that all processors (parse, download, asr, analyze) +must follow, enabling consistent state management and error handling. +""" + +from __future__ import annotations + +from abc import ABC, abstractmethod +from typing import Optional, TYPE_CHECKING + +from kb_pipeline.config import get_settings +from kb_pipeline.schemas import StepInput, StepOutput + +if TYPE_CHECKING: + from kb_pipeline.state import WorkflowStateManager + + +class BaseProcessor(ABC): + """Abstract base class for all pipeline processors. + + All processors (ParseProcessor, DownloadProcessor, ASRProcessor, AnalyzeProcessor) + must inherit from this class and implement the process() method. + + Attributes: + STEP_NAME: Class attribute identifying the processor step name. + settings: KBSettings instance from get_settings(). + state_manager: Optional WorkflowStateManager for state recording. + """ + + STEP_NAME: str + + def __init__(self, state_manager: Optional[WorkflowStateManager] = None): + """Initialize the processor. + + Args: + state_manager: Optional WorkflowStateManager for recording step state. + """ + self.settings = get_settings() + self.state_manager = state_manager + + @abstractmethod + async def process(self, input_data: StepInput) -> StepOutput: + """Process the input data and return the result. + + Args: + input_data: Input data for the processor step. + + Returns: + StepOutput with processing results. + + Raises: + Exception: If processing fails. + """ + pass diff --git a/tests/test_processors/test_base.py b/tests/test_processors/test_base.py new file mode 100644 index 0000000..17f28ec --- /dev/null +++ b/tests/test_processors/test_base.py @@ -0,0 +1,130 @@ +"""Tests for BaseProcessor abstract class + +Tests that BaseProcessor: +- Cannot be instantiated directly (abstract) +- Requires subclass to implement process() method +- Provides state_manager attribute +- Provides STEP_NAME class attribute +""" + +import pytest +from abc import ABC +from unittest.mock import MagicMock, patch + + +class TestBaseProcessorAbstract: + """BaseProcessor 抽象类测试""" + + def test_base_processor_cannot_be_instantiated(self): + """BaseProcessor 不能直接实例化""" + from kb_pipeline.processors.base import BaseProcessor + + with pytest.raises(TypeError): + BaseProcessor() + + def test_base_processor_is_abstract(self): + """BaseProcessor 是抽象类""" + from kb_pipeline.processors.base import BaseProcessor + + assert issubclass(BaseProcessor, ABC) + + def test_subclass_must_implement_process(self): + """子类必须实现 process() 方法""" + from kb_pipeline.processors.base import BaseProcessor + from kb_pipeline.schemas import StepInput, StepOutput + + # 缺少 process 实现的子类无法实例化 + class IncompleteProcessor(BaseProcessor): + STEP_NAME = "incomplete" + + with pytest.raises(TypeError): + IncompleteProcessor() + + def test_subclass_can_be_instantiated_with_process(self): + """实现了 process() 的子类可以实例化""" + from kb_pipeline.processors.base import BaseProcessor + from kb_pipeline.schemas import StepInput, StepOutput + + class CompleteProcessor(BaseProcessor): + STEP_NAME = "complete" + + async def process(self, input_data: StepInput) -> StepOutput: + pass + + # 可以实例化 + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + processor = CompleteProcessor() + assert processor is not None + + +class TestBaseProcessorAttributes: + """BaseProcessor 属性测试""" + + def test_base_processor_has_step_name_class_attribute(self): + """BaseProcessor 要求子类定义 STEP_NAME 类属性""" + from kb_pipeline.processors.base import BaseProcessor + from kb_pipeline.schemas import StepInput, StepOutput + + class TestProcessor(BaseProcessor): + STEP_NAME = "test_step" + + async def process(self, input_data: StepInput) -> StepOutput: + pass + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + processor = TestProcessor() + assert processor.STEP_NAME == "test_step" + + def test_base_processor_provides_state_manager(self): + """BaseProcessor 提供 state_manager 属性""" + from kb_pipeline.processors.base import BaseProcessor + from kb_pipeline.schemas import StepInput, StepOutput + + class TestProcessor(BaseProcessor): + STEP_NAME = "test" + + async def process(self, input_data: StepInput) -> StepOutput: + pass + + mock_state_manager = MagicMock() + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + processor = TestProcessor(state_manager=mock_state_manager) + assert processor.state_manager is mock_state_manager + + def test_base_processor_state_manager_optional(self): + """BaseProcessor state_manager 是可选的""" + from kb_pipeline.processors.base import BaseProcessor + from kb_pipeline.schemas import StepInput, StepOutput + + class TestProcessor(BaseProcessor): + STEP_NAME = "test" + + async def process(self, input_data: StepInput) -> StepOutput: + pass + + with patch("kb_pipeline.processors.base.get_settings") as mock_settings: + mock_settings.return_value = MagicMock() + processor = TestProcessor() + assert processor.state_manager is None + + def test_base_processor_provides_settings(self): + """BaseProcessor 提供 settings 属性""" + from kb_pipeline.processors.base import BaseProcessor + from kb_pipeline.schemas import StepInput, StepOutput + + class TestProcessor(BaseProcessor): + STEP_NAME = "test" + + async def process(self, input_data: StepInput) -> StepOutput: + pass + + mock_settings = MagicMock() + + with patch("kb_pipeline.processors.base.get_settings") as mock_get_settings: + mock_get_settings.return_value = mock_settings + processor = TestProcessor() + assert processor.settings is mock_settings