59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
|
"""Tests for app.db.base — Base class, async engine + sessionmaker plumbing.
|
||
|
|
|
||
|
|
These tests build an isolated in-memory SQLite async engine; they never touch
|
||
|
|
the production MySQL engine defined in app.db.base.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import sqlalchemy as sa
|
||
|
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
||
|
|
|
||
|
|
from app.db.base import Base, async_session_maker
|
||
|
|
|
||
|
|
|
||
|
|
def test_base_is_declarative_subclass():
|
||
|
|
"""Base must be a DeclarativeBase subclass (SQLAlchemy 2.0 style)."""
|
||
|
|
from sqlalchemy.orm import DeclarativeBase
|
||
|
|
|
||
|
|
assert issubclass(Base, DeclarativeBase)
|
||
|
|
|
||
|
|
|
||
|
|
def test_async_session_maker_is_async_sessionmaker():
|
||
|
|
"""async_session_maker must be an async_sessionmaker instance."""
|
||
|
|
assert isinstance(async_session_maker, async_sessionmaker)
|
||
|
|
|
||
|
|
|
||
|
|
async def test_round_trip_through_in_memory_sqlite():
|
||
|
|
"""Build a throwaway model, create tables, insert + query a row.
|
||
|
|
|
||
|
|
Uses an in-memory async SQLite engine built locally inside the test so the
|
||
|
|
production MySQL engine is never exercised here.
|
||
|
|
"""
|
||
|
|
|
||
|
|
class _Row(Base):
|
||
|
|
__tablename__ = "test_row"
|
||
|
|
|
||
|
|
id: Mapped[int] = mapped_column(primary_key=True)
|
||
|
|
name: Mapped[str] = mapped_column(sa.String(50))
|
||
|
|
|
||
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
|
|
|
||
|
|
try:
|
||
|
|
async with engine.begin() as conn:
|
||
|
|
await conn.run_sync(Base.metadata.create_all)
|
||
|
|
|
||
|
|
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
|
||
|
|
async with SessionLocal() as session:
|
||
|
|
session.add(_Row(id=1, name="hello"))
|
||
|
|
await session.commit()
|
||
|
|
|
||
|
|
fetched = (
|
||
|
|
await session.execute(sa.select(_Row).where(_Row.id == 1))
|
||
|
|
).scalar_one()
|
||
|
|
assert fetched.id == 1
|
||
|
|
assert fetched.name == "hello"
|
||
|
|
finally:
|
||
|
|
async with engine.begin() as conn:
|
||
|
|
await conn.run_sync(Base.metadata.drop_all)
|
||
|
|
await engine.dispose()
|