kb-skills/tests/test_processors/test_base.py

"""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_workflow.pipeline.processors.base import BaseProcessor

        with pytest.raises(TypeError):
            BaseProcessor()

    def test_base_processor_is_abstract(self):
        """BaseProcessor 是抽象类"""
        from kb_workflow.pipeline.processors.base import BaseProcessor

        assert issubclass(BaseProcessor, ABC)

    def test_subclass_must_implement_process(self):
        """子类必须实现 process() 方法"""
        from kb_workflow.pipeline.processors.base import BaseProcessor
        from kb_workflow.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_workflow.pipeline.processors.base import BaseProcessor
        from kb_workflow.pipeline.schemas import StepInput, StepOutput

        class CompleteProcessor(BaseProcessor):
            STEP_NAME = "complete"

            async def process(self, input_data: StepInput) -> StepOutput:
                pass

        # 可以实例化
        with patch("kb_workflow.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_workflow.pipeline.processors.base import BaseProcessor
        from kb_workflow.pipeline.schemas import StepInput, StepOutput

        class TestProcessor(BaseProcessor):
            STEP_NAME = "test_step"

            async def process(self, input_data: StepInput) -> StepOutput:
                pass

        with patch("kb_workflow.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_workflow.pipeline.processors.base import BaseProcessor
        from kb_workflow.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_workflow.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_workflow.pipeline.processors.base import BaseProcessor
        from kb_workflow.pipeline.schemas import StepInput, StepOutput

        class TestProcessor(BaseProcessor):
            STEP_NAME = "test"

            async def process(self, input_data: StepInput) -> StepOutput:
                pass

        with patch("kb_workflow.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_workflow.pipeline.processors.base import BaseProcessor
        from kb_workflow.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_workflow.pipeline.processors.base.get_settings") as mock_get_settings:
            mock_get_settings.return_value = mock_settings
            processor = TestProcessor()
            assert processor.settings is mock_settings