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
179 lines
6.9 KiB
Python
179 lines
6.9 KiB
Python
"""代理 API 集成测试 - 测试 /api/proxies/* 所有接口"""
|
||
import pytest
|
||
|
||
|
||
class TestProxiesAPI:
|
||
"""测试代理相关 API"""
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_stats(self, client):
|
||
"""测试 GET /api/proxies/stats"""
|
||
response = await client.get("/api/proxies/stats")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
assert "data" in data
|
||
assert "total" in data["data"]
|
||
assert "pending" in data["data"]
|
||
assert "available" in data["data"]
|
||
assert "scheduler_running" in data["data"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_proxies(self, client):
|
||
"""测试 POST /api/proxies"""
|
||
request_data = {
|
||
"page": 1,
|
||
"page_size": 10,
|
||
"protocol": None,
|
||
"min_score": 0,
|
||
"sort_by": "last_check",
|
||
"sort_order": "DESC"
|
||
}
|
||
response = await client.post("/api/proxies", json=request_data)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
assert "list" in data["data"]
|
||
assert "total" in data["data"]
|
||
assert "page" in data["data"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_proxies_with_protocol_filter(self, client):
|
||
"""测试 POST /api/proxies 带协议过滤"""
|
||
request_data = {
|
||
"page": 1,
|
||
"page_size": 10,
|
||
"protocol": "http",
|
||
"min_score": 0,
|
||
}
|
||
response = await client.post("/api/proxies", json=request_data)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_list_proxies_invalid_protocol(self, client):
|
||
"""测试 POST /api/proxies 无效协议"""
|
||
request_data = {
|
||
"page": 1,
|
||
"page_size": 10,
|
||
"protocol": "invalid",
|
||
}
|
||
response = await client.post("/api/proxies", json=request_data)
|
||
assert response.status_code == 422 # 验证错误
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_get_random_proxy_empty(self, client):
|
||
"""测试 GET /api/proxies/random - 空数据库"""
|
||
response = await client.get("/api/proxies/random")
|
||
# 可能返回 200(有数据) 或 404(无数据)
|
||
assert response.status_code in [200, 404]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_proxy_post_json(self, client, sample_proxy):
|
||
"""测试 POST /api/proxies/delete-one（前端默认路径，兼容 IPv6）"""
|
||
response = await client.post(
|
||
"/api/proxies/delete-one",
|
||
json={"ip": sample_proxy["ip"], "port": sample_proxy["port"]},
|
||
)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_proxy(self, client, sample_proxy):
|
||
"""测试 DELETE /api/proxies/{ip}/{port}"""
|
||
response = await client.delete(f"/api/proxies/{sample_proxy['ip']}/{sample_proxy['port']}")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_one_ipv6(self, client, db, proxy_repo):
|
||
"""POST delete-one 可删除含冒号的 IP（路径 DELETE 无法可靠表达）"""
|
||
await proxy_repo.insert_or_update(db, "2001:db8::1", 18080, "http", 40)
|
||
r = await client.post(
|
||
"/api/proxies/delete-one",
|
||
json={"ip": "2001:db8::1", "port": 18080},
|
||
)
|
||
assert r.status_code == 200
|
||
assert r.json()["code"] == 200
|
||
left = await proxy_repo.get_by_ip_port(db, "2001:db8::1", 18080)
|
||
assert left is None
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_delete_nonexistent_proxy(self, client):
|
||
"""测试 DELETE /api/proxies/{ip}/{port} - 不存在的代理"""
|
||
response = await client.delete("/api/proxies/999.999.999.999/99999")
|
||
# 即使代理不存在也返回成功（幂等删除）
|
||
assert response.status_code == 200
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_batch_delete_proxies(self, client):
|
||
"""测试 POST /api/proxies/batch-delete"""
|
||
request_data = {
|
||
"proxies": [
|
||
{"ip": "192.168.1.1", "port": 8080},
|
||
{"ip": "192.168.1.2", "port": 8081},
|
||
]
|
||
}
|
||
response = await client.post("/api/proxies/batch-delete", json=request_data)
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
assert "deleted_count" in data["data"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_batch_delete_too_many(self, client):
|
||
"""测试 POST /api/proxies/batch-delete - 超过限制"""
|
||
request_data = {
|
||
"proxies": [{"ip": f"192.168.1.{i}", "port": 8080} for i in range(1001)]
|
||
}
|
||
response = await client.post("/api/proxies/batch-delete", json=request_data)
|
||
assert response.status_code == 422 # 验证错误
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_clean_invalid_proxies(self, client):
|
||
"""测试 DELETE /api/proxies/clean-invalid"""
|
||
response = await client.delete("/api/proxies/clean-invalid")
|
||
assert response.status_code == 200
|
||
data = response.json()
|
||
assert data["code"] == 200
|
||
assert "deleted_count" in data["data"]
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_export_proxies_csv(self, client):
|
||
"""测试 GET /api/proxies/export/csv"""
|
||
response = await client.get("/api/proxies/export/csv")
|
||
assert response.status_code == 200
|
||
assert response.headers["content-type"].startswith("text/csv")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_export_proxies_txt(self, client):
|
||
"""测试 GET /api/proxies/export/txt"""
|
||
response = await client.get("/api/proxies/export/txt")
|
||
assert response.status_code == 200
|
||
assert response.headers["content-type"].startswith("text/plain")
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_export_proxies_json(self, client):
|
||
"""测试 GET /api/proxies/export/json"""
|
||
response = await client.get("/api/proxies/export/json")
|
||
assert response.status_code == 200
|
||
assert response.headers["content-type"] == "application/json"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_export_proxies_json_empty_database(self, client):
|
||
"""测试 GET /api/proxies/export/json - 空数据库"""
|
||
response = await client.get("/api/proxies/export/json")
|
||
assert response.status_code == 200
|
||
assert response.headers["content-type"] == "application/json"
|
||
# 空数据库应返回空列表 JSON
|
||
assert response.content.strip() == b"[]"
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_export_proxies_invalid_format(self, client):
|
||
"""测试 GET /api/proxies/export/invalid - 无效格式"""
|
||
response = await client.get("/api/proxies/export/invalid")
|
||
assert response.status_code == 400 # 错误响应
|