cat-shark
01827224d6
- LookingWebMain: 强制 IPv4 避免 IPv6 不可达挂起；requests 加超时防卡死 - get_seed_from_element_detail: 修复限流页崩溃与 favicon 误作来源 URL，改用 canonical/传入 URL - QbittorrentDownloader: 修复 main() 中模块/类误用导致的 download_by_seed AttributeError；请求加超时 - MySqlService: 修复数据库不可达时 query/execute/close 的 None 崩溃，close 后置空 connection - 新增 cf_session: Cloudflare 人机认证 cookie 管理 + 限流退避重试 + 自动弹浏览器刷新验证 - 新增 refresh_cf_cookies.py: 弹出浏览器手动验证并保存 cf_clearance - tests/: 真实网络/数据库测试（get_page、seed_is_exist、cf_session）+ run_tests.py runner - 清理无用脚本(final_test2/verify_replace/REPLACEMENT_INFO)，.idea 与 cookie 文件入 .gitignore 注: test_qbittorrent_downloader.py 11 个用例为既有失败（mock 目标 session 为实例属性，与实现错配），未在本次范围
132 lines
5.0 KiB
Python
132 lines
5.0 KiB
Python
"""
|
||
使用主程序调用时会使用的真实 URL 测试 get_page 功能
|
||
|
||
测试说明：
|
||
- 调用真实的 get_page 函数（不 mock requests）
|
||
- 使用主程序 main() 中实际使用的 URL 获取真实页面数据
|
||
- 当前仅覆盖 get_page 功能
|
||
|
||
注意事项：
|
||
- 依赖真实网络，站点（freejavbt.com）响应可能较慢或被 Cloudflare 限流（429），
|
||
测试中对限流做了重试与容忍处理。
|
||
- freejavbt.com 同时解析出 IPv4 和 IPv6 地址，而部分环境 IPv6 不可达会导致
|
||
requests 长时间挂起，因此在导入 requests 之前强制走 IPv4。
|
||
"""
|
||
import sys
|
||
import os
|
||
import time
|
||
import socket
|
||
|
||
socket.has_ipv6 = False
|
||
|
||
import pytest
|
||
from bs4 import BeautifulSoup
|
||
from loguru import logger
|
||
|
||
# 添加项目路径
|
||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
|
||
|
||
from LookingWebMain import get_page
|
||
|
||
# 主程序 main() 中实际使用的 URL 模板
|
||
# （LookingWebMain.main: for page in range(5): url = f"https://freejavbt.com/censored/filter?c=212&page={page+1}"）
|
||
MAIN_PAGE_URL_TEMPLATE = "https://freejavbt.com/censored/filter?c=212&page={page}"
|
||
|
||
# get_page 不检查 HTTP 状态码，Cloudflare 限流时也会返回解析后的
|
||
# "Too Many Requests" 页面，这里通过页面文本识别限流。
|
||
RATE_LIMIT_MARK = "Too Many Requests"
|
||
|
||
|
||
def is_rate_limited(page_content):
|
||
"""判断页面是否为 Cloudflare 限流响应（真实数据的一部分）"""
|
||
return page_content is not None and RATE_LIMIT_MARK in page_content.text
|
||
|
||
|
||
def get_page_with_retry(url, retries=3, sleep_seconds=5):
|
||
"""
|
||
调用真实的 get_page 获取页面，遇到限流时短暂等待后重试。
|
||
返回 (结果, 是否限流过)。失败时结果可能为 None。
|
||
"""
|
||
result = None
|
||
for attempt in range(1, retries + 1):
|
||
result = get_page(url)
|
||
if not is_rate_limited(result):
|
||
return result, attempt > 1
|
||
logger.warning("页面被限流，等待 {} 秒后重试 (第 {}/{} 次): {}", sleep_seconds, attempt, retries, url)
|
||
if attempt < retries:
|
||
time.sleep(sleep_seconds)
|
||
return result, True
|
||
|
||
|
||
class TestGetPage:
|
||
"""使用主程序真实 URL 和真实数据测试 get_page 功能"""
|
||
|
||
def setup_method(self):
|
||
"""每个测试方法前的设置"""
|
||
logger.remove()
|
||
logger.add(sys.stderr, level="DEBUG")
|
||
|
||
def test_get_page_success_with_main_url(self):
|
||
"""使用主程序第 1 页的 URL 真实请求并解析页面"""
|
||
url = MAIN_PAGE_URL_TEMPLATE.format(page=1)
|
||
logger.info("使用主程序 URL 获取真实页面: {}", url)
|
||
|
||
result, _ = get_page_with_retry(url)
|
||
|
||
assert result is not None, f"真实请求失败: {url}"
|
||
assert isinstance(result, BeautifulSoup)
|
||
logger.info("真实页面获取成功，标题: {}", result.title.string if result.title else "无标题")
|
||
|
||
def test_get_page_returns_real_content(self):
|
||
"""真实页面数据应包含主程序依赖的卡片结构（card-body）"""
|
||
url = MAIN_PAGE_URL_TEMPLATE.format(page=1)
|
||
result, _ = get_page_with_retry(url)
|
||
assert result is not None, f"真实请求失败: {url}"
|
||
assert not is_rate_limited(result), f"页面被限流，无法验证真实内容: {url}"
|
||
|
||
# 主程序 get_target_element_list 依赖的结构
|
||
cards = result.find_all("div", {"class": "card-body"})
|
||
assert len(cards) > 0, "真实页面应包含 card-body 元素"
|
||
|
||
tags = result.find_all("div", {"class": "video-list-item-tag-wrapper"})
|
||
assert len(tags) > 0, "真实页面应包含 video-list-item-tag-wrapper 元素"
|
||
|
||
logger.info("真实页面包含 {} 个卡片、{} 个标签", len(cards), len(tags))
|
||
|
||
def test_get_page_all_pages_used_by_main(self):
|
||
"""遍历主程序 main() 使用的全部 5 个页面 URL"""
|
||
for page_no in range(1, 6):
|
||
url = MAIN_PAGE_URL_TEMPLATE.format(page=page_no)
|
||
result, _ = get_page_with_retry(url)
|
||
|
||
# get_page 的真实契约：请求成功并解析出页面（失败时返回 None）
|
||
assert result is not None, f"页面获取失败: {url}"
|
||
assert isinstance(result, BeautifulSoup)
|
||
|
||
# 正常页面应包含卡片；限流页面（429）是真实的服务器行为，予以容忍
|
||
cards = result.find_all("div", {"class": "card-body"})
|
||
assert len(cards) > 0 or is_rate_limited(result), \
|
||
f"页面 {url} 既无卡片内容也不是限流响应"
|
||
logger.info("页面 {} 获取成功，包含 {} 个卡片", url, len(cards))
|
||
|
||
def test_get_page_failed_request_returns_none(self):
|
||
"""对不可达的 URL 调用真实函数，应走真实网络错误路径并返回 None"""
|
||
url = "http://invalid-domain-for-test-12345.invalid"
|
||
result = get_page(url)
|
||
|
||
assert result is None
|
||
|
||
|
||
def run_tests():
|
||
"""运行所有测试"""
|
||
pytest.main([
|
||
__file__,
|
||
"-v",
|
||
"--tb=short",
|
||
"--log-cli-level=INFO"
|
||
])
|
||
|
||
|
||
if __name__ == "__main__":
|
||
run_tests()
|