0131c8b408
- Add Free_Proxy_Website-style fpw_* plugins and register them - Per-plugin crawl timeout (crawl_timeout_seconds=120); remove global crawl_timeout setting - Validator: fix connect vs total timeout on save; SOCKS session LRU cache; drop redundant semaphore - Validation handler uses single DB connection; batch upsert after crawl; WorkerPool put_nowait - Remove unused max_retries from settings API/UI; settings maintenance SQL + init_db cleanup of deprecated keys - WebSocket dashboard stats; ProxyList pool_filter and API alignment - POST /api/proxies/delete-one for IPv6-safe deletes; task poll stops on 404 - pytest uses PROXYPOOL_DB_PATH=db/proxies.test.sqlite so tests do not wipe production DB - .gitignore: explicit proxies.test.sqlite patterns; fix plugin_service ValidationException import Made-with: Cursor
108 lines
2.8 KiB
Python
108 lines
2.8 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
|
|
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)
|