"""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_workflow.pipeline.platforms.bilibili import BilibiliAdapter from kb_workflow.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 - use AsyncMock for async method with patch.object(adapter, '_extractor') as mock_extractor: mock_extractor._extract_video_async = AsyncMock(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)