import pytest
import pytest_asyncio
import aiohttp
from proxy_hopper.backend.memory import MemoryBackend
from proxy_hopper.pool_store import IPPoolStore
from proxy_hopper.config import TargetConfig, ResolvedIP
from proxy_hopper.server import ProxyServer
from proxy_hopper.target_manager import TargetManager
from proxy_hopper_testserver import MockProxyPool, UpstreamServer
@pytest_asyncio.fixture
async def upstream():
async with UpstreamServer() as server:
yield server
@pytest_asyncio.fixture
async def proxies():
async with MockProxyPool(count=3) as pool:
yield pool
async def test_request_succeeds(proxies, upstream):
cfg = TargetConfig(
name="test",
regex=r".*",
resolved_ips=[ResolvedIP(host=h, port=p) for h, p in ...],
min_request_interval=0,
num_retries=0,
...
)
raw_backend = MemoryBackend()
await raw_backend.start()
pool_store = IPPoolStore(raw_backend)
mgr = TargetManager(cfg, pool_store)
server = ProxyServer([mgr], host="127.0.0.1", port=0, pool_store=pool_store)
await server.start()
port = server._server.sockets[0].getsockname()[1]
try:
async with aiohttp.ClientSession() as client:
async with client.get(
f"http://127.0.0.1:{port}/test",
headers={"X-Proxy-Hopper-Target": upstream.url},
) as resp:
assert resp.status == 200
finally:
await server.stop()
await raw_backend.stop()
