"""
测试 cf_session：Cloudflare 限流/人机认证识别与自动重试逻辑
"""
import sys
import os
import pytest
from bs4 import BeautifulSoup
from loguru import logger
# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
import cf_session
class TestRateLimitDetection:
"""限流/人机认证页识别（纯函数测试）"""
def setup_method(self):
logger.remove()
logger.add(sys.stderr, level="DEBUG")
def test_detects_429_rate_limit_page(self):
"""429 限流页（Too Many Requests）应被识别"""
soup = BeautifulSoup(
'Too Many Requests'
'429
',
'html.parser')
assert cf_session.is_rate_limited(soup) is True
def test_detects_403_js_challenge_page(self):
"""403 JS 挑战页（Just a moment）应被识别"""
soup = BeautifulSoup(
'Just a moment...'
'',
'html.parser')
assert cf_session.is_rate_limited(soup) is True
def test_detects_403_attention_required_page(self):
"""403 阻止页（Attention Required）应被识别"""
soup = BeautifulSoup(
'Attention Required! | Cloudflarex',
'html.parser')
assert cf_session.is_rate_limited(soup) is True
def test_normal_page_not_rate_limited(self):
"""正常页面不应被误判"""
soup = BeautifulSoup(
'可下
',
'html.parser')
assert cf_session.is_rate_limited(soup) is False
def test_none_not_rate_limited(self):
"""None 页面不应被误判"""
assert cf_session.is_rate_limited(None) is False
class TestParseCookies:
"""agent-browser cookies 输出解析"""
def test_parse_name_value_lines(self):
stdout = ("cf_clearance=abc.def-123\n"
"bnState_x=%7B%22a%22%3A1%7D\n"
"# comment line\n")
cookies = cf_session._parse_cookies(stdout)
assert cookies == {
"cf_clearance": "abc.def-123",
"bnState_x": "%7B%22a%22%3A1%7D",
}
def test_parse_empty(self):
assert cf_session._parse_cookies("") == {}
class TestFetchWithRetry:
"""自动重试的真实请求验证"""
def setup_method(self):
logger.remove()
logger.add(sys.stderr, level="DEBUG")
def test_fetch_real_url_ok(self):
"""真实 URL 正常返回 (soup, 'ok')"""
url = "https://freejavbt.com/censored/filter?c=212&page=1"
soup, status = cf_session.fetch_with_retry(url, (10, 60), retries=2, retry_delay=2)
assert status == "ok"
assert soup is not None
assert len(soup.find_all("div", {"class": "card-body"})) > 0
def test_fetch_network_error(self):
"""不可达 URL 返回 (None, 'error')"""
soup, status = cf_session.fetch_with_retry(
"http://invalid-domain-for-test-12345.invalid", (5, 10), retries=2, retry_delay=1)
assert status == "error"
assert soup is None
def run_tests():
"""运行所有测试"""
pytest.main([
__file__,
"-v",
"--tb=short",
"--log-cli-level=INFO"
])
if __name__ == "__main__":
run_tests()