78 lines
2.0 KiB
Python
78 lines
2.0 KiB
Python
|
"""Async SQLite helpers for auth HTTP integration tests."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import uuid
|
||
|
|
from collections.abc import AsyncGenerator
|
||
|
|
from datetime import timedelta
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from sqlalchemy.ext.asyncio import (
|
||
|
|
AsyncEngine,
|
||
|
|
AsyncSession,
|
||
|
|
async_sessionmaker,
|
||
|
|
create_async_engine,
|
||
|
|
)
|
||
|
|
|
||
|
|
from app.core.db import Base, utc_now
|
||
|
|
from app.core.security import get_token_expires_at
|
||
|
|
from app.features.auth.models import RefreshToken
|
||
|
|
from app.features.user.models import User
|
||
|
|
|
||
|
|
|
||
|
|
def _auth_tables() -> list:
|
||
|
|
return [User.__table__, RefreshToken.__table__]
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def auth_async_engine() -> AsyncGenerator[AsyncEngine, None]:
|
||
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
|
|
async with engine.begin() as conn:
|
||
|
|
await conn.run_sync(
|
||
|
|
lambda sync_conn: Base.metadata.create_all(
|
||
|
|
sync_conn, tables=_auth_tables()
|
||
|
|
)
|
||
|
|
)
|
||
|
|
yield engine
|
||
|
|
await engine.dispose()
|
||
|
|
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
async def auth_session_factory(
|
||
|
|
auth_async_engine: AsyncEngine,
|
||
|
|
) -> async_sessionmaker[AsyncSession]:
|
||
|
|
return async_sessionmaker(
|
||
|
|
auth_async_engine, class_=AsyncSession, expire_on_commit=False
|
||
|
|
)
|
||
|
|
|
||
|
|
|
||
|
|
async def seed_user_with_refresh_token(
|
||
|
|
session: AsyncSession,
|
||
|
|
*,
|
||
|
|
refresh_token: str = "refresh-old",
|
||
|
|
user_id: str | None = None,
|
||
|
|
phone: str | None = None,
|
||
|
|
) -> tuple[User, str]:
|
||
|
|
uid = user_id or str(uuid.uuid4())
|
||
|
|
user = User(
|
||
|
|
id=uid,
|
||
|
|
phone=phone or f"138{uuid.uuid4().int % 100_000_000:08d}",
|
||
|
|
password_hash="hashed",
|
||
|
|
nickname="Test",
|
||
|
|
subscription_type="free",
|
||
|
|
created_at=utc_now(),
|
||
|
|
language_preference="zh",
|
||
|
|
)
|
||
|
|
token_row = RefreshToken(
|
||
|
|
id=str(uuid.uuid4()),
|
||
|
|
user_id=uid,
|
||
|
|
token=refresh_token,
|
||
|
|
expires_at=get_token_expires_at(),
|
||
|
|
created_at=utc_now(),
|
||
|
|
is_revoked=False,
|
||
|
|
)
|
||
|
|
session.add(user)
|
||
|
|
session.add(token_row)
|
||
|
|
await session.commit()
|
||
|
|
return user, refresh_token
|