Step 6: Testing

Write comprehensive tests for the Todo feature.

What You'll Build

• Integration tests for HTTP endpoints
• Service or use-case unit tests
• Celery task tests
• IoC override patterns for mocking

Files to Create

Action	File Path
Create	tests/integration/core/todo/delivery/fastapi/test_controllers.py
Create	tests/unit/core/todo/test_services.py

Concept Reference

See also: Override IoC in Tests guide for mocking techniques.

Understanding the Test Architecture

The project uses:

• pytest for test framework
• pytest-django for Django integration
• Function-scoped containers for test isolation
• Test factories for creating test data

Step 1: Create HTTP Integration Tests

Create tests/integration/core/todo/delivery/fastapi/test_controllers.py:

# tests/integration/core/todo/delivery/fastapi/test_controllers.py
from http import HTTPStatus

import pytest

from fastdjango.core.todo.models import Todo
from fastdjango.core.user.models import User
from tests.integration.factories import TestClientFactory, TestUserFactory


@pytest.fixture(scope="function")
def user(user_factory: TestUserFactory) -> User:
    """Create a test user."""
    return user_factory(
        username="testuser",
        password="SecurePassword123!",
        email="test@example.com",
    )


@pytest.fixture(scope="function")
def other_user(user_factory: TestUserFactory) -> User:
    """Create another test user for access control tests."""
    return user_factory(
        username="otheruser",
        password="SecurePassword123!",
        email="other@example.com",
    )


@pytest.fixture(scope="function")
def todo(user: User) -> Todo:
    """Create a test todo."""
    return Todo.objects.create(
        user=user,
        title="Test Todo",
        description="Test description",
    )


