flowCreate.solutions

Generic Example — CRUD Tests

CRUD tests validate business logic and persistence without going through the HTTP layer. These tests should be fast, deterministic, and focus on the CRUD module’s contract.

What CRUD tests should cover

  • Create: required fields, defaults, tenant/resource ownership, and any derived fields.
  • Read: retrieving by id; not-found behavior for invalid ids.
  • Update: only intended fields change; validation/guardrails enforced.
  • Delete: entity no longer returns in reads (or is marked deleted for soft-delete patterns).

Prerequisites (fixtures)

These examples assume:

  • db_session (async AsyncSession) exists in tests/conftest.py (see Generic Example - Conftest).
  • tenant_id exists in tests/conftest.py (or equivalent “scope id” for your app).
  • sample_item exists in tests/{module}/fixtures.py and creates a persisted entity.

Template: tests/{module}/test_crud.py

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

pytestmark = pytest.mark.asyncio

# Import your module's CRUD + schemas
from database.items import crud  # adjust to your project
from database.items.schemas import ItemCreate, ItemUpdate  # adjust to your project


async def test_create_item(db_session: AsyncSession, tenant_id: str):
    item_in = ItemCreate(name="Test Item", price=100)
    item = await crud.create_item(db_session, item_in, tenant_id=tenant_id)

    assert item.id is not None
    assert item.tenant_id == tenant_id
    assert item.name == "Test Item"


async def test_get_item_by_id(db_session: AsyncSession, sample_item):
    item = await crud.get_item_by_id(db_session, sample_item.id)
    assert item is not None
    assert item.id == sample_item.id


async def test_update_item(db_session: AsyncSession, sample_item):
    update = ItemUpdate(name="Updated Name")
    updated = await crud.update_item(db_session, sample_item, update)
    assert updated.name == "Updated Name"


async def test_delete_item(db_session: AsyncSession, sample_item):
    await crud.delete_item(db_session, sample_item)
    deleted = await crud.get_item_by_id(db_session, sample_item.id)
    assert deleted is None

Template: tests/{module}/fixtures.py (sample_item)

import pytest
from sqlalchemy.ext.asyncio import AsyncSession

from database.items.models import Item  # adjust to your project


@pytest.fixture
async def sample_item(db_session: AsyncSession, tenant_id: str) -> Item:
    item = Item(tenant_id=tenant_id, name="Fixture Item", price=100)
    db_session.add(item)
    await db_session.commit()
    await db_session.refresh(item)
    return item

Common pitfalls

  • Accidentally testing routing/auth here: CRUD tests should not call the API client.
  • Non-deterministic fixtures: prefer stable values so failures are readable.
  • Isolation drift: if your CRUD functions commit, ensure your DB fixture strategy still keeps tests isolated (see Fixtures).