From 98709476ab96d2be43c23e6772692e37ffd32266 Mon Sep 17 00:00:00 2001 From: starmen1265 Date: Thu, 19 Mar 2026 23:45:49 +0800 Subject: [PATCH] test(12-02): add Bilibili video pipeline end-to-end tests - Test Bilibili video uses SRTProcessor (not AnalyzeProcessor) - Test Bilibili article uses AnalyzeProcessor (not SRTProcessor) - Test Zhihu answer uses AnalyzeProcessor - Test Markdown output format verification All 4 tests pass. Co-Authored-By: Claude Opus 4.6 --- tests/test_bilibili_pipeline.py | 422 ++++++++++++++++++++++++++++++++ 1 file changed, 422 insertions(+) create mode 100644 tests/test_bilibili_pipeline.py diff --git a/tests/test_bilibili_pipeline.py b/tests/test_bilibili_pipeline.py new file mode 100644 index 0000000..4091170 --- /dev/null +++ b/tests/test_bilibili_pipeline.py @@ -0,0 +1,422 @@ +"""Tests for Bilibili video pipeline integration + +Tests for: +- Bilibili video processing flow (uses SRTProcessor) +- Bilibili article processing flow (uses AnalyzeProcessor) +- Zhihu processing flow (uses AnalyzeProcessor) +- Platform routing logic +""" + +import asyncio +import json +from pathlib import Path +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from kb_workflow.pipeline.pipeline import Pipeline +from kb_workflow.pipeline.schemas import PipelineConfig +from kb_workflow.pipeline.processors import ( + ParseProcessor, DownloadProcessor, ASRProcessor, + AnalyzeProcessor, ArchiveProcessor +) +from kb_workflow.pipeline.processors.srt import SRTProcessor + + +# ============================================================================= +# Test Fixtures +# ============================================================================= + +@pytest.fixture +def mock_bilibili_video_url(): + """B站视频 URL""" + return "https://www.bilibili.com/video/BV1test123456" + + +@pytest.fixture +def mock_bilibili_article_url(): + """B站文章 URL""" + return "https://www.bilibili.com/read/cv123456" + + +@pytest.fixture +def mock_zhihu_answer_url(): + """知乎回答 URL""" + return "https://www.zhihu.com/question/123456/answer/789012" + + +@pytest.fixture +def sample_metadata(): + """示例元数据""" + return { + "title": "测试视频标题", + "author": "测试作者", + "duration": 300, + "pubtime": "2024-01-01", + "bvid": "BV1test123456", + "platform": "bilibili", + "url": "https://www.bilibili.com/video/BV1test123456" + } + + +@pytest.fixture +def sample_srt_content(): + """示例 SRT 内容""" + return """1 +00:00:00,000 --> 00:00:05,000 +第一句字幕内容 + +2 +00:00:05,000 --> 00:00:10,000 +第二句字幕内容 + +3 +00:00:10,000 --> 00:00:15,000 +第三句字幕内容 +""" + + +# ============================================================================= +# Test Cases +# ============================================================================= + +class TestBilibiliVideoPipeline: + """B站视频处理流程测试""" + + @pytest.mark.asyncio + async def test_bilibili_video_uses_srt_processor(self, tmp_path, mock_bilibili_video_url, sample_metadata, sample_srt_content): + """验证 B站视频使用 SRTProcessor 而非 AnalyzeProcessor""" + # Setup: 创建临时目录和文件 + content_dir = tmp_path / "bilibili_BV1test123456" + content_dir.mkdir(parents=True) + + # 创建元数据文件 + metadata_file = content_dir / "metadata.json" + metadata_file.write_text(json.dumps(sample_metadata, ensure_ascii=False), encoding='utf-8') + + # 创建 SRT 文件 + srt_file = content_dir / "bilibili_BV1test123456_subtitle.srt" + srt_file.write_text(sample_srt_content, encoding='utf-8') + + # Mock processors + with patch.object(ParseProcessor, 'process', new_callable=AsyncMock) as mock_parse, \ + patch.object(DownloadProcessor, 'process', new_callable=AsyncMock) as mock_download, \ + patch.object(ASRProcessor, 'process', new_callable=AsyncMock) as mock_asr, \ + patch.object(SRTProcessor, 'process', new_callable=AsyncMock) as mock_srt, \ + patch.object(AnalyzeProcessor, 'process', new_callable=AsyncMock) as mock_analyze, \ + patch.object(ArchiveProcessor, 'process', new_callable=AsyncMock) as mock_archive: + + # 配置 mock 返回值 + from kb_workflow.pipeline.schemas import ParseOutput, DownloadOutput, ASROutput, SRTOutput, ArchiveOutput + + mock_parse.return_value = ParseOutput( + success=True, + platform="bilibili", + content_type="video", + content_id="BV1test123456", + resolved_url=mock_bilibili_video_url, + original_url=mock_bilibili_video_url + ) + + mock_download.return_value = DownloadOutput( + success=True, + metadata_file=str(metadata_file), + content_file=str(content_dir / "video.mp4"), + title=sample_metadata["title"], + author=sample_metadata["author"], + duration=sample_metadata["duration"] + ) + + mock_asr.return_value = ASROutput( + success=True, + tool="subtitle", + has_subtitle=True, + output_files={"srt": str(srt_file), "json": str(content_dir / "subtitle.json")}, + text="第一句字幕内容 第二句字幕内容 第三句字幕内容", + duration_seconds=15.0 + ) + + mock_srt.return_value = SRTOutput( + success=True, + output_file=str(content_dir / "analysis.md"), + paragraphs=["第一句字幕内容", "第二句字幕内容", "第三句字幕内容"], + tags=[], + corrected=False + ) + + mock_archive.return_value = ArchiveOutput( + success=True, + archive_file=str(tmp_path / "archive" / "test.md"), + skipped=False, + platform="bilibili" + ) + + # 创建 Pipeline + processors = [ + ParseProcessor(), + DownloadProcessor(), + ASRProcessor(), + SRTProcessor(), + AnalyzeProcessor(), + ArchiveProcessor(), + ] + + config = PipelineConfig( + steps=["parse", "download", "asr", "srt", "analyze", "archive"], + output_dir=tmp_path, + skip_completed=False, + ) + + pipeline = Pipeline( + processors=processors, + config=config, + output_dir=tmp_path, + ) + + # 执行 + result = await pipeline.run(mock_bilibili_video_url) + + # 验证 + assert result.success, f"Pipeline failed: {result.error_message}" + + # SRTProcessor 应该被调用 + mock_srt.assert_called_once() + + # AnalyzeProcessor 不应该被调用（B站视频跳过） + mock_analyze.assert_not_called() + + @pytest.mark.asyncio + async def test_bilibili_article_uses_analyze_processor(self, tmp_path, mock_bilibili_article_url, sample_metadata): + """验证 B站文章使用 AnalyzeProcessor""" + # Setup + content_dir = tmp_path / "bilibili_cv123456" + content_dir.mkdir(parents=True) + + metadata_file = content_dir / "metadata.json" + sample_metadata["url"] = mock_bilibili_article_url + metadata_file.write_text(json.dumps(sample_metadata, ensure_ascii=False), encoding='utf-8') + + content_file = content_dir / "article.txt" + content_file.write_text("文章内容测试", encoding='utf-8') + + with patch.object(ParseProcessor, 'process', new_callable=AsyncMock) as mock_parse, \ + patch.object(DownloadProcessor, 'process', new_callable=AsyncMock) as mock_download, \ + patch.object(ASRProcessor, 'process', new_callable=AsyncMock) as mock_asr, \ + patch.object(SRTProcessor, 'process', new_callable=AsyncMock) as mock_srt, \ + patch.object(AnalyzeProcessor, 'process', new_callable=AsyncMock) as mock_analyze, \ + patch.object(ArchiveProcessor, 'process', new_callable=AsyncMock) as mock_archive: + + from kb_workflow.pipeline.schemas import ParseOutput, DownloadOutput, AnalyzeOutput, ArchiveOutput + + mock_parse.return_value = ParseOutput( + success=True, + platform="bilibili", + content_type="article", + content_id="cv123456", + resolved_url=mock_bilibili_article_url, + original_url=mock_bilibili_article_url + ) + + mock_download.return_value = DownloadOutput( + success=True, + metadata_file=str(metadata_file), + content_file=str(content_file), + title=sample_metadata["title"], + author=sample_metadata["author"] + ) + + mock_analyze.return_value = AnalyzeOutput( + success=True, + analysis_file=str(content_dir / "analysis.md"), + summary="文章摘要", + main_topics=["主题1"], + tags=["标签1"] + ) + + mock_archive.return_value = ArchiveOutput( + success=True, + archive_file=str(tmp_path / "archive" / "test.md"), + skipped=False, + platform="bilibili" + ) + + processors = [ + ParseProcessor(), + DownloadProcessor(), + ASRProcessor(), + SRTProcessor(), + AnalyzeProcessor(), + ArchiveProcessor(), + ] + + config = PipelineConfig( + steps=["parse", "download", "asr", "srt", "analyze", "archive"], + output_dir=tmp_path, + skip_completed=False, + ) + + pipeline = Pipeline( + processors=processors, + config=config, + output_dir=tmp_path, + ) + + result = await pipeline.run(mock_bilibili_article_url) + + assert result.success, f"Pipeline failed: {result.error_message}" + + # ASR 不应该被调用（文章类型跳过） + mock_asr.assert_not_called() + + # SRTProcessor 不应该被调用（非视频跳过） + mock_srt.assert_not_called() + + # AnalyzeProcessor 应该被调用 + mock_analyze.assert_called_once() + + +class TestZhihuPipeline: + """知乎处理流程测试""" + + @pytest.mark.asyncio + async def test_zhihu_answer_uses_analyze_processor(self, tmp_path, mock_zhihu_answer_url): + """验证知乎回答使用 AnalyzeProcessor""" + content_dir = tmp_path / "zhihu_789012" + content_dir.mkdir(parents=True) + + metadata = { + "title": "知乎回答测试", + "author": "知乎用户", + "platform": "zhihu", + "url": mock_zhihu_answer_url + } + metadata_file = content_dir / "metadata.json" + metadata_file.write_text(json.dumps(metadata, ensure_ascii=False), encoding='utf-8') + + content_file = content_dir / "content.txt" + content_file.write_text("知乎回答内容测试", encoding='utf-8') + + with patch.object(ParseProcessor, 'process', new_callable=AsyncMock) as mock_parse, \ + patch.object(DownloadProcessor, 'process', new_callable=AsyncMock) as mock_download, \ + patch.object(ASRProcessor, 'process', new_callable=AsyncMock) as mock_asr, \ + patch.object(SRTProcessor, 'process', new_callable=AsyncMock) as mock_srt, \ + patch.object(AnalyzeProcessor, 'process', new_callable=AsyncMock) as mock_analyze, \ + patch.object(ArchiveProcessor, 'process', new_callable=AsyncMock) as mock_archive: + + from kb_workflow.pipeline.schemas import ParseOutput, DownloadOutput, AnalyzeOutput, ArchiveOutput + + mock_parse.return_value = ParseOutput( + success=True, + platform="zhihu", + content_type="answer", + content_id="789012", + resolved_url=mock_zhihu_answer_url, + original_url=mock_zhihu_answer_url + ) + + mock_download.return_value = DownloadOutput( + success=True, + metadata_file=str(metadata_file), + content_file=str(content_file), + title=metadata["title"], + author=metadata["author"] + ) + + mock_analyze.return_value = AnalyzeOutput( + success=True, + analysis_file=str(content_dir / "analysis.md"), + summary="回答摘要", + main_topics=["主题1"], + tags=["标签1"] + ) + + mock_archive.return_value = ArchiveOutput( + success=True, + archive_file=str(tmp_path / "archive" / "test.md"), + skipped=False, + platform="zhihu" + ) + + processors = [ + ParseProcessor(), + DownloadProcessor(), + ASRProcessor(), + SRTProcessor(), + AnalyzeProcessor(), + ArchiveProcessor(), + ] + + config = PipelineConfig( + steps=["parse", "download", "asr", "srt", "analyze", "archive"], + output_dir=tmp_path, + skip_completed=False, + ) + + pipeline = Pipeline( + processors=processors, + config=config, + output_dir=tmp_path, + ) + + result = await pipeline.run(mock_zhihu_answer_url) + + assert result.success, f"Pipeline failed: {result.error_message}" + + # ASR 不应该被调用（文本类型跳过） + mock_asr.assert_not_called() + + # SRTProcessor 不应该被调用（非B站视频跳过） + mock_srt.assert_not_called() + + # AnalyzeProcessor 应该被调用 + mock_analyze.assert_called_once() + + +class TestMarkdownOutput: + """Markdown 输出格式测试""" + + def test_srt_output_markdown_format(self, tmp_path, sample_metadata, sample_srt_content): + """验证生成的 Markdown 文件格式正确""" + # 创建 SRT 文件 + srt_file = tmp_path / "test.srt" + srt_file.write_text(sample_srt_content, encoding='utf-8') + + # 创建元数据文件 + metadata_file = tmp_path / "metadata.json" + metadata_file.write_text(json.dumps(sample_metadata, ensure_ascii=False), encoding='utf-8') + + # 创建输出文件路径 + output_file = tmp_path / "analysis.md" + + # 使用 SRTProcessor 处理 + from kb_workflow.pipeline.processors.srt import SRTProcessor + from kb_workflow.pipeline.schemas import SRTInput + import asyncio + + processor = SRTProcessor(enable_correction=False) + step_input = SRTInput( + content_id="BV1test123456", + platform="bilibili", + srt_file=str(srt_file), + metadata_file=str(metadata_file), + output_file=str(output_file), + content_type="video" + ) + + result = asyncio.run(processor.process(step_input)) + + assert result.success + assert output_file.exists() + + # 验证文件内容 + content = output_file.read_text(encoding='utf-8') + + # 检查 frontmatter + assert "---" in content + assert "title:" in content + assert "date:" in content + assert "author:" in content + assert "platform: bilibili" in content + + # 检查内容结构 + assert "## 摘要" in content or "## 内容" in content