test: add integration test for proxy model round-trip

This commit is contained in:
agatha 2026-03-14 14:03:10 -04:00
parent e66b3caed0
commit 78735594cf
2 changed files with 67 additions and 0 deletions

View File

@ -0,0 +1,43 @@
import asyncio
from collections.abc import AsyncGenerator
import pytest
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from proxy_pool.config import Settings
from proxy_pool.db.base import Base
@pytest.fixture(scope="session")
def event_loop():
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session")
def test_settings() -> Settings:
return Settings(
database_url="postgresql+asyncpg://proxypool:proxypool@localhost:5432/proxypool",
redis_url="redis://localhost:6379/1",
secret_key="test-secret",
log_level="DEBUG",
)
@pytest.fixture(scope="session")
async def engine(test_settings: Settings):
engine = create_async_engine(test_settings.database_url)
yield engine
await engine.dispose()
@pytest.fixture
async def db_session(engine) -> AsyncGenerator[AsyncSession, None]:
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
async with session_factory() as session:
yield session
await session.rollback()

View File

@ -0,0 +1,24 @@
from proxy_pool.proxy.models import Proxy, ProxyProtocol, ProxySource, ProxyStatus
async def test_create_source_and_proxy(db_session):
source = ProxySource(
url="https://example.com/proxies.txt",
parser_name="plaintext",
)
db_session.add(source)
await db_session.flush()
proxy = Proxy(
ip="203.0.113.42",
port=8080,
protocol=ProxyProtocol.HTTP,
source_id=source.id,
)
db_session.add(proxy)
await db_session.flush()
assert proxy.id is not None
assert proxy.status == ProxyStatus.UNCHECKED
assert proxy.score == 0.0
assert proxy.source_id == source.id