@pytest.mark.django_db(transaction=True)
class TestTodoController:
    """Tests for todo HTTP endpoints."""

    def test_list_todos_empty(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        """Test listing todos when user has none."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.get("/v1/todos")

        assert response.status_code == HTTPStatus.OK
        data = response.json()
        assert data["todos"] == []
        assert data["count"] == 0

    def test_list_todos_with_items(
        self,
        test_client_factory: TestClientFactory,
        user: User,
        todo: Todo,
    ) -> None:
        """Test listing todos returns user's items."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.get("/v1/todos")

        assert response.status_code == HTTPStatus.OK
        data = response.json()
        assert data["count"] == 1
        assert data["todos"][0]["title"] == "Test Todo"

    def test_list_todos_filter_completed(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        """Test filtering todos by completion status."""
        # Create completed and incomplete todos
        Todo.objects.create(user=user, title="Completed", completed=True)
        Todo.objects.create(user=user, title="Incomplete", completed=False)

        with test_client_factory(auth_for_user=user) as client:
            response = client.get("/v1/todos?completed=true")

        assert response.status_code == HTTPStatus.OK
        data = response.json()
        assert data["count"] == 1
        assert data["todos"][0]["title"] == "Completed"

    def test_create_todo(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        """Test creating a new todo."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.post(
                "/v1/todos",
                json={
                    "title": "New Todo",
                    "description": "New description",
                },
            )

        assert response.status_code == HTTPStatus.CREATED
        data = response.json()
        assert data["title"] == "New Todo"
        assert data["description"] == "New description"
        assert data["completed"] is False
        assert data["user_id"] == user.id

    def test_create_todo_minimal(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        """Test creating a todo with minimal fields."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.post(
                "/v1/todos",
                json={"title": "Minimal Todo"},
            )

        assert response.status_code == HTTPStatus.CREATED
        data = response.json()
        assert data["title"] == "Minimal Todo"
        assert data["description"] == ""

    def test_create_todo_validation_error(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        """Test creating a todo with invalid data."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.post(
                "/v1/todos",
                json={"title": ""},  # Empty title
            )

        assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY

    def test_get_todo(
        self,
        test_client_factory: TestClientFactory,
        user: User,
        todo: Todo,
    ) -> None:
        """Test getting a specific todo."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.get(f"/v1/todos/{todo.id}")

        assert response.status_code == HTTPStatus.OK
        data = response.json()
        assert data["id"] == todo.id
        assert data["title"] == "Test Todo"

    def test_get_todo_not_found(
        self,
        test_client_factory: TestClientFactory,
        user: User,
    ) -> None:
        """Test getting a non-existent todo."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.get("/v1/todos/99999")

        assert response.status_code == HTTPStatus.NOT_FOUND

    def test_get_todo_access_denied(
        self,
        test_client_factory: TestClientFactory,
        other_user: User,
        todo: Todo,
    ) -> None:
        """Test accessing another user's todo."""
        with test_client_factory(auth_for_user=other_user) as client:
            response = client.get(f"/v1/todos/{todo.id}")

        assert response.status_code == HTTPStatus.FORBIDDEN

    def test_update_todo(
        self,
        test_client_factory: TestClientFactory,
        user: User,
        todo: Todo,
    ) -> None:
        """Test updating a todo."""
        with test_client_factory(auth_for_user=user) as client:
            response = client.patch(
                f"/v1/todos/{todo.id}",
                json={
                    "title": "Updated Title",
                    "completed": True,
                },
            )

        assert response.status_code == HTTPStatus.OK
        data = response.json()
        assert data["title"] == "Updated Title"
        assert data["completed"] is True

    def test_update_todo_partial(
        self,
        test_client_factory: TestClientFactory,
        user: User,
        todo: Todo,
    ) -> None:
        """Test partial update of a todo."""
        original_title = todo.title

        with test_client_factory(auth_for_user=user) as client:
            response = client.patch(
                f"/v1/todos/{todo.id}",
                json={"completed": True},
            )

        assert response.status_code == HTTPStatus.OK
        data = response.json()
        assert data["title"] == original_title  # Unchanged
        assert data["completed"] is True

    def test_delete_todo(
        self,
        test_client_factory: TestClientFactory,
        user: User,
        todo: Todo,
    ) -> None:
        """Test deleting a todo."""
        todo_id = todo.id

        with test_client_factory(auth_for_user=user) as client:
            response = client.delete(f"/v1/todos/{todo_id}")

        assert response.status_code == HTTPStatus.NO_CONTENT
        assert not Todo.objects.filter(id=todo_id).exists()

    def test_unauthenticated_access(
        self,
        test_client_factory: TestClientFactory,
    ) -> None:
        """Test that unauthenticated requests are rejected."""
        with test_client_factory() as client:  # No auth
            response = client.get("/v1/todos")

        assert response.status_code == HTTPStatus.FORBIDDEN

Step 2: Create Service Unit Tests

Create tests/unit/core/todo/test_services.py:

# tests/unit/core/todo/test_services.py
import pytest

from fastdjango.core.todo.models import Todo
from fastdjango.core.todo.exceptions import (
    TodoAccessDeniedError,
    TodoNotFoundError,
)
from fastdjango.core.todo.services import (
    TodoService,
)
from fastdjango.core.user.models import User
from fastdjango.infrastructure.django.transactions import DjangoTransactionFactory


@pytest.fixture(scope="function")
def service() -> TodoService:
    """Create a TodoService instance."""
    return TodoService(_transaction_factory=DjangoTransactionFactory())


@pytest.fixture(scope="function")
def user(transactional_db: None) -> User:
    """Create a test user."""
    return User.objects.create_user(
        username="testuser",
        email="test@example.com",
        password="password",
    )


@pytest.fixture(scope="function")
def other_user(transactional_db: None) -> User:
    """Create another test user."""
    return User.objects.create_user(
        username="otheruser",
        email="other@example.com",
        password="password",
    )


@pytest.mark.anyio
@pytest.mark.django_db(transaction=True)
class TestTodoService:
    """Unit tests for TodoService."""

    async def test_create_todo(self, service: TodoService, user: User) -> None:
        """Test creating a todo."""
        todo = await service.create_todo(
            user=user,
            title="Test Todo",
            description="Test description",
        )

        assert todo.id is not None
        assert todo.title == "Test Todo"
        assert todo.description == "Test description"
        assert todo.completed is False
        assert todo.user_id == user.id

    async def test_get_todo_by_id(self, service: TodoService, user: User) -> None:
        """Test getting a todo by ID."""
        created = await service.create_todo(user=user, title="Test")
        retrieved = await service.get_todo_by_id(todo_id=created.id, user=user)

        assert retrieved.id == created.id

    async def test_get_todo_by_id_not_found(
        self,
        service: TodoService,
        user: User,
    ) -> None:
        """Test getting a non-existent todo."""
        with pytest.raises(TodoNotFoundError):
            await service.get_todo_by_id(todo_id=99999, user=user)

    async def test_get_todo_by_id_access_denied(
        self,
        service: TodoService,
        user: User,
        other_user: User,
    ) -> None:
        """Test accessing another user's todo."""
        todo = await service.create_todo(user=user, title="Private Todo")

        with pytest.raises(TodoAccessDeniedError):
            await service.get_todo_by_id(todo_id=todo.id, user=other_user)

    async def test_list_todos_for_user(
        self,
        service: TodoService,
        user: User,
    ) -> None:
        """Test listing todos for a user."""
        await service.create_todo(user=user, title="Todo 1")
        await service.create_todo(user=user, title="Todo 2")

        todos = await service.list_todos_for_user(user=user)

        assert len(todos) == 2

    async def test_list_todos_filter_completed(
        self,
        service: TodoService,
        user: User,
    ) -> None:
        """Test filtering todos by completion status."""
        await service.create_todo(user=user, title="Incomplete")
        completed = await service.create_todo(user=user, title="Completed")
        await service.mark_completed(todo_id=completed.id, user=user)

        incomplete_todos = await service.list_todos_for_user(user=user, completed=False)
        completed_todos = await service.list_todos_for_user(user=user, completed=True)

        assert len(incomplete_todos) == 1
        assert len(completed_todos) == 1

    async def test_update_todo(self, service: TodoService, user: User) -> None:
        """Test updating a todo."""
        todo = await service.create_todo(user=user, title="Original")

        updated = await service.update_todo(
            todo_id=todo.id,
            user=user,
            title="Updated",
            completed=True,
        )

        assert updated.title == "Updated"
        assert updated.completed is True

    async def test_delete_todo(self, service: TodoService, user: User) -> None:
        """Test deleting a todo."""
        todo = await service.create_todo(user=user, title="To Delete")

        await service.delete_todo(todo_id=todo.id, user=user)

        assert not await Todo.objects.filter(id=todo.id).aexists()

    async def test_mark_completed(self, service: TodoService, user: User) -> None:
        """Test marking a todo as completed."""
        todo = await service.create_todo(user=user, title="Test")

        result = await service.mark_completed(todo_id=todo.id, user=user)

        assert result.completed is True

    async def test_mark_incomplete(self, service: TodoService, user: User) -> None:
        """Test marking a todo as incomplete."""
        todo = await service.create_todo(user=user, title="Test")
        await service.mark_completed(todo_id=todo.id, user=user)

        result = await service.mark_incomplete(todo_id=todo.id, user=user)

        assert result.completed is False

    async def test_delete_completed_todos(
        self,
        service: TodoService,
        user: User,
    ) -> None:
        """Test deleting all completed todos."""
        await service.create_todo(user=user, title="Keep")
        completed1 = await service.create_todo(user=user, title="Delete 1")
        completed2 = await service.create_todo(user=user, title="Delete 2")
        await service.mark_completed(todo_id=completed1.id, user=user)
        await service.mark_completed(todo_id=completed2.id, user=user)

        deleted_count = await service.delete_completed_todos(user=user)

        assert deleted_count == 2
        assert await Todo.objects.filter(user=user).acount() == 1

Step 3: Create Celery Task Tests

Add task tests to tests/integration/core/todo/delivery/celery/test_todo_cleanup.py:

# tests/integration/core/todo/delivery/celery/test_todo_cleanup.py
import pytest

from fastdjango.core.todo.models import Todo
from fastdjango.core.user.models import User
from tests.integration.factories import (
    TestCeleryWorkerFactory,
    TestTasksRegistryFactory,
    TestUserFactory,
)


@pytest.fixture(scope="function")
def user(user_factory: TestUserFactory) -> User:
    """Create a test user."""
    return user_factory(username="testuser", password="password")


@pytest.mark.django_db(transaction=True)
class TestTodoCleanupTask:
    """Tests for todo cleanup Celery task."""

    def test_cleanup_completed_todos(
        self,
        celery_worker_factory: TestCeleryWorkerFactory,
        tasks_registry_factory: TestTasksRegistryFactory,
        user: User,
    ) -> None:
        """Test that cleanup task deletes completed todos."""
        # Create test data
        Todo.objects.create(user=user, title="Keep", completed=False)
        Todo.objects.create(user=user, title="Delete", completed=True)

        registry = tasks_registry_factory()

        with celery_worker_factory():
            result = registry.todo_cleanup.delay().get(timeout=10)

        assert result["todos_deleted"] == 1
        assert Todo.objects.filter(user=user).count() == 1

Running Tests

# Run all tests
make test

# Run with coverage report
make test

# Run specific test file
uv run pytest tests/integration/core/todo/delivery/fastapi/test_controllers.py

# Run with verbose output
uv run pytest -v tests/

# Run specific test class
uv run pytest tests/integration/core/todo/delivery/fastapi/test_controllers.py::TestTodoController

# Run specific test method
uv run pytest tests/integration/core/todo/delivery/fastapi/test_controllers.py::TestTodoController::test_create_todo

Test Patterns

Using Test Factories

# TestClientFactory - HTTP testing with optional auth
with test_client_factory(auth_for_user=user) as client:
    response = client.get("/v1/todos")

# TestUserFactory - Create test users
user = user_factory(username="test", password="pass")

# TestCeleryWorkerFactory - Run Celery workers
with celery_worker_factory():
    result = registry.my_task.delay().get(timeout=10)

Per-Test Isolation

Each test gets a fresh container. Fixtures are function-scoped by default:

@pytest.fixture(scope="function")
def container() -> Container:
    return get_container()

Transaction Rollback

Use @pytest.mark.django_db(transaction=True) for database tests:

@pytest.mark.django_db(transaction=True)
class TestMyFeature:
    def test_something(self) -> None:
        # Database changes are rolled back after test
        ...

Summary

You've learned:

• Integration testing HTTP endpoints with TestClientFactory
• Unit testing services or use cases directly
• Testing Celery tasks with TestCeleryWorkerFactory
• Test isolation patterns and fixtures

Congratulations!

You've completed the tutorial! You now know how to:

1. Create Django models with proper relationships
2. Build services that encapsulate business logic
3. Wire dependencies with the IoC container
4. Create HTTP endpoints with authentication
5. Add background task processing
6. Configure observability
7. Write comprehensive tests

Next Steps

• Concepts - Deeper understanding of the architecture
• How-To Guides - Solve specific problems
• Reference - Complete configuration details