67 lines
1.7 KiB
Python
67 lines
1.7 KiB
Python
import asyncio
|
|
|
|
import pytest
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from services.user_service.main import app
|
|
from shared.config import settings
|
|
from shared.database import Base
|
|
|
|
# Test database URL
|
|
TEST_DATABASE_URL = (
|
|
"postgresql+asyncpg://admin:password@localhost:5432/women_safety_test"
|
|
)
|
|
|
|
# Test engine
|
|
test_engine = create_async_engine(TEST_DATABASE_URL, echo=True)
|
|
TestAsyncSession = sessionmaker(
|
|
test_engine, class_=AsyncSession, expire_on_commit=False
|
|
)
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
def event_loop():
|
|
"""Create an instance of the default event loop for the test session."""
|
|
loop = asyncio.get_event_loop_policy().new_event_loop()
|
|
yield loop
|
|
loop.close()
|
|
|
|
|
|
@pytest.fixture(scope="session")
|
|
async def setup_database():
|
|
"""Set up test database"""
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield
|
|
async with test_engine.begin() as conn:
|
|
await conn.run_sync(Base.metadata.drop_all)
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session(setup_database):
|
|
"""Create a test database session"""
|
|
async with TestAsyncSession() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
|
|
@pytest.fixture
|
|
async def client():
|
|
"""Create test client"""
|
|
async with AsyncClient(app=app, base_url="http://testserver") as ac:
|
|
yield ac
|
|
|
|
|
|
@pytest.fixture
|
|
def user_data():
|
|
"""Sample user data for testing"""
|
|
return {
|
|
"email": "test@example.com",
|
|
"password": "testpassword123",
|
|
"first_name": "Test",
|
|
"last_name": "User",
|
|
"phone": "+1234567890",
|
|
}
|