starmen1265
d17fbd8118
- Add test for article content returning Markdown format - Add test for YAML frontmatter with required fields - Add test for video content returning empty (ASR handling) - Add test for unsupported content type error Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
252 lines
11 KiB
Python
252 lines
11 KiB
Python
"""Tests for BilibiliAdapter
|
|
|
|
This module tests the BilibiliAdapter implementation:
|
|
1. fetch_metadata("BVxxx", "video") returns correct video metadata
|
|
2. 403 error triggers fallback strategy (API -> Playwright)
|
|
3. Network errors are retried 3 times
|
|
4. get_output_filename returns correct format
|
|
5. Failed requests call WorkflowStateManager.mark_step_failed()
|
|
"""
|
|
|
|
import pytest
|
|
from unittest.mock import MagicMock, AsyncMock, patch
|
|
import asyncio
|
|
|
|
from kb_pipeline.platforms.bilibili import BilibiliAdapter
|
|
from kb_pipeline.errors import NetworkError, AuthError
|
|
|
|
|
|
class TestBilibiliAdapterContentType:
|
|
"""Test content type identification"""
|
|
|
|
def test_get_content_type_video(self):
|
|
"""Test identifying video content from URL"""
|
|
adapter = BilibiliAdapter()
|
|
assert adapter.get_content_type("https://www.bilibili.com/video/BV1xx411c7mD") == "video"
|
|
assert adapter.get_content_type("https://bilibili.com/video/BV1test") == "video"
|
|
|
|
def test_get_content_type_article(self):
|
|
"""Test identifying article content from URL"""
|
|
adapter = BilibiliAdapter()
|
|
assert adapter.get_content_type("https://www.bilibili.com/read/cv12345678") == "article"
|
|
assert adapter.get_content_type("https://bilibili.com/read/cv12345678") == "article"
|
|
|
|
def test_get_content_type_unknown(self):
|
|
"""Test identifying unknown content type"""
|
|
adapter = BilibiliAdapter()
|
|
assert adapter.get_content_type("https://www.bilibili.com/") == "unknown"
|
|
assert adapter.get_content_type("https://www.bilibili.com/space/123456") == "unknown"
|
|
|
|
|
|
class TestBilibiliAdapterVideoMetadata:
|
|
"""Test video metadata extraction"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_video_metadata_success(self, mock_video_metadata, mock_state_manager):
|
|
"""Test 1: fetch_metadata("BVxxx", "video") returns correct video metadata"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
# Mock the extractor
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_video_metadata.return_value = mock_video_metadata
|
|
|
|
result = await adapter.fetch_metadata("BV1test12345", "video")
|
|
|
|
assert result is not None
|
|
assert result['metadata']['title'] == "Test Video Title"
|
|
assert result['metadata']['author'] == "Test Author"
|
|
assert result['metadata']['bvid'] == "BV1test12345"
|
|
assert result['stats']['view'] == 100000
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_metadata_marks_step_failed_on_error(self, mock_video_metadata, mock_state_manager):
|
|
"""Test 5: Failed requests call WorkflowStateManager.mark_step_failed()"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
# Mock the extractor to raise NetworkError
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_video_metadata.side_effect = NetworkError(
|
|
step_name="metadata",
|
|
input_summary="BV1test12345",
|
|
status_code=500
|
|
)
|
|
|
|
with pytest.raises(NetworkError):
|
|
await adapter.fetch_metadata("BV1test12345", "video")
|
|
|
|
# Verify mark_step_failed was called
|
|
mock_state_manager.mark_step_failed.assert_called_once()
|
|
call_args = mock_state_manager.mark_step_failed.call_args
|
|
assert call_args[0][0] == "metadata" # step_name
|
|
assert "500" in call_args[0][1] # error_message contains status code
|
|
|
|
|
|
class TestBilibiliAdapterOutputFilename:
|
|
"""Test output filename generation"""
|
|
|
|
def test_get_output_filename_video_metadata(self):
|
|
"""Test 4: get_output_filename returns correct format for video metadata"""
|
|
adapter = BilibiliAdapter()
|
|
result = adapter.get_output_filename("BV1test12345", "video", "metadata.json")
|
|
assert result == "bilibili_BV1test12345_metadata.json"
|
|
|
|
def test_get_output_filename_article_content(self):
|
|
"""Test output filename for article content"""
|
|
adapter = BilibiliAdapter()
|
|
result = adapter.get_output_filename("cv12345678", "article", "content.md")
|
|
assert result == "bilibili_cv12345678_content.md"
|
|
|
|
|
|
class TestBilibiliAdapterRetry:
|
|
"""Test retry mechanism"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_network_error_retries_3_times(self, mock_state_manager):
|
|
"""Test 3: Network errors are retried 3 times"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
call_count = 0
|
|
|
|
def side_effect(*args, **kwargs):
|
|
nonlocal call_count
|
|
call_count += 1
|
|
raise NetworkError(
|
|
step_name="metadata",
|
|
input_summary="BV1test12345",
|
|
status_code=503
|
|
)
|
|
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_video_metadata.side_effect = side_effect
|
|
|
|
with pytest.raises(NetworkError):
|
|
await adapter.fetch_metadata("BV1test12345", "video")
|
|
|
|
# Should have been called 3 times (retry attempts)
|
|
assert call_count == 3
|
|
|
|
|
|
class TestBilibiliAdapterFallback:
|
|
"""Test fallback strategy for 403 errors"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_403_error_triggers_fallback_note(self, mock_state_manager):
|
|
"""Test 2: 403 error should be handled (fallback logic exists)"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
# The fallback is handled inside BilibiliExtractor, not BilibiliAdapter
|
|
# BilibiliAdapter just wraps the extractor and handles state recording
|
|
# This test verifies that AuthError (403) is properly recorded
|
|
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_video_metadata.side_effect = AuthError(
|
|
step_name="metadata",
|
|
input_summary="BV1test12345",
|
|
auth_type="API"
|
|
)
|
|
|
|
with pytest.raises(AuthError):
|
|
await adapter.fetch_metadata("BV1test12345", "video")
|
|
|
|
# Verify mark_step_failed was called
|
|
mock_state_manager.mark_step_failed.assert_called_once()
|
|
|
|
|
|
class TestBilibiliAdapterArticleMetadata:
|
|
"""Test article metadata extraction"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_article_metadata_success(self, mock_article_metadata, mock_state_manager):
|
|
"""Test fetch_metadata for article content"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_article_metadata.return_value = mock_article_metadata
|
|
|
|
result = await adapter.fetch_metadata("cv12345678", "article")
|
|
|
|
assert result is not None
|
|
assert result['title'] == "Test Article Title"
|
|
assert result['author'] == "Test Author"
|
|
assert result['cvid'] == "cv12345678"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_metadata_unsupported_content_type(self, mock_state_manager):
|
|
"""Test fetch_metadata with unsupported content type raises ValueError"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
with pytest.raises(ValueError) as exc_info:
|
|
await adapter.fetch_metadata("test123", "unknown")
|
|
|
|
assert "Unsupported content type" in str(exc_info.value)
|
|
|
|
|
|
class TestBilibiliAdapterArticleContent:
|
|
"""Test article content extraction (Task 3)"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_article_content_returns_markdown(self, mock_article_metadata, mock_article_content, mock_state_manager):
|
|
"""Test 2: fetch_content("cvxxx", "article") returns Markdown format content"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_article_metadata.return_value = mock_article_metadata
|
|
|
|
# Mock the internal _fetch_article_content method
|
|
with patch.object(adapter, '_fetch_article_content', return_value=mock_article_content):
|
|
result = await adapter.fetch_content("cv12345678", "article")
|
|
|
|
# Should return a non-empty string
|
|
assert result is not None
|
|
assert len(result) > 0
|
|
# Should contain the content
|
|
assert "Test Article Title" in result or "article" in result.lower()
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_article_content_has_yaml_frontmatter(self, mock_article_metadata, mock_article_content, mock_state_manager):
|
|
"""Test 3: Markdown includes YAML Frontmatter with required fields"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
with patch.object(adapter, '_extractor') as mock_extractor:
|
|
mock_extractor.extract_article_metadata.return_value = mock_article_metadata
|
|
|
|
with patch.object(adapter, '_fetch_article_content', return_value=mock_article_content):
|
|
result = await adapter.fetch_content("cv12345678", "article")
|
|
|
|
# Check YAML frontmatter structure
|
|
assert result.startswith("---\n"), "Content should start with YAML frontmatter"
|
|
|
|
# Check required fields in frontmatter
|
|
assert "title:" in result
|
|
assert "author:" in result
|
|
assert "source_url:" in result
|
|
assert "platform:" in result
|
|
assert "content_type:" in result
|
|
assert "fetched_at:" in result
|
|
|
|
# Check specific values
|
|
assert "Test Article Title" in result
|
|
assert "Test Author" in result
|
|
assert "bilibili" in result
|
|
assert "https://www.bilibili.com/read/cv12345678" in result
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_video_content_returns_empty(self, mock_state_manager):
|
|
"""Test that fetch_content for video returns empty string (handled by ASR)"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
result = await adapter.fetch_content("BV1test12345", "video")
|
|
|
|
# Video content should return empty string (handled by ASR in Phase 3)
|
|
assert result == ""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_fetch_content_unsupported_type_raises_error(self, mock_state_manager):
|
|
"""Test that fetch_content with unsupported type raises ValueError"""
|
|
adapter = BilibiliAdapter(state_manager=mock_state_manager)
|
|
|
|
with pytest.raises(ValueError) as exc_info:
|
|
await adapter.fetch_content("test123", "unknown")
|
|
|
|
assert "Unsupported content type" in str(exc_info.value)
|