Add Integration Tests for a Task Management API with Database

Problem/Feature Description

A development team maintains a FastAPI service that stores tasks in a SQLite database via SQLAlchemy. They recently merged two PRs in quick succession and discovered that their existing tests were passing intermittently — sometimes a test would find data left behind by a previous test and fail, other times it would pass. The team suspects their database setup in tests is incorrect.

You have been asked to write a proper database-backed test suite that eliminates test pollution. The suite needs to exercise the task creation and listing endpoints and must be structured so that each test always starts with a predictable empty state — even if a previous test fails halfway through.

Output Specification

Produce the following files:

• tests/conftest.py — database engine fixture, database session fixture, a cleanup fixture, and a test client fixture that injects the test session
• tests/test_tasks.py — at least three tests covering: creating a task, listing tasks, and fetching a single task by ID
• pyproject.toml — with pytest async configuration

Input Files

The following files are provided as inputs. Extract them before beginning.

=============== FILE: app/database.py =============== from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker, declarative_base

SQLALCHEMY_DATABASE_URL = "sqlite:///./tasks.db"

engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}) SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine) Base = declarative_base()

def get_db(): db = SessionLocal() try: yield db finally: db.close()

=============== FILE: app/models.py =============== from sqlalchemy import Column, Integer, String, Boolean from app.database import Base

class Task(Base): tablename = "tasks"

id = Column(Integer, primary_key=True, index=True)
title = Column(String, nullable=False)
description = Column(String, default="")
completed = Column(Boolean, default=False)

=============== FILE: app/main.py =============== from fastapi import FastAPI, HTTPException, Depends from pydantic import BaseModel from sqlalchemy.orm import Session from app.database import get_db, Base, engine from app import models

Base.metadata.create_all(bind=engine)

app = FastAPI()

class TaskCreate(BaseModel): title: str description: str = ""

@app.post("/tasks", status_code=201) def create_task(body: TaskCreate, db: Session = Depends(get_db)): task = models.Task(title=body.title, description=body.description) db.add(task) db.commit() db.refresh(task) return {"data": {"id": task.id, "title": task.title, "description": task.description, "completed": task.completed}}

@app.get("/tasks") def list_tasks(db: Session = Depends(get_db)): tasks = db.query(models.Task).all() return {"data": [{"id": t.id, "title": t.title, "description": t.description, "completed": t.completed} for t in tasks]}

@app.get("/tasks/{task_id}") def get_task(task_id: int, db: Session = Depends(get_db)): task = db.query(models.Task).filter(models.Task.id == task_id).first() if not task: raise HTTPException(status_code=404, detail="Task not found") return {"data": {"id": task.id, "title": task.title, "description": task.description, "completed": task.completed}}