qzl
88 lines
2.5 KiB
Python
88 lines
2.5 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
from uuid import UUID, uuid4
|
|
|
|
import pytest
|
|
from sqlalchemy import Column, String, Table
|
|
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
|
from sqlalchemy.orm import Mapped, mapped_column
|
|
|
|
from core.db.base import Base, SoftDeleteMixin
|
|
from core.db.base_repository import BaseRepository
|
|
|
|
|
|
class Widget(SoftDeleteMixin, Base):
|
|
__tablename__ = "widgets"
|
|
|
|
id: Mapped[UUID] = mapped_column(primary_key=True)
|
|
name: Mapped[str] = mapped_column(String(50), nullable=False)
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_engine():
|
|
auth_users = Table(
|
|
"users",
|
|
Base.metadata,
|
|
Column("id", String, primary_key=True),
|
|
schema="auth",
|
|
extend_existing=True,
|
|
)
|
|
engine = create_async_engine("sqlite+aiosqlite:///:memory:", echo=False)
|
|
async with engine.begin() as conn:
|
|
await conn.exec_driver_sql("ATTACH DATABASE ':memory:' AS auth")
|
|
await conn.run_sync(Base.metadata.create_all)
|
|
yield engine
|
|
Base.metadata.remove(auth_users)
|
|
await engine.dispose()
|
|
|
|
|
|
@pytest.fixture
|
|
async def db_session(db_engine):
|
|
async_session = async_sessionmaker(
|
|
bind=db_engine,
|
|
class_=AsyncSession,
|
|
expire_on_commit=False,
|
|
)
|
|
async with async_session() as session:
|
|
yield session
|
|
await session.rollback()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_by_id_filters_soft_deleted(db_session: AsyncSession) -> None:
|
|
repository = BaseRepository(db_session, Widget)
|
|
widget_id = uuid4()
|
|
|
|
widget = Widget(id=widget_id, name="widget")
|
|
db_session.add(widget)
|
|
await db_session.commit()
|
|
|
|
found = await repository.get_by_id(widget_id)
|
|
assert found is not None
|
|
|
|
deleted = await repository.soft_delete_by_id(widget_id)
|
|
assert deleted is not None
|
|
assert deleted.deleted_at is not None
|
|
|
|
missing = await repository.get_by_id(widget_id)
|
|
assert missing is None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_soft_delete_sets_timestamp(db_session: AsyncSession) -> None:
|
|
repository = BaseRepository(db_session, Widget)
|
|
widget_id = uuid4()
|
|
|
|
widget = Widget(id=widget_id, name="widget")
|
|
db_session.add(widget)
|
|
await db_session.commit()
|
|
|
|
deleted = await repository.soft_delete_by_id(widget_id)
|
|
assert deleted is not None
|
|
assert isinstance(deleted.deleted_at, datetime)
|
|
deleted_at = deleted.deleted_at
|
|
if deleted_at.tzinfo is None:
|
|
deleted_at = deleted_at.replace(tzinfo=timezone.utc)
|
|
assert deleted_at <= datetime.now(timezone.utc)
|