ce667dba13
- pytest_collection_modifyitems: skip @pytest.mark.network unless PROXYPOOL_RUN_NETWORK_TESTS=1 - Document opt-in in tests/README.md - e2e: replace removed crawl_timeout with validation_timeout Made-with: Cursor
128 lines
3.5 KiB
Python
128 lines
3.5 KiB
Python
"""pytest 配置文件和 fixtures"""
|
||
# 必须在任何 app.* 导入之前：下方 app fixture 会清空表，不可与生产共用 db/proxies.sqlite
|
||
import os
|
||
|
||
os.environ["PROXYPOOL_DB_PATH"] = "db/proxies.test.sqlite"
|
||
|
||
import asyncio
|
||
import sys
|
||
|
||
if sys.platform == "win32":
|
||
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
|
||
|
||
import pytest
|
||
|
||
|
||
def _network_tests_enabled() -> bool:
|
||
v = os.environ.get("PROXYPOOL_RUN_NETWORK_TESTS", "").strip().lower()
|
||
return v in ("1", "true", "yes", "on")
|
||
|
||
|
||
def pytest_collection_modifyitems(config, items) -> None:
|
||
"""默认跳过 @pytest.mark.network，避免全量 pytest 触发真实爬取卡住数十分种。"""
|
||
if _network_tests_enabled():
|
||
return
|
||
skip = pytest.mark.skip(
|
||
reason=(
|
||
"外网/真实爬取用例默认跳过。需要验收时设置环境变量 "
|
||
"PROXYPOOL_RUN_NETWORK_TESTS=1 后再运行对应文件或 -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)
|