fix: 修复主程序卡死与下载器集成,新增真实数据测试

- 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 为实例属性,与实现错配),未在本次范围
This commit is contained in:
2026-08-30 12:24:26 +08:00
parent f8e8fd38b4
commit 01827224d6
14 changed files with 1327 additions and 18 deletions
+170
View File
@@ -0,0 +1,170 @@
"""
测试 seed_is_exist 功能(真实数据库)
测试说明:
- 调用真实的 seed_is_exist 函数和真实 MySQL 数据库(192.168.123.199
- 首次运行自动创建 LookingWeb 库和 seeds_info 表(幂等,可重复运行)
- 测试插入的数据在测后自动清理,不污染数据库
"""
import sys
import os
import uuid
import pymysql
import pytest
from loguru import logger
# 添加项目路径
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
from dataService.MySqlService import DatabaseConfig, MySQLDatabase, seed_is_exist, save_seed
from dataService.dto.SeedInfo import SeedInfo
CREATE_DATABASE_SQL = (
"CREATE DATABASE IF NOT EXISTS `LookingWeb` "
"DEFAULT CHARACTER SET utf8mb4 COLLATE utf8mb4_unicode_ci"
)
CREATE_TABLE_SQL = """
CREATE TABLE IF NOT EXISTS `seeds_info` (
id INT UNSIGNED NOT NULL AUTO_INCREMENT,
name VARCHAR(255) DEFAULT NULL,
seed_url VARCHAR(512) NOT NULL,
have_download VARCHAR(16) DEFAULT NULL,
create_time DATETIME DEFAULT NULL,
update_time DATETIME DEFAULT NULL,
url VARCHAR(512) DEFAULT NULL,
PRIMARY KEY (id),
KEY idx_seed_url (seed_url)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4
"""
# 本模块测试插入过的种子 URL,测后统一清理
_inserted_urls = []
def ensure_schema():
"""幂等创建 LookingWeb 库和 seeds_info 表(真实数据库)"""
# 第一步:不带库名连接,先创建数据库
conn = pymysql.connect(
host=DatabaseConfig.HOST, port=DatabaseConfig.PORT,
user=DatabaseConfig.USER, password=DatabaseConfig.PASSWORD,
charset='utf8mb4', cursorclass=pymysql.cursors.DictCursor)
try:
with conn.cursor() as cur:
cur.execute(CREATE_DATABASE_SQL)
conn.commit()
logger.info("数据库就绪: {}(已确认存在)", DatabaseConfig.DB)
finally:
conn.close()
# 第二步:连接 LookingWeb 库,创建 seeds_info 表
conn = pymysql.connect(
host=DatabaseConfig.HOST, port=DatabaseConfig.PORT,
user=DatabaseConfig.USER, password=DatabaseConfig.PASSWORD,
db=DatabaseConfig.DB, charset='utf8mb4',
cursorclass=pymysql.cursors.DictCursor)
try:
with conn.cursor() as cur:
cur.execute(CREATE_TABLE_SQL)
conn.commit()
logger.info("数据表就绪: seeds_info")
finally:
conn.close()
def make_test_seed():
"""创建带唯一 seed_url 的真实 SeedInfo 对象,并登记以便测后清理"""
url = f"magnet:?xt=urn:btih:{uuid.uuid4().hex}"
_inserted_urls.append(url)
return SeedInfo(name="测试种子", seedUrl=url, haveDownload="1", url=url)
@pytest.fixture(scope="module", autouse=True)
def db_schema():
"""模块级:确保真实数据库和表存在"""
ensure_schema()
yield
@pytest.fixture(autouse=True)
def cleanup_inserted_seeds():
"""每个测试后:清理本模块插入的种子行,保证测试可重复运行"""
yield
if not _inserted_urls:
return
with MySQLDatabase.create_connection() as db:
for seed_url in _inserted_urls:
db.execute("DELETE FROM seeds_info WHERE seed_url = %s", (seed_url,))
_inserted_urls.clear()
class TestSeedIsExist:
"""使用真实数据库测试 seed_is_exist"""
def setup_method(self):
logger.remove()
logger.add(sys.stderr, level="DEBUG")
def test_seed_not_exists_returns_false(self):
"""数据库中不存在的种子 URL 应返回 False"""
seed = make_test_seed()
logger.info("测试种子不存在场景: {}", seed.seedUrl)
result = seed_is_exist(seed)
assert result is False
def test_seed_exists_returns_true(self):
"""先通过真实的 save_seed 插入种子,seed_is_exist 应返回 True"""
seed = make_test_seed()
assert save_seed(seed) is True, "前置条件:save_seed 应成功插入种子"
logger.info("测试种子已插入: {}", seed.seedUrl)
result = seed_is_exist(seed)
assert result is True
def test_multiple_seeds_all_exist(self):
"""多个已存在种子都应被识别为存在"""
seeds = [make_test_seed() for _ in range(3)]
for seed in seeds:
assert save_seed(seed) is True
for seed in seeds:
assert seed_is_exist(seed) is True, f"种子应已存在: {seed.seedUrl}"
def test_seed_exists_after_delete_returns_false(self):
"""删除后的种子再查询应返回 False(完整的真实生命周期)"""
from dataService.MySqlService import delete_seed
seed = make_test_seed()
assert save_seed(seed) is True
assert seed_is_exist(seed) is True
assert delete_seed(seed) is True, "前置条件:delete_seed 应成功删除种子"
assert seed_is_exist(seed) is False
def test_db_unreachable_returns_false(self, monkeypatch):
"""数据库不可达时(真实错误路径)应返回 False,而不是抛异常"""
# 指向同主机一个已关闭的端口,模拟数据库不可达
monkeypatch.setattr(DatabaseConfig, "PORT", 3307)
seed = make_test_seed()
result = seed_is_exist(seed)
assert result is False
def run_tests():
"""运行所有测试"""
pytest.main([
__file__,
"-v",
"--tb=short",
"--log-cli-level=INFO"
])
if __name__ == "__main__":
run_tests()