120 lines
3.7 KiB
Python
120 lines
3.7 KiB
Python
import uuid
|
|
import random
|
|
|
|
import pytest
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from proxy_pool.app import create_app
|
|
from proxy_pool.proxy.models import Proxy, ProxyProtocol, ProxySource, ProxyStatus
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
app = create_app()
|
|
async with app.router.lifespan_context(app):
|
|
transport = ASGITransport(app=app)
|
|
async with AsyncClient(transport=transport, base_url="http://test") as client:
|
|
yield client
|
|
|
|
|
|
@pytest.fixture
|
|
async def seeded_client(client):
|
|
source_resp = await client.post(
|
|
"/sources",
|
|
json={
|
|
"url": f"https://example.com/test-{uuid.uuid4().hex[:8]}.txt",
|
|
"parser_name": "plaintext",
|
|
},
|
|
)
|
|
source_id = source_resp.json()["id"]
|
|
|
|
base = random.randint(1, 200)
|
|
app = client._transport.app
|
|
session_factory = app.state.session_factory
|
|
async with session_factory() as db:
|
|
for i in range(5):
|
|
proxy = Proxy(
|
|
ip=f"203.0.{base}.{i + 1}",
|
|
port=8080 + i,
|
|
protocol=ProxyProtocol.HTTP,
|
|
source_id=uuid.UUID(source_id),
|
|
status=ProxyStatus.ACTIVE if i < 3 else ProxyStatus.DEAD,
|
|
score=0.9 - (i * 0.1),
|
|
country="US" if i < 3 else "DE",
|
|
)
|
|
db.add(proxy)
|
|
await db.commit()
|
|
|
|
yield client
|
|
|
|
|
|
class TestProxyQueryRoutes:
|
|
async def test_list_proxies_empty(self, client):
|
|
response = await client.get("/proxies")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "items" in data
|
|
assert "total_count" in data
|
|
|
|
async def test_list_proxies_with_data(self, seeded_client):
|
|
response = await seeded_client.get("/proxies")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["total_count"] >= 5
|
|
|
|
async def test_filter_by_status(self, seeded_client):
|
|
response = await seeded_client.get("/proxies?status=active")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for item in data["items"]:
|
|
assert item["status"] == "active"
|
|
|
|
async def test_filter_by_country(self, seeded_client):
|
|
response = await seeded_client.get("/proxies?country=US")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for item in data["items"]:
|
|
assert item["country"] == "US"
|
|
|
|
async def test_filter_by_min_score(self, seeded_client):
|
|
response = await seeded_client.get("/proxies?min_score=0.7")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
for item in data["items"]:
|
|
assert item["score"] >= 0.7
|
|
|
|
async def test_sort_by_score_desc(self, seeded_client):
|
|
response = await seeded_client.get(
|
|
"/proxies?sort_by=score&sort_order=desc"
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
scores = [item["score"] for item in response.json()["items"]]
|
|
assert scores == sorted(scores, reverse=True)
|
|
|
|
async def test_pagination(self, seeded_client):
|
|
response = await seeded_client.get("/proxies?limit=2&offset=0")
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["items"]) <= 2
|
|
assert data["limit"] == 2
|
|
assert data["offset"] == 0
|
|
|
|
async def test_invalid_limit(self, client):
|
|
response = await client.get("/proxies?limit=999")
|
|
|
|
assert response.status_code == 422
|
|
|
|
async def test_get_proxy_not_found(self, client):
|
|
response = await client.get(
|
|
"/proxies/00000000-0000-0000-0000-000000000000"
|
|
)
|
|
|
|
assert response.status_code == 404 |