diff --git a/PLANNING.md b/PLANNING.md new file mode 100644 index 0000000..d654d25 --- /dev/null +++ b/PLANNING.md @@ -0,0 +1,153 @@ +# Todo Application – Architecture Plan + +## Technology Stack + +| Layer | Technology | +|----------------|----------------------------------------------| +| Web framework | FastAPI 0.110+ | +| ORM | SQLAlchemy 2.0+ | +| Validation | Pydantic v2 | +| Database | SQLite in-memory (`sqlite://`) | +| ASGI server | uvicorn | +| Testing | pytest + httpx + fastapi.testclient | + +> **Note:** The SQLite in-memory database is ephemeral — all data is lost +> when the process restarts. + +## Project Structure + +``` +. +├── app/ +│ ├── __init__.py +│ ├── main.py # FastAPI application factory & lifespan +│ ├── database.py # Engine, SessionLocal, Base, get_db +│ ├── models.py # SQLAlchemy ORM models +│ ├── schemas.py # Pydantic request/response schemas +│ └── routers/ +│ ├── __init__.py +│ └── todos.py # /todos CRUD endpoints +├── tests/ +│ ├── __init__.py +│ ├── test_database.py +│ ├── test_models.py +│ ├── test_main_startup.py +│ └── test_todos.py +├── requirements.txt +├── Dockerfile +├── docker-compose.yml +├── PLANNING.md # ← you are here +└── RUNNING.md +``` + +## Database Layer + +The database module (`app/database.py`) exposes: + +- **`engine`** – `create_engine("sqlite://", connect_args={"check_same_thread": False})` +- **`SessionLocal`** – `sessionmaker(autocommit=False, autoflush=False, bind=engine)` +- **`Base`** – `declarative_base()` used by all models +- **`get_db()`** – generator dependency for FastAPI `Depends(get_db)` + +`connect_args={"check_same_thread": False}` is **required** because SQLite +by default only allows the creating thread to access the connection, but +FastAPI/uvicorn may serve requests from a different thread. + +## Data Model + +### `todos` table (SQLAlchemy model: `Todo`) + +| Column | Type | Constraints | +|-------------|---------------|------------------------------------------| +| id | Integer | PRIMARY KEY, AUTOINCREMENT | +| title | String(255) | NOT NULL | +| description | String(1024) | NULLABLE | +| completed | Boolean | NOT NULL, DEFAULT False | +| created_at | DateTime | NOT NULL, DEFAULT `datetime.utcnow` | + +> `created_at` is **server-set** and is never accepted from client input. + +## Pydantic Schemas + +### `TodoCreate` (POST body) + +| Field | Type | Constraints | +|-------------|-----------------|------------------------------| +| title | str | required, min_length=1 | +| description | Optional[str] | default None | +| completed | bool | default False | + +> Title must have a **minimum length of 1** to prevent empty-string titles. + +### `TodoUpdate` (PUT body) + +| Field | Type | Constraints | +|-------------|-----------------|------------------------------| +| title | Optional[str] | min_length=1 if provided | +| description | Optional[str] | default None | +| completed | Optional[bool] | default None | + +> Uses **partial update semantics**: only provided (non-None) fields are +> written; fields not supplied in the JSON body remain unchanged. + +### `TodoResponse` (all responses) + +| Field | Type | +|-------------|-----------------| +| id | int | +| title | str | +| description | Optional[str] | +| completed | bool | +| created_at | datetime | + +Config: `from_attributes = True` (Pydantic v2 orm_mode equivalent). + +## API Endpoints + +| Method | Path | Status | Description | +|--------|-----------------|--------|------------------------| +| GET | `/todos` | 200 | List all todos | +| GET | `/todos/{id}` | 200 | Get one todo | +| POST | `/todos` | 201 | Create a new todo | +| PUT | `/todos/{id}` | 200 | Update an existing todo| +| DELETE | `/todos/{id}` | 204 | Delete a todo | + +## Router Functions + +Each endpoint handler receives the DB session via +`db: Session = Depends(get_db)` and delegates to straight SQLAlchemy +queries (no repository layer needed at this scale). + +## Application Startup + +```python +@asynccontextmanager +async def lifespan(app: FastAPI): + Base.metadata.create_all(bind=engine) + yield +``` + +This ensures all tables are created when the process starts. + +## Error Handling + +Missing resources raise: + +```python +raise HTTPException(status_code=404, detail="Todo not found") +``` + +Validation errors are handled automatically by FastAPI / Pydantic (422). + +## Testing Strategy + +- Use `fastapi.testclient.TestClient` for synchronous integration tests. +- Each test file recreates tables via `Base.metadata.create_all` / + `Base.metadata.drop_all` in an `autouse` fixture. +- Unit tests verify column definitions, defaults, and constraints. +- Integration tests hit every endpoint and cover 404 / validation paths. + +## Docker Configuration + +A `Dockerfile` builds a slim Python image; `docker-compose.yml` maps +port 8000 and runs `uvicorn app.main:app --host 0.0.0.0 --port 8000`. diff --git a/RUNNING.md b/RUNNING.md index 77896cf..c33ca08 100644 --- a/RUNNING.md +++ b/RUNNING.md @@ -3,31 +3,80 @@ ## Prerequisites - Python 3.10 or later +- pip (Python package manager) +- (Optional) Docker & Docker Compose -## Install dependencies +--- + +## Option 1 — Local Setup + +### 1. Install Dependencies + +```bash +pip install -r requirements.txt +``` + +### 2. Run the Server + +```bash +uvicorn app.main:app --reload +``` + +The API is available at **http://localhost:8000**. + +Interactive API documentation (Swagger UI) is at **http://localhost:8000/docs**. + +Alternative ReDoc documentation is at **http://localhost:8000/redoc**. + +### 3. Seed Sample Data + +To populate the database with sample todo items for quick testing: ```bash -pip install fastapi uvicorn pydantic +python seed_data.py ``` -For running the test suite you will also need: +This inserts several example todos so you can immediately exercise the +API without manually creating items first. Run it **after** the server +has started at least once (so the database tables exist), or the script +will create the tables itself. + +### 4. Run Tests ```bash -pip install httpx pytest +pytest -v ``` -## Start the server +--- + +## Option 2 — Docker + +### 1. Build & Start ```bash -uvicorn main:app --reload --host 0.0.0.0 --port 8000 +docker compose up --build ``` -The API will be available at . +The API is available at **http://localhost:8000**. -Interactive docs are served at . +Interactive API documentation (Swagger UI) is at **http://localhost:8000/docs**. -## Run the tests +### 2. Run Tests (inside the container) ```bash -pytest tests/ +docker compose exec app pytest -v ``` + +--- + +## Notes + +- **No authentication** is required — this is a public Todo API. +- The database is **SQLite in-memory** — all data is lost when the + process restarts. +- CORS is configured to allow all origins (`*`) for development + convenience. +- The seed script connects to the same in-memory database **only when + run inside the same process** (e.g., during tests). When running + against a live server, use the API endpoints or switch to a + file-based SQLite URL. diff --git a/app/__init__.py b/app/__init__.py new file mode 100644 index 0000000..a210d9f --- /dev/null +++ b/app/__init__.py @@ -0,0 +1,6 @@ +"""Todo application package. + +This package contains the FastAPI application, database layer, +Pydantic schemas, SQLAlchemy models, and API routers for the +Todo REST API. +""" diff --git a/app/crud.py b/app/crud.py new file mode 100644 index 0000000..594f964 --- /dev/null +++ b/app/crud.py @@ -0,0 +1,124 @@ +"""CRUD operations for the Todo model. + +Provides database-layer functions for creating, reading, updating, and +deleting Todo items. Each function accepts a SQLAlchemy ``Session`` as +its first argument so that callers (typically FastAPI route handlers) +can inject the session via dependency injection. + +Functions +--------- +- ``get_todos`` – Retrieve a paginated list of todos. +- ``get_todo`` – Retrieve a single todo by primary key. +- ``create_todo`` – Persist a new todo from a ``TodoCreate`` schema. +- ``update_todo`` – Partially update an existing todo from a ``TodoUpdate`` schema. +- ``delete_todo`` – Remove a todo by primary key. +""" + +from __future__ import annotations + +from typing import List, Optional + +from sqlalchemy.orm import Session + +from app.models import Todo +from app.schemas import TodoCreate, TodoUpdate + + +def get_todos(db: Session, skip: int = 0, limit: int = 100) -> List[Todo]: + """Return a paginated list of Todo items. + + Args: + db: Active SQLAlchemy database session. + skip: Number of rows to skip (offset). Defaults to ``0``. + limit: Maximum number of rows to return. Defaults to ``100``. + + Returns: + A list of :class:`~app.models.Todo` instances ordered by ``id``. + """ + return db.query(Todo).offset(skip).limit(limit).all() + + +def get_todo(db: Session, todo_id: int) -> Optional[Todo]: + """Return a single Todo by its primary key, or ``None`` if not found. + + Args: + db: Active SQLAlchemy database session. + todo_id: The integer primary key of the desired todo. + + Returns: + The matching :class:`~app.models.Todo` instance, or ``None``. + """ + return db.query(Todo).filter(Todo.id == todo_id).first() + + +def create_todo(db: Session, todo: TodoCreate) -> Todo: + """Create and persist a new Todo item. + + The ``completed`` flag defaults to ``False`` and ``created_at`` is + set automatically by the database column default. + + Args: + db: Active SQLAlchemy database session. + todo: A validated :class:`~app.schemas.TodoCreate` instance. + + Returns: + The newly created :class:`~app.models.Todo` with server-set + fields (``id``, ``created_at``) populated. + """ + db_todo = Todo( + title=todo.title, + description=todo.description, + ) + db.add(db_todo) + db.commit() + db.refresh(db_todo) + return db_todo + + +def update_todo(db: Session, todo_id: int, todo: TodoUpdate) -> Optional[Todo]: + """Partially update an existing Todo item. + + Only fields that are explicitly provided (not ``None``) in *todo* + are written to the database. Fields omitted by the client remain + unchanged. + + Args: + db: Active SQLAlchemy database session. + todo_id: The integer primary key of the todo to update. + todo: A validated :class:`~app.schemas.TodoUpdate` instance + containing the fields to change. + + Returns: + The updated :class:`~app.models.Todo` instance, or ``None`` if + no todo with the given *todo_id* exists. + """ + db_todo = db.query(Todo).filter(Todo.id == todo_id).first() + if db_todo is None: + return None + + update_data = todo.model_dump(exclude_unset=True) + for field, value in update_data.items(): + setattr(db_todo, field, value) + + db.commit() + db.refresh(db_todo) + return db_todo + + +def delete_todo(db: Session, todo_id: int) -> bool: + """Delete a Todo item by primary key. + + Args: + db: Active SQLAlchemy database session. + todo_id: The integer primary key of the todo to delete. + + Returns: + ``True`` if the todo existed and was deleted, ``False`` otherwise. + """ + db_todo = db.query(Todo).filter(Todo.id == todo_id).first() + if db_todo is None: + return False + + db.delete(db_todo) + db.commit() + return True diff --git a/app/database.py b/app/database.py new file mode 100644 index 0000000..ffb0a7d --- /dev/null +++ b/app/database.py @@ -0,0 +1,50 @@ +"""SQLite in-memory database configuration using SQLAlchemy. + +Provides the database engine, session factory, declarative base, and +a FastAPI-compatible dependency for obtaining database sessions. + +- Engine uses ``sqlite://`` (in-memory) with ``check_same_thread=False`` + so that the same connection can be reused across ASGI threads. +- ``SessionLocal`` is a ``sessionmaker`` bound to the engine. +- ``Base`` is the declarative base for all ORM models. +- ``get_db`` is a generator dependency that yields a session and + ensures it is closed after the request completes. +""" + +from __future__ import annotations + +from typing import Generator + +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, declarative_base, sessionmaker + +SQLALCHEMY_DATABASE_URL: str = "sqlite://" + +engine = create_engine( + SQLALCHEMY_DATABASE_URL, + connect_args={"check_same_thread": False}, +) + +SessionLocal: sessionmaker[Session] = sessionmaker( + autocommit=False, + autoflush=False, + bind=engine, +) + +Base = declarative_base() + + +def get_db() -> Generator[Session, None, None]: + """FastAPI dependency that provides a SQLAlchemy database session. + + Yields a ``Session`` instance and guarantees it is closed when the + request finishes, regardless of whether an exception occurred. + + Yields: + Session: A SQLAlchemy database session. + """ + db = SessionLocal() + try: + yield db + finally: + db.close() diff --git a/app/main.py b/app/main.py new file mode 100644 index 0000000..73ff2c0 --- /dev/null +++ b/app/main.py @@ -0,0 +1,58 @@ +"""FastAPI application entry point. + +Initialises the FastAPI app with CORS middleware and includes +the Todo API router. This module is the single source of the +``app`` instance used by uvicorn. + +On startup the SQLAlchemy ``Base.metadata.create_all`` call ensures +that all ORM tables exist in the (in-memory) SQLite database. +""" + +from __future__ import annotations + +from contextlib import asynccontextmanager +from typing import AsyncIterator + +from fastapi import FastAPI +from fastapi.middleware.cors import CORSMiddleware + +from app.database import Base, engine +from app.models import Todo # noqa: F401 – ensure model is registered on Base +from app.routers import router + + +@asynccontextmanager +async def lifespan(_app: FastAPI) -> AsyncIterator[None]: + """Application lifespan handler – create DB tables on startup.""" + Base.metadata.create_all(bind=engine) + yield + + +app = FastAPI( + title="Todo API", + description="A simple Todo REST API backed by SQLite (in-memory).", + version="1.0.0", + lifespan=lifespan, +) + +# --------------------------------------------------------------------------- +# CORS Middleware – allow all origins during development. +# --------------------------------------------------------------------------- +app.add_middleware( + CORSMiddleware, + allow_origins=["*"], + allow_credentials=True, + allow_methods=["*"], + allow_headers=["*"], +) + +# --------------------------------------------------------------------------- +# Include API routers +# --------------------------------------------------------------------------- +app.include_router(router) + + +@app.get("/", tags=["root"]) +async def root() -> dict: + """Return a welcome message at the API root.""" + return {"message": "Welcome to the Todo API"} diff --git a/app/models.py b/app/models.py new file mode 100644 index 0000000..974a987 --- /dev/null +++ b/app/models.py @@ -0,0 +1,46 @@ +"""SQLAlchemy ORM models for the Todo application. + +Defines the ``Todo`` model mapped to the ``todos`` table with the +following columns: + +- **id** – Integer primary key with autoincrement. +- **title** – Non-nullable string (max 255 characters). +- **description** – Nullable string (max 1024 characters). +- **completed** – Boolean defaulting to ``False``. +- **created_at** – DateTime defaulting to ``datetime.utcnow``. +""" + +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import Boolean, Column, DateTime, Integer, String + +from app.database import Base + + +class Todo(Base): + """ORM model representing a single Todo item. + + Attributes: + id: Unique auto-incrementing identifier. + title: Short title of the todo (required). + description: Optional longer description. + completed: Whether the todo has been completed. + created_at: Timestamp of when the todo was created (server-set). + """ + + __tablename__ = "todos" + + id: int = Column(Integer, primary_key=True, autoincrement=True) + title: str = Column(String(255), nullable=False) + description: str | None = Column(String(1024), nullable=True) + completed: bool = Column(Boolean, default=False, nullable=False) + created_at: datetime = Column(DateTime, default=datetime.utcnow, nullable=False) + + def __repr__(self) -> str: + """Return a developer-friendly string representation.""" + return ( + f"" + ) diff --git a/app/routers/__init__.py b/app/routers/__init__.py new file mode 100644 index 0000000..e02a067 --- /dev/null +++ b/app/routers/__init__.py @@ -0,0 +1,18 @@ +"""API routers package. + +Re-exports the top-level ``router`` so that ``app.main`` can simply do:: + + from app.routers import router + +All concrete endpoint modules are included into the ``router`` defined +here. +""" + +from __future__ import annotations + +from fastapi import APIRouter + +from app.routes import router as todos_router + +router = APIRouter() +router.include_router(todos_router) diff --git a/app/routes.py b/app/routes.py new file mode 100644 index 0000000..712efa6 --- /dev/null +++ b/app/routes.py @@ -0,0 +1,170 @@ +"""API route endpoints for the Todo resource. + +Defines an ``APIRouter`` with the following endpoints: + +- ``GET /todos`` – List all todos with optional ``skip``/``limit`` pagination. +- ``GET /todos/{todo_id}`` – Retrieve a single todo by ID (404 if missing). +- ``POST /todos`` – Create a new todo (returns 201). +- ``PUT /todos/{todo_id}`` – Update an existing todo (404 if missing). +- ``DELETE /todos/{todo_id}`` – Delete a todo (returns 204, 404 if missing). + +All endpoints use ``Depends(get_db)`` for database session injection. +""" + +from __future__ import annotations + +from typing import List + +from fastapi import APIRouter, Depends, HTTPException, status +from sqlalchemy.orm import Session + +from app.crud import create_todo, delete_todo, get_todo, get_todos, update_todo +from app.database import get_db +from app.schemas import TodoCreate, TodoResponse, TodoUpdate + +router = APIRouter( + prefix="/todos", + tags=["todos"], +) + + +@router.get( + "", + response_model=List[TodoResponse], + status_code=status.HTTP_200_OK, + summary="List all todos", +) +def list_todos( + skip: int = 0, + limit: int = 100, + db: Session = Depends(get_db), +) -> List[TodoResponse]: + """Return a paginated list of todo items. + + Args: + skip: Number of records to skip (offset). Defaults to ``0``. + limit: Maximum number of records to return. Defaults to ``100``. + db: SQLAlchemy database session (injected). + + Returns: + A list of :class:`~app.schemas.TodoResponse` objects. + """ + return get_todos(db, skip=skip, limit=limit) + + +@router.get( + "/{todo_id}", + response_model=TodoResponse, + status_code=status.HTTP_200_OK, + summary="Get a single todo", +) +def read_todo( + todo_id: int, + db: Session = Depends(get_db), +) -> TodoResponse: + """Retrieve a single todo item by its primary key. + + Args: + todo_id: The integer ID of the desired todo. + db: SQLAlchemy database session (injected). + + Returns: + The matching :class:`~app.schemas.TodoResponse`. + + Raises: + HTTPException: 404 if no todo with the given ID exists. + """ + db_todo = get_todo(db, todo_id=todo_id) + if db_todo is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Todo not found", + ) + return db_todo + + +@router.post( + "", + response_model=TodoResponse, + status_code=status.HTTP_201_CREATED, + summary="Create a new todo", +) +def create_todo_endpoint( + todo: TodoCreate, + db: Session = Depends(get_db), +) -> TodoResponse: + """Create and persist a new todo item. + + The ``id`` and ``created_at`` fields are set automatically by the + server and returned in the response. + + Args: + todo: Validated :class:`~app.schemas.TodoCreate` request body. + db: SQLAlchemy database session (injected). + + Returns: + The newly created :class:`~app.schemas.TodoResponse`. + """ + return create_todo(db, todo=todo) + + +@router.put( + "/{todo_id}", + response_model=TodoResponse, + status_code=status.HTTP_200_OK, + summary="Update an existing todo", +) +def update_todo_endpoint( + todo_id: int, + todo: TodoUpdate, + db: Session = Depends(get_db), +) -> TodoResponse: + """Partially update an existing todo item. + + Only the fields provided in the request body are modified; omitted + fields remain unchanged. + + Args: + todo_id: The integer ID of the todo to update. + todo: Validated :class:`~app.schemas.TodoUpdate` request body. + db: SQLAlchemy database session (injected). + + Returns: + The updated :class:`~app.schemas.TodoResponse`. + + Raises: + HTTPException: 404 if no todo with the given ID exists. + """ + db_todo = update_todo(db, todo_id=todo_id, todo=todo) + if db_todo is None: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Todo not found", + ) + return db_todo + + +@router.delete( + "/{todo_id}", + status_code=status.HTTP_204_NO_CONTENT, + summary="Delete a todo", +) +def delete_todo_endpoint( + todo_id: int, + db: Session = Depends(get_db), +) -> None: + """Delete a todo item by its primary key. + + Args: + todo_id: The integer ID of the todo to delete. + db: SQLAlchemy database session (injected). + + Raises: + HTTPException: 404 if no todo with the given ID exists. + """ + deleted = delete_todo(db, todo_id=todo_id) + if not deleted: + raise HTTPException( + status_code=status.HTTP_404_NOT_FOUND, + detail="Todo not found", + ) diff --git a/app/schemas.py b/app/schemas.py new file mode 100644 index 0000000..f695817 --- /dev/null +++ b/app/schemas.py @@ -0,0 +1,100 @@ +"""Pydantic schemas for the Todo API. + +Defines request and response validation models: + +- **TodoBase** – Shared fields for creation (title, optional description). +- **TodoCreate** – Inherits TodoBase; used for ``POST /todos`` requests. +- **TodoUpdate** – All-optional fields for partial updates via ``PUT``. +- **TodoResponse** – Full representation returned to clients, including + server-set fields (``id``, ``completed``, ``created_at``). Configured + with ``orm_mode = True`` so it can be constructed directly from + SQLAlchemy model instances. +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Optional + +from pydantic import BaseModel, Field + + +class TodoBase(BaseModel): + """Base schema containing the common writable Todo fields. + + Attributes: + title: Short title of the todo item (min 1 character). + description: Optional longer description of the todo. + """ + + title: str = Field( + ..., + min_length=1, + max_length=255, + description="Title of the todo item", + ) + description: Optional[str] = Field( + None, + max_length=1024, + description="Optional longer description of the todo", + ) + + +class TodoCreate(TodoBase): + """Schema for creating a new Todo item. + + Inherits ``title`` and ``description`` from :class:`TodoBase`. + No additional fields are required — ``completed`` and ``created_at`` + are set by the server. + """ + + +class TodoUpdate(BaseModel): + """Schema for partially updating an existing Todo item. + + Every field is optional so that clients may send only the fields + they wish to change. Fields set to ``None`` (or omitted) are + left unchanged on the server side. + + Attributes: + title: New title (min 1 character if provided). + description: New description, or ``None`` to leave unchanged. + completed: New completion status, or ``None`` to leave unchanged. + """ + + title: Optional[str] = Field( + None, + min_length=1, + max_length=255, + description="Updated title of the todo item", + ) + description: Optional[str] = Field( + None, + max_length=1024, + description="Updated description of the todo item", + ) + completed: Optional[bool] = Field( + None, + description="Updated completion status", + ) + + +class TodoResponse(TodoBase): + """Schema returned to clients representing a persisted Todo item. + + Includes all base fields plus the server-managed attributes. + + Attributes: + id: Unique auto-incrementing identifier. + completed: Whether the todo has been completed. + created_at: Timestamp of when the todo was created (UTC). + """ + + id: int + completed: bool + created_at: datetime + + class Config: + """Pydantic model configuration.""" + + orm_mode = True diff --git a/requirements.txt b/requirements.txt index 5a6ebf7..daae002 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,5 +1,6 @@ -fastapi>=0.100.0 +fastapi>=0.110.0 uvicorn>=0.23.0 +sqlalchemy>=2.0.0 pydantic>=2.0.0 pytest>=7.0.0 pytest-timeout>=2.1.0 diff --git a/seed_data.py b/seed_data.py new file mode 100644 index 0000000..6a8ee60 --- /dev/null +++ b/seed_data.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""Seed the database with sample Todo items for quick testing. + +Usage +----- +:: + + python seed_data.py + +The script creates the database tables (if they don't already exist) +and inserts a handful of representative todos covering various +combinations of title, description, and completion status. + +Because the default configuration uses an **in-memory** SQLite database, +this script is most useful when imported programmatically (e.g. from +tests) via :func:`seed` rather than run as a standalone process against +a live server. To persist seeded data to a running server, either +change ``SQLALCHEMY_DATABASE_URL`` in ``app/database.py`` to a +file-based SQLite path or call the ``POST /todos`` endpoint directly. +""" + +from __future__ import annotations + +from typing import List + +from sqlalchemy.orm import Session + +from app.database import Base, SessionLocal, engine +from app.models import Todo + +# --------------------------------------------------------------------------- +# Sample data +# --------------------------------------------------------------------------- + +SAMPLE_TODOS: List[dict] = [ + { + "title": "Buy groceries", + "description": "Milk, eggs, bread, and fresh vegetables from the farmers market.", + "completed": False, + }, + { + "title": "Read a book", + "description": "Finish reading 'Designing Data-Intensive Applications' by Martin Kleppmann.", + "completed": False, + }, + { + "title": "Morning jog", + "description": "Run 5 km around the park before breakfast.", + "completed": True, + }, + { + "title": "Write unit tests", + "description": "Add pytest coverage for the new CRUD endpoints.", + "completed": False, + }, + { + "title": "Clean the kitchen", + "description": None, + "completed": True, + }, + { + "title": "Plan weekend trip", + "description": "Research destinations, book accommodation, and create a packing list.", + "completed": False, + }, + { + "title": "Update resume", + "description": "Add recent project experience and refresh the skills section.", + "completed": False, + }, +] + + +def seed(db: Session) -> List[Todo]: + """Insert sample todos into the database. + + Args: + db: An active SQLAlchemy database session. + + Returns: + A list of the newly created :class:`~app.models.Todo` instances + with server-set fields (``id``, ``created_at``) populated. + """ + created: List[Todo] = [] + for item in SAMPLE_TODOS: + todo = Todo( + title=item["title"], + description=item["description"], + completed=item["completed"], + ) + db.add(todo) + created.append(todo) + + db.commit() + for todo in created: + db.refresh(todo) + + return created + + +def main() -> None: + """Entry point: create tables and seed sample data.""" + Base.metadata.create_all(bind=engine) + + db = SessionLocal() + try: + todos = seed(db) + print(f"Seeded {len(todos)} sample todos:") + for todo in todos: + status = "✓" if todo.completed else "✗" + print(f" [{status}] {todo.id}: {todo.title}") + finally: + db.close() + + +if __name__ == "__main__": + main() diff --git a/tests/__init__.py b/tests/__init__.py index e69de29..269b97f 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -0,0 +1 @@ +"""Test package for the Todo application.""" diff --git a/tests/test_app_setup.py b/tests/test_app_setup.py new file mode 100644 index 0000000..de259f9 --- /dev/null +++ b/tests/test_app_setup.py @@ -0,0 +1,102 @@ +"""Tests for the application setup and scaffolding. + +Verifies that the FastAPI app initialises correctly, CORS middleware +is configured, the router is included, and the root endpoint responds. +""" + +from __future__ import annotations + +import pytest +from fastapi.testclient import TestClient +from starlette.middleware.cors import CORSMiddleware + +from app.main import app + + +@pytest.fixture() +def client() -> TestClient: + """Return a TestClient bound to the application.""" + return TestClient(app) + + +class TestAppInitialisation: + """Verify basic application configuration.""" + + def test_app_title(self) -> None: + """App title should be set.""" + assert app.title == "Todo API" + + def test_app_version(self) -> None: + """App version should be set.""" + assert app.version == "1.0.0" + + def test_cors_middleware_present(self) -> None: + """CORSMiddleware must be registered.""" + middleware_classes = [ + type(m) for m in getattr(app, "user_middleware", []) + ] + # FastAPI stores user_middleware as Middleware objects + # We check the cls attribute instead + middleware_cls_names = [] + for m in app.user_middleware: + middleware_cls_names.append(m.cls.__name__) + assert "CORSMiddleware" in middleware_cls_names + + +class TestRootEndpoint: + """Verify the root (/) endpoint.""" + + def test_root_returns_200(self, client: TestClient) -> None: + """GET / should return 200 OK.""" + response = client.get("/") + assert response.status_code == 200 + + def test_root_returns_welcome_message(self, client: TestClient) -> None: + """GET / should return a JSON welcome message.""" + response = client.get("/") + data = response.json() + assert "message" in data + assert "Welcome" in data["message"] + + +class TestCORSHeaders: + """Verify that CORS headers are present on responses.""" + + def test_cors_allows_origin(self, client: TestClient) -> None: + """Response should include Access-Control-Allow-Origin header.""" + response = client.get( + "/", + headers={"Origin": "http://example.com"}, + ) + assert response.status_code == 200 + assert "access-control-allow-origin" in response.headers + assert response.headers["access-control-allow-origin"] == "*" + + def test_cors_preflight(self, client: TestClient) -> None: + """OPTIONS preflight should succeed with CORS headers.""" + response = client.options( + "/", + headers={ + "Origin": "http://example.com", + "Access-Control-Request-Method": "GET", + }, + ) + assert response.status_code == 200 + assert "access-control-allow-origin" in response.headers + + +class TestOpenAPISchema: + """Verify that OpenAPI schema is generated correctly.""" + + def test_openapi_json(self, client: TestClient) -> None: + """GET /openapi.json should return the schema.""" + response = client.get("/openapi.json") + assert response.status_code == 200 + schema = response.json() + assert schema["info"]["title"] == "Todo API" + assert schema["info"]["version"] == "1.0.0" + + def test_docs_endpoint(self, client: TestClient) -> None: + """GET /docs should return the Swagger UI page.""" + response = client.get("/docs") + assert response.status_code == 200 diff --git a/tests/test_crud.py b/tests/test_crud.py new file mode 100644 index 0000000..cf706fe --- /dev/null +++ b/tests/test_crud.py @@ -0,0 +1,261 @@ +"""Tests for app.crud CRUD operations. + +Uses a dedicated in-memory SQLite database for each test function so +that tests are fully isolated and repeatable. +""" + +from __future__ import annotations + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from app.database import Base +from app.crud import create_todo, delete_todo, get_todo, get_todos, update_todo +from app.models import Todo # noqa: F401 – register model on Base +from app.schemas import TodoCreate, TodoUpdate + + +@pytest.fixture() +def db() -> Session: # type: ignore[misc] + """Yield a fresh SQLAlchemy session backed by an in-memory SQLite DB.""" + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + ) + Base.metadata.create_all(bind=engine) + testing_session_local = sessionmaker( + autocommit=False, + autoflush=False, + bind=engine, + ) + session = testing_session_local() + try: + yield session + finally: + session.close() + Base.metadata.drop_all(bind=engine) + + +# --------------------------------------------------------------------------- +# get_todos +# --------------------------------------------------------------------------- + + +class TestGetTodos: + """Tests for the get_todos function.""" + + def test_returns_empty_list_when_no_todos(self, db: Session) -> None: + """An empty database should yield an empty list.""" + result = get_todos(db) + assert result == [] + + def test_returns_all_todos(self, db: Session) -> None: + """All persisted todos are returned when no pagination is applied.""" + create_todo(db, TodoCreate(title="First")) + create_todo(db, TodoCreate(title="Second")) + create_todo(db, TodoCreate(title="Third")) + + result = get_todos(db) + assert len(result) == 3 + + def test_skip_parameter(self, db: Session) -> None: + """The skip parameter should offset returned rows.""" + for i in range(5): + create_todo(db, TodoCreate(title=f"Todo {i}")) + + result = get_todos(db, skip=3) + assert len(result) == 2 + + def test_limit_parameter(self, db: Session) -> None: + """The limit parameter should cap the number of returned rows.""" + for i in range(5): + create_todo(db, TodoCreate(title=f"Todo {i}")) + + result = get_todos(db, limit=2) + assert len(result) == 2 + + def test_skip_and_limit_combined(self, db: Session) -> None: + """Skip and limit work together for proper pagination.""" + for i in range(10): + create_todo(db, TodoCreate(title=f"Todo {i}")) + + result = get_todos(db, skip=2, limit=3) + assert len(result) == 3 + + +# --------------------------------------------------------------------------- +# get_todo +# --------------------------------------------------------------------------- + + +class TestGetTodo: + """Tests for the get_todo function.""" + + def test_returns_existing_todo(self, db: Session) -> None: + """A valid id should return the corresponding Todo.""" + created = create_todo(db, TodoCreate(title="My Todo", description="Details")) + result = get_todo(db, created.id) + + assert result is not None + assert result.id == created.id + assert result.title == "My Todo" + assert result.description == "Details" + assert result.completed is False + + def test_returns_none_for_missing_id(self, db: Session) -> None: + """A non-existent id should return None.""" + result = get_todo(db, 999) + assert result is None + + +# --------------------------------------------------------------------------- +# create_todo +# --------------------------------------------------------------------------- + + +class TestCreateTodo: + """Tests for the create_todo function.""" + + def test_creates_todo_with_title_only(self, db: Session) -> None: + """A todo can be created with just a title.""" + todo = create_todo(db, TodoCreate(title="Buy groceries")) + + assert todo.id is not None + assert todo.title == "Buy groceries" + assert todo.description is None + assert todo.completed is False + assert todo.created_at is not None + + def test_creates_todo_with_description(self, db: Session) -> None: + """A todo can be created with both title and description.""" + todo = create_todo( + db, + TodoCreate(title="Read book", description="Chapter 3"), + ) + + assert todo.title == "Read book" + assert todo.description == "Chapter 3" + + def test_created_todo_is_persisted(self, db: Session) -> None: + """The created todo should be retrievable from the DB.""" + todo = create_todo(db, TodoCreate(title="Persisted")) + fetched = get_todo(db, todo.id) + + assert fetched is not None + assert fetched.title == "Persisted" + + def test_auto_increment_ids(self, db: Session) -> None: + """Successive creates should yield incrementing ids.""" + first = create_todo(db, TodoCreate(title="First")) + second = create_todo(db, TodoCreate(title="Second")) + + assert second.id > first.id + + +# --------------------------------------------------------------------------- +# update_todo +# --------------------------------------------------------------------------- + + +class TestUpdateTodo: + """Tests for the update_todo function.""" + + def test_update_title(self, db: Session) -> None: + """Updating the title field should persist the change.""" + todo = create_todo(db, TodoCreate(title="Old Title")) + updated = update_todo(db, todo.id, TodoUpdate(title="New Title")) + + assert updated is not None + assert updated.title == "New Title" + + def test_update_description(self, db: Session) -> None: + """Updating the description field should persist the change.""" + todo = create_todo(db, TodoCreate(title="Task", description="Old desc")) + updated = update_todo(db, todo.id, TodoUpdate(description="New desc")) + + assert updated is not None + assert updated.description == "New desc" + # Title should be unchanged. + assert updated.title == "Task" + + def test_update_completed(self, db: Session) -> None: + """Updating the completed flag should persist the change.""" + todo = create_todo(db, TodoCreate(title="Task")) + assert todo.completed is False + + updated = update_todo(db, todo.id, TodoUpdate(completed=True)) + + assert updated is not None + assert updated.completed is True + + def test_partial_update_leaves_other_fields(self, db: Session) -> None: + """Fields not included in the update payload remain unchanged.""" + todo = create_todo( + db, + TodoCreate(title="Original", description="Keep me"), + ) + updated = update_todo(db, todo.id, TodoUpdate(completed=True)) + + assert updated is not None + assert updated.title == "Original" + assert updated.description == "Keep me" + assert updated.completed is True + + def test_update_nonexistent_returns_none(self, db: Session) -> None: + """Attempting to update a non-existent todo returns None.""" + result = update_todo(db, 999, TodoUpdate(title="Ghost")) + assert result is None + + def test_update_multiple_fields(self, db: Session) -> None: + """Multiple fields can be updated in a single call.""" + todo = create_todo(db, TodoCreate(title="Old", description="Old desc")) + updated = update_todo( + db, + todo.id, + TodoUpdate(title="New", description="New desc", completed=True), + ) + + assert updated is not None + assert updated.title == "New" + assert updated.description == "New desc" + assert updated.completed is True + + +# --------------------------------------------------------------------------- +# delete_todo +# --------------------------------------------------------------------------- + + +class TestDeleteTodo: + """Tests for the delete_todo function.""" + + def test_delete_existing_todo(self, db: Session) -> None: + """Deleting an existing todo should return True.""" + todo = create_todo(db, TodoCreate(title="Delete me")) + result = delete_todo(db, todo.id) + + assert result is True + + def test_deleted_todo_no_longer_exists(self, db: Session) -> None: + """After deletion the todo should not be retrievable.""" + todo = create_todo(db, TodoCreate(title="Gone soon")) + delete_todo(db, todo.id) + + assert get_todo(db, todo.id) is None + + def test_delete_nonexistent_returns_false(self, db: Session) -> None: + """Deleting a non-existent todo should return False.""" + result = delete_todo(db, 999) + assert result is False + + def test_delete_does_not_affect_other_todos(self, db: Session) -> None: + """Deleting one todo should leave other todos intact.""" + todo1 = create_todo(db, TodoCreate(title="Keep me")) + todo2 = create_todo(db, TodoCreate(title="Delete me")) + + delete_todo(db, todo2.id) + + assert get_todo(db, todo1.id) is not None + assert get_todo(db, todo2.id) is None + assert len(get_todos(db)) == 1 diff --git a/tests/test_database.py b/tests/test_database.py new file mode 100644 index 0000000..82a3de3 --- /dev/null +++ b/tests/test_database.py @@ -0,0 +1,89 @@ +"""Tests for app.database module. + +Verifies engine configuration, session factory behaviour, and the +``get_db`` dependency generator. +""" + +from __future__ import annotations + +from sqlalchemy import text +from sqlalchemy.orm import Session + +from app.database import Base, SessionLocal, engine, get_db + + +class TestEngine: + """Tests for the SQLAlchemy engine configuration.""" + + def test_engine_url_is_sqlite_in_memory(self) -> None: + """Engine must point at an in-memory SQLite database.""" + url_str = str(engine.url) + assert url_str == "sqlite://" + + def test_engine_can_connect(self) -> None: + """Engine should be able to establish a connection.""" + with engine.connect() as conn: + result = conn.execute(text("SELECT 1")) + assert result.scalar() == 1 + + +class TestSessionLocal: + """Tests for the SessionLocal session factory.""" + + def test_session_local_returns_session(self) -> None: + """SessionLocal() should return a valid Session instance.""" + session = SessionLocal() + try: + assert isinstance(session, Session) + finally: + session.close() + + def test_session_is_bound_to_engine(self) -> None: + """Session should be bound to the configured engine.""" + session = SessionLocal() + try: + assert session.bind is engine + finally: + session.close() + + +class TestBase: + """Tests for the declarative Base.""" + + def test_base_has_metadata(self) -> None: + """Base should expose a metadata attribute.""" + assert Base.metadata is not None + + +class TestGetDb: + """Tests for the get_db dependency generator.""" + + def test_get_db_yields_session(self) -> None: + """get_db should yield a Session and close it afterwards.""" + gen = get_db() + session = next(gen) + assert isinstance(session, Session) + # Exhaust the generator so the finally block runs. + try: + next(gen) + except StopIteration: + pass + + def test_get_db_closes_session(self) -> None: + """After the generator is exhausted the session should be closed.""" + gen = get_db() + session = next(gen) + try: + next(gen) + except StopIteration: + pass + # Accessing session.is_active on a closed session should not raise, + # but the session's internal connection should be released. + # We verify by simply running a new query – the old session should + # not interfere. + new_session = SessionLocal() + try: + result = new_session.execute(text("SELECT 1")) + assert result.scalar() == 1 + finally: + new_session.close() diff --git a/tests/test_main_startup.py b/tests/test_main_startup.py new file mode 100644 index 0000000..c569db4 --- /dev/null +++ b/tests/test_main_startup.py @@ -0,0 +1,39 @@ +"""Tests verifying that the FastAPI application creates tables on startup. + +Uses the ASGI lifespan protocol via httpx.AsyncClient / TestClient to +ensure ``Base.metadata.create_all`` is invoked during the startup event. +""" + +from __future__ import annotations + +from sqlalchemy import inspect, text + +from app.database import Base, engine +from app.models import Todo # noqa: F401 – register model + + +def test_tables_created_on_startup() -> None: + """After importing and starting the app, the 'todos' table must exist.""" + # Importing the app triggers the lifespan when used with TestClient. + from fastapi.testclient import TestClient + + from app.main import app + + with TestClient(app) as client: + # The lifespan has now fired; verify the table exists. + inspector = inspect(engine) + assert "todos" in inspector.get_table_names() + + # Also verify the root endpoint still works. + response = client.get("/") + assert response.status_code == 200 + assert response.json() == {"message": "Welcome to the Todo API"} + + +def test_create_all_is_idempotent() -> None: + """Calling create_all multiple times should not raise.""" + Base.metadata.create_all(bind=engine) + Base.metadata.create_all(bind=engine) + + inspector = inspect(engine) + assert "todos" in inspector.get_table_names() diff --git a/tests/test_models.py b/tests/test_models.py index e2706b1..93c3c2b 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,80 +1,205 @@ -"""Tests for Pydantic models defined in models.py.""" +"""Tests for app.models module. + +Verifies the Todo SQLAlchemy model definition, column attributes, +defaults, and basic CRUD operations against the in-memory database. +""" from __future__ import annotations +from datetime import datetime + import pytest -from pydantic import ValidationError +from sqlalchemy import inspect, text +from sqlalchemy.orm import Session -from models import TodoCreate, TodoResponse, TodoUpdate +from app.database import Base, SessionLocal, engine +from app.models import Todo -class TestTodoCreate: - """Tests for the TodoCreate model.""" +@pytest.fixture(autouse=True) +def _setup_tables() -> None: + """Create all tables before each test and drop them after.""" + Base.metadata.create_all(bind=engine) + yield # type: ignore[misc] + Base.metadata.drop_all(bind=engine) - def test_create_with_title_only(self) -> None: - """Title is the only required field.""" - todo = TodoCreate(title="Buy milk") - assert todo.title == "Buy milk" - assert todo.description is None + +@pytest.fixture() +def db_session() -> Session: + """Provide a fresh database session for a single test.""" + session = SessionLocal() + try: + yield session # type: ignore[misc] + finally: + session.close() + + +# --------------------------------------------------------------------------- +# Schema / column definition tests +# --------------------------------------------------------------------------- + + +class TestTodoTableSchema: + """Verify the 'todos' table schema matches the specification.""" + + def test_table_name(self) -> None: + """Model should map to the 'todos' table.""" + assert Todo.__tablename__ == "todos" + + def test_table_exists_after_create_all(self) -> None: + """After create_all the 'todos' table must exist in the DB.""" + inspector = inspect(engine) + table_names = inspector.get_table_names() + assert "todos" in table_names + + def test_columns_present(self) -> None: + """Table must contain the five required columns.""" + inspector = inspect(engine) + columns = {col["name"] for col in inspector.get_columns("todos")} + assert columns == {"id", "title", "description", "completed", "created_at"} + + def test_id_is_primary_key(self) -> None: + """The 'id' column must be the primary key.""" + inspector = inspect(engine) + pk = inspector.get_pk_constraint("todos") + assert "id" in pk["constrained_columns"] + + def test_title_not_nullable(self) -> None: + """The 'title' column must be NOT NULL.""" + inspector = inspect(engine) + cols = {c["name"]: c for c in inspector.get_columns("todos")} + assert cols["title"]["nullable"] is False + + def test_description_nullable(self) -> None: + """The 'description' column must be nullable.""" + inspector = inspect(engine) + cols = {c["name"]: c for c in inspector.get_columns("todos")} + assert cols["description"]["nullable"] is True + + +# --------------------------------------------------------------------------- +# CRUD / default-value tests +# --------------------------------------------------------------------------- + + +class TestTodoDefaults: + """Verify default column values when creating Todo instances.""" + + def test_completed_defaults_to_false(self, db_session: Session) -> None: + """A new Todo should have completed=False by default.""" + todo = Todo(title="Test item") + db_session.add(todo) + db_session.commit() + db_session.refresh(todo) assert todo.completed is False - def test_create_with_all_fields(self) -> None: - """All fields can be provided explicitly.""" - todo = TodoCreate(title="Buy milk", description="2% milk", completed=True) - assert todo.title == "Buy milk" - assert todo.description == "2% milk" + def test_description_defaults_to_none(self, db_session: Session) -> None: + """A new Todo without a description should have description=None.""" + todo = Todo(title="No description") + db_session.add(todo) + db_session.commit() + db_session.refresh(todo) + assert todo.description is None + + def test_created_at_is_set_automatically(self, db_session: Session) -> None: + """created_at should be populated automatically on insert.""" + before = datetime.utcnow() + todo = Todo(title="Timestamped") + db_session.add(todo) + db_session.commit() + db_session.refresh(todo) + after = datetime.utcnow() + + assert todo.created_at is not None + assert isinstance(todo.created_at, datetime) + assert before <= todo.created_at <= after + + def test_id_is_autoincremented(self, db_session: Session) -> None: + """Sequential inserts should yield incrementing ids.""" + todo1 = Todo(title="First") + todo2 = Todo(title="Second") + db_session.add_all([todo1, todo2]) + db_session.commit() + db_session.refresh(todo1) + db_session.refresh(todo2) + assert todo1.id is not None + assert todo2.id is not None + assert todo2.id > todo1.id + + +# --------------------------------------------------------------------------- +# CRUD operation tests +# --------------------------------------------------------------------------- + + +class TestTodoCRUD: + """Verify basic create / read / update / delete operations.""" + + def test_create_and_read(self, db_session: Session) -> None: + """Creating a Todo and reading it back should return matching data.""" + todo = Todo(title="Buy milk", description="2% milk from store") + db_session.add(todo) + db_session.commit() + + fetched = db_session.query(Todo).filter_by(id=todo.id).first() + assert fetched is not None + assert fetched.title == "Buy milk" + assert fetched.description == "2% milk from store" + assert fetched.completed is False + + def test_update_todo(self, db_session: Session) -> None: + """Updating a Todo's fields should persist the changes.""" + todo = Todo(title="Old title") + db_session.add(todo) + db_session.commit() + + todo.title = "New title" + todo.completed = True + db_session.commit() + db_session.refresh(todo) + + assert todo.title == "New title" assert todo.completed is True - def test_create_missing_title_raises(self) -> None: - """Omitting title must raise a validation error.""" - with pytest.raises(ValidationError): - TodoCreate() # type: ignore[call-arg] - - def test_create_empty_title_raises(self) -> None: - """An empty string title must raise a validation error.""" - with pytest.raises(ValidationError): - TodoCreate(title="") - - -class TestTodoUpdate: - """Tests for the TodoUpdate model.""" - - def test_update_all_none_by_default(self) -> None: - """All fields default to None.""" - update = TodoUpdate() - assert update.title is None - assert update.description is None - assert update.completed is None - - def test_update_partial(self) -> None: - """Only supplied fields are set.""" - update = TodoUpdate(completed=True) - assert update.completed is True - assert update.title is None - - -class TestTodoResponse: - """Tests for the TodoResponse model.""" - - def test_response_roundtrip(self) -> None: - """All fields are populated correctly.""" - data = { - "id": 1, - "title": "Test", - "description": "A test todo", - "completed": False, - "created_at": "2024-01-01T00:00:00+00:00", - } - resp = TodoResponse(**data) - assert resp.id == 1 - assert resp.title == "Test" - assert resp.description == "A test todo" - assert resp.completed is False - assert resp.created_at == "2024-01-01T00:00:00+00:00" - - def test_response_description_defaults_to_none(self) -> None: - """Description is optional and defaults to None.""" - resp = TodoResponse( - id=2, title="No desc", completed=False, created_at="2024-01-01T00:00:00" + def test_delete_todo(self, db_session: Session) -> None: + """Deleting a Todo should remove it from the database.""" + todo = Todo(title="To delete") + db_session.add(todo) + db_session.commit() + todo_id = todo.id + + db_session.delete(todo) + db_session.commit() + + assert db_session.query(Todo).filter_by(id=todo_id).first() is None + + def test_query_all(self, db_session: Session) -> None: + """Querying all Todos should return every inserted row.""" + db_session.add_all( + [Todo(title="A"), Todo(title="B"), Todo(title="C")] ) - assert resp.description is None + db_session.commit() + + todos = db_session.query(Todo).all() + assert len(todos) == 3 + + +# --------------------------------------------------------------------------- +# Repr test +# --------------------------------------------------------------------------- + + +class TestTodoRepr: + """Verify the __repr__ of the Todo model.""" + + def test_repr(self, db_session: Session) -> None: + """__repr__ should contain id, title, and completed.""" + todo = Todo(title="Repr test") + db_session.add(todo) + db_session.commit() + db_session.refresh(todo) + + r = repr(todo) + assert "Repr test" in r + assert str(todo.id) in r + assert "False" in r diff --git a/tests/test_routes.py b/tests/test_routes.py new file mode 100644 index 0000000..370f9dc --- /dev/null +++ b/tests/test_routes.py @@ -0,0 +1,343 @@ +"""Tests for the Todo API route endpoints. + +Covers all CRUD operations exposed via ``app/routes.py``: + +- ``GET /todos`` – list with pagination +- ``GET /todos/{todo_id}`` – get by id, including 404 +- ``POST /todos`` – create, status 201 +- ``PUT /todos/{todo_id}`` – update, including 404 +- ``DELETE /todos/{todo_id}`` – delete 204, including 404 + +Uses an independent in-memory SQLite database for full isolation. +""" + +from __future__ import annotations + +from typing import Generator + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from app.database import Base, get_db +from app.main import app + +# --------------------------------------------------------------------------- +# Test database setup +# --------------------------------------------------------------------------- +SQLALCHEMY_DATABASE_URL = "sqlite://" + +engine = create_engine( + SQLALCHEMY_DATABASE_URL, + connect_args={"check_same_thread": False}, +) +TestingSessionLocal: sessionmaker[Session] = sessionmaker( + autocommit=False, + autoflush=False, + bind=engine, +) + + +def override_get_db() -> Generator[Session, None, None]: + """Yield a test database session.""" + db = TestingSessionLocal() + try: + yield db + finally: + db.close() + + +app.dependency_overrides[get_db] = override_get_db + + +@pytest.fixture(autouse=True) +def _setup_database() -> Generator[None, None, None]: + """Create all tables before each test and drop them after.""" + Base.metadata.create_all(bind=engine) + yield + Base.metadata.drop_all(bind=engine) + + +client = TestClient(app) + + +# --------------------------------------------------------------------------- +# Helper +# --------------------------------------------------------------------------- +def _create_todo( + title: str = "Test Todo", + description: str | None = None, +) -> dict: + """Create a todo via POST and return the JSON response body.""" + payload: dict = {"title": title} + if description is not None: + payload["description"] = description + response = client.post("/todos", json=payload) + assert response.status_code == 201 + return response.json() + + +# --------------------------------------------------------------------------- +# POST /todos +# --------------------------------------------------------------------------- +class TestCreateTodo: + """Tests for the POST /todos endpoint.""" + + def test_create_todo_minimal(self) -> None: + """Create a todo with only a title.""" + response = client.post("/todos", json={"title": "Buy milk"}) + assert response.status_code == 201 + data = response.json() + assert data["title"] == "Buy milk" + assert data["description"] is None + assert data["completed"] is False + assert "id" in data + assert "created_at" in data + + def test_create_todo_with_description(self) -> None: + """Create a todo with title and description.""" + response = client.post( + "/todos", + json={"title": "Groceries", "description": "Eggs, bread, cheese"}, + ) + assert response.status_code == 201 + data = response.json() + assert data["title"] == "Groceries" + assert data["description"] == "Eggs, bread, cheese" + + def test_create_todo_empty_title_rejected(self) -> None: + """An empty title string must be rejected (422).""" + response = client.post("/todos", json={"title": ""}) + assert response.status_code == 422 + + def test_create_todo_missing_title_rejected(self) -> None: + """A request body without a title must be rejected (422).""" + response = client.post("/todos", json={}) + assert response.status_code == 422 + + +# --------------------------------------------------------------------------- +# GET /todos +# --------------------------------------------------------------------------- +class TestListTodos: + """Tests for the GET /todos endpoint.""" + + def test_list_empty(self) -> None: + """An empty database returns an empty list.""" + response = client.get("/todos") + assert response.status_code == 200 + assert response.json() == [] + + def test_list_multiple(self) -> None: + """Multiple todos are returned.""" + _create_todo("First") + _create_todo("Second") + response = client.get("/todos") + assert response.status_code == 200 + data = response.json() + assert len(data) == 2 + + def test_list_with_skip(self) -> None: + """The skip query parameter offsets results.""" + _create_todo("A") + _create_todo("B") + _create_todo("C") + response = client.get("/todos", params={"skip": 2}) + assert response.status_code == 200 + data = response.json() + assert len(data) == 1 + + def test_list_with_limit(self) -> None: + """The limit query parameter caps results.""" + _create_todo("A") + _create_todo("B") + _create_todo("C") + response = client.get("/todos", params={"limit": 2}) + assert response.status_code == 200 + data = response.json() + assert len(data) == 2 + + def test_list_with_skip_and_limit(self) -> None: + """Both skip and limit work together.""" + for i in range(5): + _create_todo(f"Item {i}") + response = client.get("/todos", params={"skip": 1, "limit": 2}) + assert response.status_code == 200 + data = response.json() + assert len(data) == 2 + + +# --------------------------------------------------------------------------- +# GET /todos/{todo_id} +# --------------------------------------------------------------------------- +class TestGetTodo: + """Tests for the GET /todos/{todo_id} endpoint.""" + + def test_get_existing(self) -> None: + """Retrieve a todo by its ID.""" + created = _create_todo("Read book") + todo_id = created["id"] + response = client.get(f"/todos/{todo_id}") + assert response.status_code == 200 + data = response.json() + assert data["id"] == todo_id + assert data["title"] == "Read book" + + def test_get_not_found(self) -> None: + """A non-existent ID returns 404 with proper detail.""" + response = client.get("/todos/99999") + assert response.status_code == 404 + assert response.json()["detail"] == "Todo not found" + + +# --------------------------------------------------------------------------- +# PUT /todos/{todo_id} +# --------------------------------------------------------------------------- +class TestUpdateTodo: + """Tests for the PUT /todos/{todo_id} endpoint.""" + + def test_update_title(self) -> None: + """Update only the title of an existing todo.""" + created = _create_todo("Old title") + todo_id = created["id"] + response = client.put( + f"/todos/{todo_id}", + json={"title": "New title"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["title"] == "New title" + # Other fields remain unchanged + assert data["completed"] is False + + def test_update_completed(self) -> None: + """Mark a todo as completed.""" + created = _create_todo("Task") + todo_id = created["id"] + response = client.put( + f"/todos/{todo_id}", + json={"completed": True}, + ) + assert response.status_code == 200 + assert response.json()["completed"] is True + + def test_update_description(self) -> None: + """Update the description of a todo.""" + created = _create_todo("Task", description="old desc") + todo_id = created["id"] + response = client.put( + f"/todos/{todo_id}", + json={"description": "new desc"}, + ) + assert response.status_code == 200 + assert response.json()["description"] == "new desc" + + def test_update_multiple_fields(self) -> None: + """Update multiple fields in a single request.""" + created = _create_todo("Original") + todo_id = created["id"] + response = client.put( + f"/todos/{todo_id}", + json={"title": "Updated", "completed": True, "description": "Done"}, + ) + assert response.status_code == 200 + data = response.json() + assert data["title"] == "Updated" + assert data["completed"] is True + assert data["description"] == "Done" + + def test_update_not_found(self) -> None: + """Updating a non-existent todo returns 404.""" + response = client.put( + "/todos/99999", + json={"title": "Nope"}, + ) + assert response.status_code == 404 + assert response.json()["detail"] == "Todo not found" + + def test_update_empty_title_rejected(self) -> None: + """An empty title string in an update must be rejected (422).""" + created = _create_todo("Valid") + todo_id = created["id"] + response = client.put( + f"/todos/{todo_id}", + json={"title": ""}, + ) + assert response.status_code == 422 + + +# --------------------------------------------------------------------------- +# DELETE /todos/{todo_id} +# --------------------------------------------------------------------------- +class TestDeleteTodo: + """Tests for the DELETE /todos/{todo_id} endpoint.""" + + def test_delete_existing(self) -> None: + """Deleting an existing todo returns 204 with no body.""" + created = _create_todo("To be deleted") + todo_id = created["id"] + response = client.delete(f"/todos/{todo_id}") + assert response.status_code == 204 + assert response.content == b"" + + # Verify it is gone + get_response = client.get(f"/todos/{todo_id}") + assert get_response.status_code == 404 + + def test_delete_not_found(self) -> None: + """Deleting a non-existent todo returns 404.""" + response = client.delete("/todos/99999") + assert response.status_code == 404 + assert response.json()["detail"] == "Todo not found" + + def test_delete_idempotent(self) -> None: + """Deleting the same todo twice returns 404 on the second call.""" + created = _create_todo("Delete me") + todo_id = created["id"] + first = client.delete(f"/todos/{todo_id}") + assert first.status_code == 204 + second = client.delete(f"/todos/{todo_id}") + assert second.status_code == 404 + + +# --------------------------------------------------------------------------- +# Integration / round-trip tests +# --------------------------------------------------------------------------- +class TestRoundTrip: + """End-to-end tests exercising multiple endpoints in sequence.""" + + def test_create_read_update_delete(self) -> None: + """Full lifecycle: create → read → update → read → delete → 404.""" + # Create + created = _create_todo("Lifecycle test", description="Step 1") + todo_id = created["id"] + assert created["completed"] is False + + # Read + read_resp = client.get(f"/todos/{todo_id}") + assert read_resp.status_code == 200 + assert read_resp.json()["title"] == "Lifecycle test" + + # Update + update_resp = client.put( + f"/todos/{todo_id}", + json={"completed": True, "description": "Done"}, + ) + assert update_resp.status_code == 200 + assert update_resp.json()["completed"] is True + assert update_resp.json()["description"] == "Done" + # Title should not have changed + assert update_resp.json()["title"] == "Lifecycle test" + + # Read again to confirm persistence + read_resp2 = client.get(f"/todos/{todo_id}") + assert read_resp2.json()["completed"] is True + + # Delete + del_resp = client.delete(f"/todos/{todo_id}") + assert del_resp.status_code == 204 + + # Verify deletion + final_resp = client.get(f"/todos/{todo_id}") + assert final_resp.status_code == 404 diff --git a/tests/test_schemas.py b/tests/test_schemas.py new file mode 100644 index 0000000..34dfa91 --- /dev/null +++ b/tests/test_schemas.py @@ -0,0 +1,251 @@ +"""Tests for app.schemas Pydantic models. + +Covers validation rules, default values, orm_mode compatibility, +and edge cases for TodoBase, TodoCreate, TodoUpdate, and TodoResponse. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any, Dict + +import pytest +from pydantic import ValidationError + +from app.schemas import TodoBase, TodoCreate, TodoResponse, TodoUpdate + + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +class _FakeTodoORM: + """Mimics a SQLAlchemy model instance for orm_mode testing.""" + + def __init__(self, **kwargs: Any) -> None: + """Initialise with arbitrary attributes.""" + for key, value in kwargs.items(): + setattr(self, key, value) + + +def _valid_response_data() -> Dict[str, Any]: + """Return a minimal valid dict for TodoResponse.""" + return { + "id": 1, + "title": "Buy milk", + "description": None, + "completed": False, + "created_at": datetime(2024, 1, 1, 12, 0, 0), + } + + +# --------------------------------------------------------------------------- +# TodoBase +# --------------------------------------------------------------------------- + + +class TestTodoBase: + """Tests for the TodoBase schema.""" + + def test_valid_with_title_only(self) -> None: + """TodoBase accepts just a title; description defaults to None.""" + schema = TodoBase(title="Do laundry") + assert schema.title == "Do laundry" + assert schema.description is None + + def test_valid_with_title_and_description(self) -> None: + """TodoBase accepts both title and description.""" + schema = TodoBase(title="Clean house", description="Kitchen and bathroom") + assert schema.title == "Clean house" + assert schema.description == "Kitchen and bathroom" + + def test_title_required(self) -> None: + """Omitting title raises a ValidationError.""" + with pytest.raises(ValidationError) as exc_info: + TodoBase() # type: ignore[call-arg] + assert "title" in str(exc_info.value) + + def test_empty_title_rejected(self) -> None: + """An empty string title violates the min_length constraint.""" + with pytest.raises(ValidationError) as exc_info: + TodoBase(title="") + assert "title" in str(exc_info.value) + + def test_title_max_length(self) -> None: + """A title exceeding 255 characters is rejected.""" + with pytest.raises(ValidationError): + TodoBase(title="x" * 256) + + def test_description_max_length(self) -> None: + """A description exceeding 1024 characters is rejected.""" + with pytest.raises(ValidationError): + TodoBase(title="Valid", description="y" * 1025) + + +# --------------------------------------------------------------------------- +# TodoCreate +# --------------------------------------------------------------------------- + + +class TestTodoCreate: + """Tests for the TodoCreate schema.""" + + def test_inherits_todo_base(self) -> None: + """TodoCreate is a subclass of TodoBase.""" + assert issubclass(TodoCreate, TodoBase) + + def test_valid_creation(self) -> None: + """TodoCreate works with title only.""" + schema = TodoCreate(title="Write tests") + assert schema.title == "Write tests" + assert schema.description is None + + def test_valid_creation_with_description(self) -> None: + """TodoCreate works with title and description.""" + schema = TodoCreate(title="Write tests", description="Unit and integration") + assert schema.description == "Unit and integration" + + def test_title_required(self) -> None: + """Omitting title in TodoCreate raises ValidationError.""" + with pytest.raises(ValidationError): + TodoCreate() # type: ignore[call-arg] + + +# --------------------------------------------------------------------------- +# TodoUpdate +# --------------------------------------------------------------------------- + + +class TestTodoUpdate: + """Tests for the TodoUpdate schema.""" + + def test_all_fields_optional(self) -> None: + """TodoUpdate can be created with no arguments.""" + schema = TodoUpdate() + assert schema.title is None + assert schema.description is None + assert schema.completed is None + + def test_partial_update_title(self) -> None: + """Only title can be supplied.""" + schema = TodoUpdate(title="New title") + assert schema.title == "New title" + assert schema.description is None + assert schema.completed is None + + def test_partial_update_completed(self) -> None: + """Only completed can be supplied.""" + schema = TodoUpdate(completed=True) + assert schema.completed is True + assert schema.title is None + + def test_partial_update_description(self) -> None: + """Only description can be supplied.""" + schema = TodoUpdate(description="Updated desc") + assert schema.description == "Updated desc" + + def test_all_fields_supplied(self) -> None: + """All three fields can be supplied together.""" + schema = TodoUpdate( + title="Revised", description="Revised desc", completed=True + ) + assert schema.title == "Revised" + assert schema.description == "Revised desc" + assert schema.completed is True + + def test_empty_title_rejected(self) -> None: + """An empty string title is rejected even in updates.""" + with pytest.raises(ValidationError): + TodoUpdate(title="") + + def test_title_max_length(self) -> None: + """Title exceeding 255 chars is rejected in updates.""" + with pytest.raises(ValidationError): + TodoUpdate(title="x" * 256) + + def test_description_max_length(self) -> None: + """Description exceeding 1024 chars is rejected in updates.""" + with pytest.raises(ValidationError): + TodoUpdate(description="y" * 1025) + + +# --------------------------------------------------------------------------- +# TodoResponse +# --------------------------------------------------------------------------- + + +class TestTodoResponse: + """Tests for the TodoResponse schema.""" + + def test_valid_response(self) -> None: + """TodoResponse accepts all required fields.""" + data = _valid_response_data() + schema = TodoResponse(**data) + assert schema.id == 1 + assert schema.title == "Buy milk" + assert schema.description is None + assert schema.completed is False + assert isinstance(schema.created_at, datetime) + + def test_inherits_todo_base(self) -> None: + """TodoResponse is a subclass of TodoBase.""" + assert issubclass(TodoResponse, TodoBase) + + def test_orm_mode_enabled(self) -> None: + """TodoResponse Config has orm_mode set to True.""" + assert TodoResponse.Config.orm_mode is True + + def test_from_orm_object(self) -> None: + """TodoResponse can be constructed from an ORM-like object.""" + now = datetime.now(tz=timezone.utc) + orm_obj = _FakeTodoORM( + id=42, + title="ORM test", + description="From ORM", + completed=True, + created_at=now, + ) + schema = TodoResponse.from_orm(orm_obj) + assert schema.id == 42 + assert schema.title == "ORM test" + assert schema.description == "From ORM" + assert schema.completed is True + assert schema.created_at == now + + def test_missing_id_rejected(self) -> None: + """Omitting id raises ValidationError.""" + data = _valid_response_data() + del data["id"] + with pytest.raises(ValidationError): + TodoResponse(**data) + + def test_missing_completed_rejected(self) -> None: + """Omitting completed raises ValidationError.""" + data = _valid_response_data() + del data["completed"] + with pytest.raises(ValidationError): + TodoResponse(**data) + + def test_missing_created_at_rejected(self) -> None: + """Omitting created_at raises ValidationError.""" + data = _valid_response_data() + del data["created_at"] + with pytest.raises(ValidationError): + TodoResponse(**data) + + def test_response_with_description(self) -> None: + """TodoResponse correctly stores a non-None description.""" + data = _valid_response_data() + data["description"] = "Some details" + schema = TodoResponse(**data) + assert schema.description == "Some details" + + def test_created_at_accepts_string(self) -> None: + """TodoResponse coerces an ISO datetime string to a datetime object.""" + data = _valid_response_data() + data["created_at"] = "2024-06-15T10:30:00" + schema = TodoResponse(**data) + assert isinstance(schema.created_at, datetime) + assert schema.created_at.year == 2024 + assert schema.created_at.month == 6 diff --git a/tests/test_seed_data.py b/tests/test_seed_data.py new file mode 100644 index 0000000..e1b4b7a --- /dev/null +++ b/tests/test_seed_data.py @@ -0,0 +1,111 @@ +"""Tests for the seed_data module. + +Verifies that :func:`seed_data.seed` correctly inserts the expected +sample todos into a fresh in-memory database, and that the inserted +records have the right fields and values. +""" + +from __future__ import annotations + +from typing import Generator + +import pytest +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker + +from app.database import Base +from app.models import Todo +from seed_data import SAMPLE_TODOS, seed + + +@pytest.fixture() +def db_session() -> Generator[Session, None, None]: + """Provide a clean in-memory SQLite session for each test.""" + engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + ) + Base.metadata.create_all(bind=engine) + testing_session = sessionmaker(autocommit=False, autoflush=False, bind=engine) + session = testing_session() + try: + yield session + finally: + session.close() + Base.metadata.drop_all(bind=engine) + + +class TestSeed: + """Tests for the seed function.""" + + def test_seed_returns_correct_count(self, db_session: Session) -> None: + """seed() should return the same number of todos as SAMPLE_TODOS.""" + result = seed(db_session) + assert len(result) == len(SAMPLE_TODOS) + + def test_seed_items_have_ids(self, db_session: Session) -> None: + """Every seeded todo should have a non-None integer id.""" + result = seed(db_session) + for todo in result: + assert todo.id is not None + assert isinstance(todo.id, int) + + def test_seed_items_have_created_at(self, db_session: Session) -> None: + """Every seeded todo should have a created_at timestamp.""" + result = seed(db_session) + for todo in result: + assert todo.created_at is not None + + def test_seed_titles_match(self, db_session: Session) -> None: + """Seeded todo titles should match the sample data.""" + result = seed(db_session) + expected_titles = [item["title"] for item in SAMPLE_TODOS] + actual_titles = [todo.title for todo in result] + assert actual_titles == expected_titles + + def test_seed_descriptions_match(self, db_session: Session) -> None: + """Seeded todo descriptions should match the sample data.""" + result = seed(db_session) + for todo, sample in zip(result, SAMPLE_TODOS): + assert todo.description == sample["description"] + + def test_seed_completed_flags_match(self, db_session: Session) -> None: + """Seeded todo completed flags should match the sample data.""" + result = seed(db_session) + for todo, sample in zip(result, SAMPLE_TODOS): + assert todo.completed == sample["completed"] + + def test_seed_persists_to_database(self, db_session: Session) -> None: + """After seeding, querying the database should return all items.""" + seed(db_session) + count = db_session.query(Todo).count() + assert count == len(SAMPLE_TODOS) + + def test_seed_twice_doubles_records(self, db_session: Session) -> None: + """Calling seed() twice should insert records cumulatively.""" + seed(db_session) + seed(db_session) + count = db_session.query(Todo).count() + assert count == len(SAMPLE_TODOS) * 2 + + def test_seed_contains_completed_and_incomplete(self, db_session: Session) -> None: + """Sample data should include at least one completed and one incomplete todo.""" + result = seed(db_session) + completed = [t for t in result if t.completed] + incomplete = [t for t in result if not t.completed] + assert len(completed) >= 1, "Expected at least one completed todo in seed data" + assert len(incomplete) >= 1, "Expected at least one incomplete todo in seed data" + + def test_seed_contains_item_without_description(self, db_session: Session) -> None: + """Sample data should include at least one todo with no description.""" + result = seed(db_session) + none_descriptions = [t for t in result if t.description is None] + assert len(none_descriptions) >= 1, ( + "Expected at least one todo with description=None in seed data" + ) + + def test_sample_todos_titles_non_empty(self) -> None: + """All sample todo titles should be non-empty strings.""" + for item in SAMPLE_TODOS: + assert isinstance(item["title"], str) + assert len(item["title"]) >= 1