kb-skills/tests/test_processors/test_parse.py

"""Tests for ParseProcessor

Tests that ParseProcessor:
- Parses Bilibili video URLs correctly
- Parses Zhihu answer URLs correctly
- Returns ParseOutput with success=False for unknown URLs
- Records state when state_manager is provided
- Can be imported from kb_pipeline.processors
"""

import pytest
from unittest.mock import MagicMock, patch, AsyncMock


class TestParseProcessorBilibili:
    """ParseProcessor Bilibili URL 测试"""

    @pytest.mark.asyncio
    async def test_parse_bilibili_video_url(self):
        """ParseProcessor 正确解析 Bilibili 视频 URL"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor()
            input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx411c7mD")

            result = await processor.process(input_data)

            assert result.success is True
            assert result.platform == "bilibili"
            assert result.content_type == "video"
            assert result.content_id == "BV1xx411c7mD"
            assert result.original_url == "https://www.bilibili.com/video/BV1xx411c7mD"

    @pytest.mark.asyncio
    async def test_parse_bilibili_article_url(self):
        """ParseProcessor 正确解析 Bilibili 专栏 URL"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor()
            input_data = ParseInput(url="https://www.bilibili.com/read/cv12345678")

            result = await processor.process(input_data)

            assert result.success is True
            assert result.platform == "bilibili"
            assert result.content_type == "article"
            assert result.content_id == "cv12345678"


class TestParseProcessorZhihu:
    """ParseProcessor Zhihu URL 测试"""

    @pytest.mark.asyncio
    async def test_parse_zhihu_answer_url(self):
        """ParseProcessor 正确解析 Zhihu 回答 URL"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor()
            input_data = ParseInput(url="https://www.zhihu.com/question/123456/answer/789012")

            result = await processor.process(input_data)

            assert result.success is True
            assert result.platform == "zhihu"
            assert result.content_type == "answer"
            assert result.content_id == "789012"

    @pytest.mark.asyncio
    async def test_parse_zhihu_zhuanlan_url(self):
        """ParseProcessor 正确解析 Zhihu 专栏 URL"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor()
            input_data = ParseInput(url="https://zhuanlan.zhihu.com/p/12345678")

            result = await processor.process(input_data)

            assert result.success is True
            assert result.platform == "zhihu"
            assert result.content_type == "zhuanlan"
            assert result.content_id == "12345678"


class TestParseProcessorUnknown:
    """ParseProcessor 未知 URL 测试"""

    @pytest.mark.asyncio
    async def test_parse_unknown_url_returns_failure(self):
        """ParseProcessor 对未知 URL 返回 success=False"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor()
            input_data = ParseInput(url="https://www.youtube.com/watch?v=abc123")

            result = await processor.process(input_data)

            assert result.success is False
            assert "Unsupported" in result.error_message or "unsupported" in result.error_message.lower()


class TestParseProcessorStateRecording:
    """ParseProcessor 状态记录测试"""

    @pytest.mark.asyncio
    async def test_state_recording_on_success(self):
        """ParseProcessor 成功时记录状态"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput, WorkflowState

        mock_state_manager = MagicMock()
        mock_state = MagicMock(spec=WorkflowState)
        mock_state_manager.load.return_value = mock_state

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor(state_manager=mock_state_manager)
            input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx")

            result = await processor.process(input_data)

            assert result.success is True
            mock_state.mark_step_completed.assert_called_once()
            call_args = mock_state.mark_step_completed.call_args
            assert call_args[0][0] == "parse"

    @pytest.mark.asyncio
    async def test_no_state_recording_without_manager(self):
        """无 state_manager 时不记录状态"""
        from kb_pipeline.processors.parse import ParseProcessor
        from kb_pipeline.schemas import ParseInput

        with patch("kb_pipeline.processors.base.get_settings") as mock_settings:
            mock_settings.return_value = MagicMock()

            processor = ParseProcessor(state_manager=None)
            input_data = ParseInput(url="https://www.bilibili.com/video/BV1xx")

            result = await processor.process(input_data)

            assert result.success is True
            # 不应该抛异常


class TestParseProcessorModuleExport:
    """ParseProcessor 模块导出测试"""

    def test_parse_processor_importable_from_processors(self):
        """ParseProcessor 可以从 kb_pipeline.processors 导入"""
        from kb_pipeline.processors import ParseProcessor

        assert ParseProcessor is not None

    def test_base_processor_importable_from_processors(self):
        """BaseProcessor 可以从 kb_pipeline.processors 导入"""
        from kb_pipeline.processors import BaseProcessor

        assert BaseProcessor is not None

    def test_asr_processor_still_importable(self):
        """ASRProcessor 仍然可以从 kb_pipeline.processors 导入"""
        from kb_pipeline.processors import ASRProcessor

        assert ASRProcessor is not None