7bc6d4e4de
- 后端改为 config/app.json；pytest 使用 config/app.test.json 与 set_config_file，不再依赖环境变量；移除 pydantic-settings。 - 前端 API/WebSocket 由 config/webui.json 经 Vite define 注入。 - 代理分数按延迟与随机取用次数计算，新增 use_count 与 proxy_scoring；保存设置时同步调度器启停。 - 仪表盘双饼图（可用/待验证协议）；设置页去掉调度器启停按钮并移动立即验证；爬取全部结束后自动提交全量验证。 - 删除 script/settings_maintain.py（此前已标记删除）。 Made-with: Cursor
131 lines
3.5 KiB
Python
131 lines
3.5 KiB
Python
"""pytest 配置文件和 fixtures"""
|
||
# 必须在任何会加载 app.core.config 的导入之前（测试库与生产库隔离）
|
||
from pathlib import Path
|
||
|
||
from app.core.config_paths import set_config_file
|
||
|
||
set_config_file(Path(__file__).resolve().parents[1] / "config" / "app.test.json")
|
||
|
||
import asyncio
|
||
import sys
|
||
|
||
if sys.platform == "win32":
|
||
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
||
|
||
import pytest
|
||
|
||
|
||
def _network_tests_enabled() -> bool:
|
||
from app.core.config import settings
|
||
|
||
return bool(getattr(settings, "run_network_tests", False))
|
||
|
||
|
||
def pytest_collection_modifyitems(config, items) -> None:
|
||
"""默认跳过 @pytest.mark.network，避免全量 pytest 触发真实爬取卡住数十分种。"""
|
||
if _network_tests_enabled():
|
||
return
|
||
skip = pytest.mark.skip(
|
||
reason=(
|
||
"外网/真实爬取用例默认跳过。需要验收时在 config/app.test.json 中设置 "
|
||
"\"run_network_tests\": true 后再运行对应文件或 -m network。"
|
||
)
|
||
)
|
||
for item in items:
|
||
if "network" in item.keywords:
|
||
item.add_marker(skip)
|
||
import pytest_asyncio
|
||
from typing import AsyncGenerator
|
||
from httpx import AsyncClient, ASGITransport
|
||
|
||
from app.api import create_app
|
||
from app.core.db import init_db, get_db
|
||
from app.core.plugin_system.registry import registry
|
||
from app.plugins import (
|
||
Fate0Plugin,
|
||
ProxyListDownloadPlugin,
|
||
Ip3366Plugin,
|
||
Ip89Plugin,
|
||
KuaiDaiLiPlugin,
|
||
SpeedXPlugin,
|
||
YunDaiLiPlugin,
|
||
ProxyScrapePlugin,
|
||
FpwProxyListDownloadPlugin,
|
||
FpwSocksSslProxyPlugin,
|
||
FpwSpysOnePlugin,
|
||
FpwProxynovaPlugin,
|
||
FpwHidemyPlugin,
|
||
FpwPremproxyPlugin,
|
||
FpwFreeproxylistsPlugin,
|
||
FpwGatherproxyPlugin,
|
||
FpwCheckerproxyPlugin,
|
||
)
|
||
from app.repositories.proxy_repo import ProxyRepository
|
||
|
||
|
||
@pytest_asyncio.fixture(scope="function")
|
||
async def app():
|
||
"""创建应用实例"""
|
||
await init_db()
|
||
async with get_db() as db:
|
||
await db.execute("DELETE FROM proxies")
|
||
await db.execute("DELETE FROM settings")
|
||
await db.commit()
|
||
|
||
registry.clear()
|
||
for plugin_cls in [
|
||
Fate0Plugin,
|
||
ProxyListDownloadPlugin,
|
||
Ip3366Plugin,
|
||
Ip89Plugin,
|
||
KuaiDaiLiPlugin,
|
||
SpeedXPlugin,
|
||
YunDaiLiPlugin,
|
||
ProxyScrapePlugin,
|
||
FpwProxyListDownloadPlugin,
|
||
FpwSocksSslProxyPlugin,
|
||
FpwSpysOnePlugin,
|
||
FpwProxynovaPlugin,
|
||
FpwHidemyPlugin,
|
||
FpwPremproxyPlugin,
|
||
FpwFreeproxylistsPlugin,
|
||
FpwGatherproxyPlugin,
|
||
FpwCheckerproxyPlugin,
|
||
]:
|
||
registry.register(plugin_cls)
|
||
|
||
test_app = create_app()
|
||
async with test_app.router.lifespan_context(test_app):
|
||
yield test_app
|
||
|
||
await asyncio.sleep(0.1)
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def client(app) -> AsyncGenerator[AsyncClient, None]:
|
||
"""创建异步 HTTP 客户端"""
|
||
transport = ASGITransport(app=app)
|
||
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
||
yield client
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db():
|
||
"""获取数据库连接"""
|
||
async with get_db() as db:
|
||
yield db
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def proxy_repo():
|
||
"""获取代理仓库"""
|
||
return ProxyRepository()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def sample_proxy(db, proxy_repo):
|
||
"""创建一个测试代理"""
|
||
await proxy_repo.insert_or_update(db, "192.168.1.1", 8080, "http", 50)
|
||
yield {"ip": "192.168.1.1", "port": 8080, "protocol": "http", "score": 50}
|
||
await proxy_repo.delete(db, "192.168.1.1", 8080)
|