starmen1265
9e1452d2a9
- 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
79 lines
2.6 KiB
Python
79 lines
2.6 KiB
Python
"""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()
|