feat(08-03): integrate proxy, delay, and retry into ZhihuCrawler
- Add CrawlerConfig with ProxyConfig, delay, and retry settings - Integrate proxy configuration via _start_browser_with_proxy() - Add random delay via _apply_random_delay() - Add retry logic via _crawl_with_retry using tenacity - Add ZhihuApiInterceptor tests for x-zse-96 handling - All tests pass
This commit is contained in:
@@ -0,0 +1,78 @@
|
||||
"""Tests for API Interception and x-zse-96 Signature Handling"""
|
||||
|
||||
import pytest
|
||||
from unittest.mock import Mock
|
||||
|
||||
from kb_workflow.crawler.zhihu_crawler import ZhihuApiInterceptor
|
||||
|
||||
|
||||
class TestApiIntercept:
|
||||
"""Tests for API interception and x-zse-96 signature handling."""
|
||||
|
||||
def test_intercept_answer_api(self):
|
||||
"""Intercept answer API and capture response data."""
|
||||
interceptor = ZhihuApiInterceptor()
|
||||
|
||||
# Mock route object
|
||||
route = Mock()
|
||||
route.request.url = "https://www.zhihu.com/api/v4/answers/12345"
|
||||
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {"id": 12345, "content": "test answer content"}
|
||||
route.fetch.return_value = mock_response
|
||||
|
||||
interceptor.handle_route(route)
|
||||
|
||||
assert "answer" in interceptor.captured_data
|
||||
assert interceptor.captured_data["answer"]["id"] == 12345
|
||||
|
||||
def test_intercept_article_api(self):
|
||||
"""Intercept article API and capture response data."""
|
||||
interceptor = ZhihuApiInterceptor()
|
||||
|
||||
route = Mock()
|
||||
route.request.url = "https://zhuanlan.zhihu.com/api/v4/articles/67890"
|
||||
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {"id": 67890, "title": "test article"}
|
||||
route.fetch.return_value = mock_response
|
||||
|
||||
interceptor.handle_route(route)
|
||||
|
||||
assert "article" in interceptor.captured_data
|
||||
assert interceptor.captured_data["article"]["id"] == 67890
|
||||
|
||||
def test_non_api_requests_continue(self):
|
||||
"""Non-API requests continue normally."""
|
||||
interceptor = ZhihuApiInterceptor()
|
||||
|
||||
route = Mock()
|
||||
route.request.url = "https://www.zhihu.com/static/some.js"
|
||||
|
||||
interceptor.handle_route(route)
|
||||
|
||||
# Should call continue_ for non-API requests
|
||||
route.continue_.assert_called_once()
|
||||
route.fulfill.assert_not_called()
|
||||
|
||||
def test_x_zse_96_auto_handled(self):
|
||||
"""x-zse-96 signature is automatically handled by browser via route.fetch().
|
||||
|
||||
The interceptor automatically uses the browser's fetch() which includes
|
||||
the x-zse-96 signature headers generated by the browser context.
|
||||
"""
|
||||
interceptor = ZhihuApiInterceptor()
|
||||
|
||||
route = Mock()
|
||||
route.request.url = "https://www.zhihu.com/api/v4/answers/123"
|
||||
|
||||
mock_response = Mock()
|
||||
mock_response.json.return_value = {"id": 123, "signed": True}
|
||||
route.fetch.return_value = mock_response
|
||||
|
||||
interceptor.handle_route(route)
|
||||
|
||||
# fetch() is called (which would include browser-generated x-zse-96)
|
||||
route.fetch.assert_called_once()
|
||||
# fulfill() returns response to page
|
||||
route.fulfill.assert_called_once()
|
||||
Reference in New Issue
Block a user