This repository has been archived on 2026-08-07. You can view files and clone it. You cannot open issues or pull requests or push a commit.
Files
kb-skills/tests/test_platforms/test_zhihu.py
T
starmen1265 95218c18cd test(09-00): add Wave 0 test scaffolding for Pipeline integration
- Add mock fixtures for ZhihuCrawler CrawlResult models:
  - mock_crawl_answer_result: simulates answer crawl with comments
  - mock_crawl_article_result: simulates zhuanlan article crawl
  - mock_comments_list: test data for comment formatting
  - mock_zhihu_login_result: simulates QR code login success
  - mock_crawl_failed_result: for error handling tests

- Add test skeletons (marked skip) for Wave 1 implementation:
  - test_fetch_metadata_answer
  - test_fetch_metadata_article
  - test_fetch_content_with_comments
  - test_cookie_expired_triggers_login
  - test_idempotent_execution
  - test_comment_formatting

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-17 22:59:07 +08:00

456 lines
16 KiB
Python

"""Tests for ZhihuAdapter
Tests cover:
- Answer content extraction
- Zhuanlan (article) content extraction
- Cookie-based authentication
- Login page detection
- AuthError handling
Wave 0 Test Scaffolding (Phase 09 - Pipeline Integration):
- Mock fixtures for ZhihuCrawler CrawlResult
- Test skeletons for integration with new crawler
"""
import pytest
from unittest.mock import MagicMock, patch, PropertyMock
import asyncio
from pathlib import Path
from kb_workflow.pipeline.platforms.zhihu import ZhihuAdapter
from kb_workflow.pipeline.errors import AuthError
# =============================================================================
# Wave 0 Mock Fixtures for ZhihuCrawler Integration (Phase 09)
# =============================================================================
@pytest.fixture
def mock_crawl_answer_result():
"""Mock CrawlResult from ZhihuCrawler.crawl_answer().
Simulates successful answer crawl with metadata and content.
Used for testing ZhihuAdapter integration with new crawler.
"""
from kb_workflow.crawler.models import (
CrawlResult, AnswerData, AuthorInfo, ContentMetadata, CommentData, CommentAuthor
)
return CrawlResult(
success=True,
content_type="answer",
data=AnswerData(
id=789012,
question_id=123456,
question_title="Test Question Title for Pipeline Integration",
content="<p>This is the <strong>answer</strong> content from ZhihuCrawler.</p>",
excerpt="This is the answer content from ZhihuCrawler.",
metadata=ContentMetadata(
author=AuthorInfo(
name="Test Author",
url_token="test-author-123",
headline="Test Author Headline"
),
voteup_count=1234,
comment_count=56,
created_time=1705312800, # 2024-01-15 10:00:00 UTC
updated_time=1705399200 # 2024-01-16 10:00:00 UTC
)
),
comments=[
CommentData(
id=111,
content="Great answer! Very helpful.",
author=CommentAuthor(name="Commenter1", url_token="commenter-1"),
created_time=1705316400,
like_count=42,
child_count=3,
is_author=False
),
CommentData(
id=222,
content="Thanks for sharing!",
author=CommentAuthor(name="Commenter2", url_token="commenter-2"),
created_time=1705320000,
like_count=15,
child_count=0,
is_author=False
)
],
error=None
)
@pytest.fixture
def mock_crawl_article_result():
"""Mock CrawlResult from ZhihuCrawler.crawl_article().
Simulates successful zhuanlan article crawl.
Used for testing ZhihuAdapter integration with new crawler.
"""
from kb_workflow.crawler.models import (
CrawlResult, ArticleData, AuthorInfo, ContentMetadata, CommentData, CommentAuthor
)
return CrawlResult(
success=True,
content_type="article",
data=ArticleData(
id=123456,
title="Test Article Title for Pipeline Integration",
content="<h1>Article Content</h1><p>This is the <em>article</em> content.</p>",
excerpt="Article Content This is the article content.",
metadata=ContentMetadata(
author=AuthorInfo(
name="Test Article Author",
url_token="test-article-author",
headline="Writer and Developer"
),
voteup_count=5678,
comment_count=89,
created_time=1705226400, # 2024-01-14 10:00:00 UTC
updated_time=None
)
),
comments=[
CommentData(
id=333,
content="Excellent article!",
author=CommentAuthor(name="Reader1", url_token="reader-1"),
created_time=1705230000,
like_count=28,
child_count=1,
is_author=False
)
],
error=None
)
@pytest.fixture
def mock_comments_list():
"""Mock list of CommentData for testing comment formatting.
Provides a realistic set of comments with various attributes.
"""
from kb_workflow.crawler.models import CommentData, CommentAuthor
return [
CommentData(
id=1001,
content="This is the first comment with some content.",
author=CommentAuthor(name="User1", url_token="user-1"),
created_time=1705316400,
like_count=100,
child_count=5,
is_author=False
),
CommentData(
id=1002,
content="This is the second comment, from the author.",
author=CommentAuthor(name="AuthorName", url_token="author-token"),
created_time=1705320000,
like_count=50,
child_count=2,
is_author=True
),
CommentData(
id=1003,
content="Third comment here.",
author=CommentAuthor(name="User3", url_token="user-3"),
created_time=1705323600,
like_count=25,
child_count=0,
is_author=False
)
]
@pytest.fixture
def mock_zhihu_login_result():
"""Mock ZhihuLoginResult from ZhihuLogin.login_by_qrcode().
Simulates successful QR code login result.
Used for testing cookie refresh flow.
"""
from kb_workflow.auth.zhihu_login import ZhihuLoginResult
from pathlib import Path
return ZhihuLoginResult(
success=True,
cookie_str="z_c0=test_cookie_value; _xsrf=test_xsrf; d_c0=test_d_c0",
qrcode_path=Path("/tmp/test_qrcode.png"),
error_message=None
)
@pytest.fixture
def mock_crawl_failed_result():
"""Mock failed CrawlResult for error handling tests."""
from kb_workflow.crawler.models import CrawlResult
return CrawlResult(
success=False,
content_type="answer",
data=None,
comments=None,
error="Cookie expired or invalid"
)
class TestZhihuAdapterAnswerContent:
"""Tests for ZhihuAdapter answer extraction"""
def test_get_content_type_answer(self):
"""Test content type detection for answer URLs"""
adapter = ZhihuAdapter()
adapter._cookie = "test" # Avoid config loading
assert adapter.get_content_type("https://www.zhihu.com/question/123456/answer/789012") == "answer"
assert adapter.get_content_type("https://zhihu.com/question/123/answer/456") == "answer"
def test_get_content_type_zhuanlan(self):
"""Test content type detection for zhuanlan URLs"""
adapter = ZhihuAdapter()
adapter._cookie = "test" # Avoid config loading
assert adapter.get_content_type("https://zhuanlan.zhihu.com/p/123456") == "zhuanlan"
assert adapter.get_content_type("https://zhuanlan.zhihu.com/p/abc123") == "zhuanlan"
def test_get_content_type_unknown(self):
"""Test content type detection for unknown URLs"""
adapter = ZhihuAdapter()
adapter._cookie = "test" # Avoid config loading
assert adapter.get_content_type("https://www.zhihu.com/question/123456") == "unknown"
assert adapter.get_content_type("https://www.zhihu.com/") == "unknown"
assert adapter.get_content_type("https://example.com") == "unknown"
def test_get_output_filename(self):
"""Test output filename generation"""
adapter = ZhihuAdapter()
adapter._cookie = "test" # Avoid config loading
assert adapter.get_output_filename("789012", "answer", "metadata.json") == "zhihu_789012_metadata.json"
assert adapter.get_output_filename("123456", "zhuanlan", "content.md") == "zhihu_123456_content.md"
@pytest.mark.asyncio
async def test_fetch_metadata_answer(self, mock_zhihu_answer_result):
"""Test fetching answer metadata"""
adapter = ZhihuAdapter()
adapter._cookie = "test_cookie" # Set cookie directly
with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_answer_result):
metadata = await adapter.fetch_metadata("789012", "answer")
assert metadata["title"] == "Test Question Title"
assert metadata["author"] == "Test Author"
assert metadata["type"] == "answer"
@pytest.mark.asyncio
async def test_fetch_content_answer(self, mock_zhihu_answer_result):
"""Test fetching answer content as Markdown"""
adapter = ZhihuAdapter()
adapter._cookie = "test_cookie" # Set cookie directly
with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_answer_result):
content = await adapter.fetch_content("789012", "answer")
assert "---" in content # YAML frontmatter
assert "title:" in content
assert "Test Question Title" in content
assert "This is the answer content" in content
@pytest.mark.asyncio
async def test_fetch_metadata_no_cookie_raises_auth_error(self):
"""Test that missing cookie raises AuthError"""
adapter = ZhihuAdapter()
adapter._cookie = "" # Empty cookie
with pytest.raises(AuthError) as exc_info:
await adapter.fetch_metadata("789012", "answer")
assert exc_info.value.auth_type == "zhihu_cookie"
class TestZhihuAdapterZhuanlanContent:
"""Tests for ZhihuAdapter zhuanlan (article) extraction"""
@pytest.mark.asyncio
async def test_fetch_metadata_zhuanlan(self, mock_zhihu_zhuanlan_result):
"""Test fetching zhuanlan metadata"""
adapter = ZhihuAdapter()
adapter._cookie = "test_cookie" # Set cookie directly
with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_zhuanlan_result):
metadata = await adapter.fetch_metadata("123456", "zhuanlan")
assert metadata["title"] == "Test Article Title"
assert metadata["author"] == "Test Author"
assert metadata["type"] == "zhuanlan"
@pytest.mark.asyncio
async def test_fetch_content_zhuanlan(self, mock_zhihu_zhuanlan_result):
"""Test fetching zhuanlan content as Markdown"""
adapter = ZhihuAdapter()
adapter._cookie = "test_cookie" # Set cookie directly
with patch.object(adapter, '_extract_with_playwright', return_value=mock_zhihu_zhuanlan_result):
content = await adapter.fetch_content("123456", "zhuanlan")
assert "---" in content # YAML frontmatter
assert "title:" in content
assert "Test Article Title" in content
assert "This is the article content" in content
@pytest.mark.asyncio
async def test_detect_login_page_raises_auth_error(self):
"""Test that login page detection raises AuthError"""
adapter = ZhihuAdapter()
adapter._cookie = "expired_cookie" # Set cookie directly
# Simulate login page response (empty content or login title)
login_page_result = {
"type": "answer",
"title": "登录 - 知乎",
"content": "",
"content_markdown": "",
}
with patch.object(adapter, '_extract_with_playwright', return_value=login_page_result):
with pytest.raises(AuthError) as exc_info:
await adapter.fetch_metadata("789012", "answer")
assert exc_info.value.auth_type == "zhihu_cookie"
assert "过期" in str(exc_info.value.cause) or "expired" in str(exc_info.value.cause).lower()
@pytest.mark.asyncio
async def test_detect_short_content_raises_auth_error(self):
"""Test that very short content (< 100 chars) raises AuthError"""
adapter = ZhihuAdapter()
adapter._cookie = "test_cookie"
# Simulate response with very short content
short_content_result = {
"type": "answer",
"title": "Some Title",
"content": "x", # Very short content
"content_markdown": "",
}
with patch.object(adapter, '_extract_with_playwright', return_value=short_content_result):
with pytest.raises(AuthError) as exc_info:
await adapter.fetch_metadata("789012", "answer")
assert exc_info.value.auth_type == "zhihu_cookie"
class TestZhihuAdapterCookieHandling:
"""Tests for cookie handling"""
def test_cookie_from_settings(self):
"""Test cookie is loaded from settings"""
with patch('kb_workflow.pipeline.platforms.zhihu.get_settings') as mock_settings:
mock_secret = MagicMock()
mock_secret.get_secret_value.return_value = "test_cookie_value"
mock_settings.return_value.zhihu_cookie = mock_secret
adapter = ZhihuAdapter()
assert adapter.cookie == "test_cookie_value"
def test_platform_constant(self):
"""Test PLATFORM constant"""
adapter = ZhihuAdapter()
adapter._cookie = "test" # Avoid config loading
assert adapter.PLATFORM == "zhihu"
# =============================================================================
# Wave 0 Test Skeletons for Pipeline Integration (Phase 09)
# These tests are marked skip until ZhihuAdapter is updated to use ZhihuCrawler
# =============================================================================
@pytest.mark.skip(reason="Wave 0 skeleton - awaiting ZhihuCrawler integration")
class TestZhihuAdapterCrawlerIntegration:
"""Test skeletons for ZhihuAdapter integration with ZhihuCrawler.
These tests will verify the adapter correctly uses the new ZhihuCrawler
instead of ZhihuPlaywrightExtractor.
"""
@pytest.mark.asyncio
async def test_fetch_metadata_answer(self, mock_crawl_answer_result):
"""Test fetching answer metadata via ZhihuCrawler.
Verifies:
- ZhihuCrawler.crawl_answer() is called with correct ID
- Metadata is correctly extracted from CrawlResult
- Output format matches expected adapter response
"""
pass
@pytest.mark.asyncio
async def test_fetch_metadata_article(self, mock_crawl_article_result):
"""Test fetching zhuanlan article metadata via ZhihuCrawler.
Verifies:
- ZhihuCrawler.crawl_article() is called with correct ID
- Metadata is correctly extracted from CrawlResult
- Output format matches expected adapter response
"""
pass
@pytest.mark.asyncio
async def test_fetch_content_with_comments(
self, mock_crawl_answer_result, mock_comments_list
):
"""Test fetching content with comments appended.
Verifies:
- Content is fetched with include_comments=True
- Comments are appended to content in correct Markdown format
- Comment format: numbered list with author mention
"""
pass
@pytest.mark.asyncio
async def test_cookie_expired_triggers_login(
self, mock_crawl_failed_result, mock_zhihu_login_result
):
"""Test that expired cookie triggers QR code login flow.
Verifies:
- CrawlResult with success=False triggers login
- ZhihuLogin.login_by_qrcode() is called
- New cookie is used to retry the fetch
- User sees QR code prompt
"""
pass
@pytest.mark.asyncio
async def test_idempotent_execution(self, mock_crawl_answer_result):
"""Test that repeated calls with same content_id are idempotent.
Verifies:
- Multiple calls with same ID produce same result
- No duplicate fetches are made
- State is correctly managed
"""
pass
def test_comment_formatting(self, mock_comments_list):
"""Test comment formatting to Markdown.
Verifies:
- Comments are formatted as numbered list
- Author name is bold/mentioned
- Like count is included
- Format matches expected style:
1. **@AuthorName** (100 likes): Comment content...
"""
pass
# WAVE_0_COMPLETE: Test scaffolding ready for implementation