test(02-02): add failing tests for PlatformAdapter base class
- Add test for abstract class instantiation prevention - Add test for incomplete implementation detection - Add test for complete implementation instantiation - Add test for async method signatures - Add shared fixtures for platform tests Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Tests for platform adapters"""
|
||||
@@ -0,0 +1,134 @@
|
||||
"""Shared fixtures for platform adapter tests"""
|
||||
|
||||
import pytest
|
||||
from pathlib import Path
|
||||
import tempfile
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from kb_pipeline.state import WorkflowStateManager
|
||||
from kb_pipeline.schemas import WorkflowState
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def tmp_content_dir():
|
||||
"""Create a temporary directory for content files"""
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
yield Path(tmpdir)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def state_manager(tmp_content_dir):
|
||||
"""Create a WorkflowStateManager with a temporary directory"""
|
||||
return WorkflowStateManager(tmp_content_dir)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_state_manager():
|
||||
"""Create a mock WorkflowStateManager for testing"""
|
||||
return MagicMock(spec=WorkflowStateManager)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_video_metadata():
|
||||
"""Mock video metadata response from BilibiliExtractor"""
|
||||
return {
|
||||
'metadata': {
|
||||
'title': 'Test Video Title',
|
||||
'author': 'Test Author',
|
||||
'bvid': 'BV1test12345',
|
||||
'aid': 12345678,
|
||||
'mid': 9876543,
|
||||
'pubtime': '2024-01-15 10:30:00',
|
||||
'duration': '10:25',
|
||||
'url': 'https://www.bilibili.com/video/BV1test12345',
|
||||
'tags': 'test, video, bilibili',
|
||||
},
|
||||
'stats': {
|
||||
'view': 100000,
|
||||
'danmaku': 500,
|
||||
'reply': 200,
|
||||
'favorite': 1000,
|
||||
'coin': 800,
|
||||
'share': 300,
|
||||
'like': 5000,
|
||||
},
|
||||
'subtitle': '',
|
||||
'description': 'This is a test video description.'
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_article_metadata():
|
||||
"""Mock article metadata response from BilibiliExtractor"""
|
||||
return {
|
||||
'cvid': 'cv12345678',
|
||||
'title': 'Test Article Title',
|
||||
'author': 'Test Author',
|
||||
'mid': '9876543',
|
||||
'publish_time': '2024-01-15 10:30:00',
|
||||
'view_count': 50000,
|
||||
'like_count': 2000,
|
||||
'coin_count': 500,
|
||||
'share_count': 100,
|
||||
'reply_count': 150,
|
||||
'url': 'https://www.bilibili.com/read/cv12345678',
|
||||
'summary': 'This is a test article summary.',
|
||||
'word_count': 3000,
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_article_content():
|
||||
"""Mock article content HTML/Markdown"""
|
||||
return """# Test Article Title
|
||||
|
||||
This is the first paragraph of the test article.
|
||||
|
||||
## Section 1
|
||||
|
||||
Content for section 1.
|
||||
|
||||
## Section 2
|
||||
|
||||
Content for section 2.
|
||||
"""
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_zhihu_extractor():
|
||||
"""Mock ZhihuPlaywrightExtractor for testing."""
|
||||
from unittest.mock import MagicMock
|
||||
extractor = MagicMock()
|
||||
extractor.extract = MagicMock()
|
||||
return extractor
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_zhihu_answer_result():
|
||||
"""Sample Zhihu answer extraction result."""
|
||||
return {
|
||||
"type": "answer",
|
||||
"title": "Test Question Title",
|
||||
"author": "Test Author",
|
||||
"content": "This is the answer content.",
|
||||
"content_markdown": "# Answer\n\nThis is the answer content.",
|
||||
"publish_time": "2024-01-15T10:30:00",
|
||||
"url": "https://www.zhihu.com/question/123456/answer/789012",
|
||||
"question_id": "123456",
|
||||
"answer_id": "789012",
|
||||
}
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_zhihu_zhuanlan_result():
|
||||
"""Sample Zhihu zhuanlan extraction result."""
|
||||
return {
|
||||
"type": "zhuanlan",
|
||||
"title": "Test Article Title",
|
||||
"author": "Test Author",
|
||||
"content": "This is the article content.",
|
||||
"content_markdown": "# Article\n\nThis is the article content.",
|
||||
"publish_time": "2024-01-15T10:30:00",
|
||||
"url": "https://zhuanlan.zhihu.com/p/123456",
|
||||
"article_id": "123456",
|
||||
}
|
||||
@@ -0,0 +1,122 @@
|
||||
"""Tests for PlatformAdapter base class
|
||||
|
||||
This module tests the interface contracts for the PlatformAdapter abstract base class:
|
||||
1. PlatformAdapter is an abstract class and cannot be instantiated directly
|
||||
2. Subclasses must implement all abstract methods
|
||||
"""
|
||||
|
||||
import pytest
|
||||
from abc import ABC
|
||||
from typing import Dict, Any
|
||||
|
||||
from kb_pipeline.platforms.base import PlatformAdapter
|
||||
|
||||
|
||||
class TestPlatformAdapterInterface:
|
||||
"""Test the PlatformAdapter interface contracts"""
|
||||
|
||||
def test_platform_adapter_is_abstract(self):
|
||||
"""Test 1: PlatformAdapter is an abstract class and cannot be instantiated directly"""
|
||||
# Verify it's an ABC subclass
|
||||
assert issubclass(PlatformAdapter, ABC)
|
||||
|
||||
# Verify it cannot be instantiated directly
|
||||
with pytest.raises(TypeError) as exc_info:
|
||||
PlatformAdapter()
|
||||
|
||||
# Error should mention abstract methods
|
||||
assert "abstract" in str(exc_info.value).lower()
|
||||
|
||||
def test_subclass_must_implement_all_methods(self):
|
||||
"""Test 2: Subclass must implement all abstract methods"""
|
||||
# Create an incomplete implementation
|
||||
class IncompleteAdapter(PlatformAdapter):
|
||||
def get_content_type(self, url: str) -> str:
|
||||
return "video"
|
||||
|
||||
# Missing: fetch_metadata, fetch_content, get_output_filename
|
||||
# This should fail to instantiate
|
||||
|
||||
with pytest.raises(TypeError) as exc_info:
|
||||
IncompleteAdapter()
|
||||
|
||||
# Verify error mentions missing methods
|
||||
error_msg = str(exc_info.value).lower()
|
||||
assert "abstract" in error_msg or "implement" in error_msg
|
||||
|
||||
def test_complete_implementation_can_be_instantiated(self):
|
||||
"""Test that a complete implementation can be instantiated"""
|
||||
class CompleteAdapter(PlatformAdapter):
|
||||
def get_content_type(self, url: str) -> str:
|
||||
return "video"
|
||||
|
||||
async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]:
|
||||
return {"title": "test"}
|
||||
|
||||
async def fetch_content(self, content_id: str, content_type: str) -> str:
|
||||
return "# Test Content"
|
||||
|
||||
def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str:
|
||||
return f"test_{content_id}_{suffix}"
|
||||
|
||||
# Should instantiate without error
|
||||
adapter = CompleteAdapter()
|
||||
assert adapter is not None
|
||||
assert adapter.get_content_type("https://example.com/video/123") == "video"
|
||||
|
||||
|
||||
class TestPlatformAdapterMethods:
|
||||
"""Test the abstract method signatures"""
|
||||
|
||||
def test_get_content_type_signature(self):
|
||||
"""Test get_content_type method signature"""
|
||||
class TestAdapter(PlatformAdapter):
|
||||
def get_content_type(self, url: str) -> str:
|
||||
return "video"
|
||||
|
||||
async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]:
|
||||
return {}
|
||||
|
||||
async def fetch_content(self, content_id: str, content_type: str) -> str:
|
||||
return ""
|
||||
|
||||
def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str:
|
||||
return ""
|
||||
|
||||
adapter = TestAdapter()
|
||||
# Method should accept URL string and return content type string
|
||||
result = adapter.get_content_type("https://www.bilibili.com/video/BV1test")
|
||||
assert isinstance(result, str)
|
||||
|
||||
def test_fetch_metadata_is_async(self):
|
||||
"""Test fetch_metadata is an async method"""
|
||||
import inspect
|
||||
assert inspect.iscoroutinefunction(PlatformAdapter.fetch_metadata)
|
||||
|
||||
def test_fetch_content_is_async(self):
|
||||
"""Test fetch_content is an async method"""
|
||||
import inspect
|
||||
assert inspect.iscoroutinefunction(PlatformAdapter.fetch_content)
|
||||
|
||||
def test_get_output_filename_signature(self):
|
||||
"""Test get_output_filename method signature"""
|
||||
class TestAdapter(PlatformAdapter):
|
||||
def get_content_type(self, url: str) -> str:
|
||||
return ""
|
||||
|
||||
async def fetch_metadata(self, content_id: str, content_type: str) -> Dict[str, Any]:
|
||||
return {}
|
||||
|
||||
async def fetch_content(self, content_id: str, content_type: str) -> str:
|
||||
return ""
|
||||
|
||||
def get_output_filename(self, content_id: str, content_type: str, suffix: str) -> str:
|
||||
return f"{content_type}_{content_id}_{suffix}"
|
||||
|
||||
adapter = TestAdapter()
|
||||
# Method should accept content_id, content_type, suffix and return filename string
|
||||
result = adapter.get_output_filename("BV1test", "video", "metadata.json")
|
||||
assert isinstance(result, str)
|
||||
assert "BV1test" in result
|
||||
assert "video" in result
|
||||
assert "metadata.json" in result
|
||||
Reference in New Issue
Block a user