Fixtures, временная база и dependency override

Независимые проверки Roomly

Тесты Roomly создают учебные данные, но не должны обращаться к рабочей базе или зависеть от записей другого теста. Подготовьте для каждого теста отдельное пустое хранилище.

Запросы одного теста должны видеть созданные им записи. Другой тест начинает с чистой базы независимо от порядка запуска. После завершения, в том числе при обычном падении проверки, освободите временные ресурсы и восстановите прежние подмены зависимостей приложения.

Что уже дано

Перед вашим кодом предоставлены pytest, TestClient и полный самостоятельный API ниже. Обычный DATABASE_URL задаёт служебная подготовка: это собственный файл-маркер, изображающий прежнюю базу, к нему тесты не должны обращаться. create_db_engine создаёт ленивый engine с FK на каждом соединении; обычная init_db не вызывается при импорте. API и его фабрику переписывать не нужно.

from datetime import datetime, timezone
from sqlalchemy import create_engine, event, select, ForeignKey, DateTime
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship, sessionmaker, Session

class Base(DeclarativeBase):
    pass

class Room(Base):
    __tablename__ = "rooms"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str]
    capacity: Mapped[int]
    bookings: Mapped[list["Booking"]] = relationship(back_populates="room")

class Booking(Base):
    __tablename__ = "bookings"
    id: Mapped[int] = mapped_column(primary_key=True)
    room_id: Mapped[int] = mapped_column(ForeignKey("rooms.id"))
    starts_at: Mapped[datetime] = mapped_column(DateTime(timezone=False))
    ends_at: Mapped[datetime] = mapped_column(DateTime(timezone=False))
    room: Mapped["Room"] = relationship(back_populates="bookings")

def create_db_engine(url):
    configured_engine = create_engine(url, connect_args={"check_same_thread": False})
    @event.listens_for(configured_engine, "connect")
    def enable_foreign_keys(connection, connection_record):
        connection.execute("PRAGMA foreign_keys=ON")
    return configured_engine

engine = create_db_engine(DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
def init_db():
    Base.metadata.create_all(bind=engine)

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

def to_db_utc(value: datetime) -> datetime:
    if value.utcoffset() is None:
        raise ValueError("Timezone required")
    return value.astimezone(timezone.utc).replace(tzinfo=None)

from fastapi import FastAPI, Depends, HTTPException
from pydantic import field_validator, model_validator
from pydantic import BaseModel, Field, ConfigDict
from typing import Optional

class RoomIn(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    capacity: int = Field(..., ge=1, le=50)

class RoomSchema(BaseModel):
    model_config = ConfigDict(from_attributes=True)
    id: int
    name: str
    capacity: int
class RoomPatch(BaseModel):
    name: Optional[str] = Field(None, min_length=2, max_length=50)
    capacity: Optional[int] = Field(None, ge=1, le=50)
class BookingIn(BaseModel):
    starts_at: datetime
    ends_at: datetime

    @field_validator("starts_at", "ends_at")
    @classmethod
    def require_timezone(cls, value: datetime) -> datetime:
        if value.utcoffset() is None:
            raise ValueError("Timezone required")
        return value.astimezone(timezone.utc)

    @model_validator(mode="after")
    def check_interval(self) -> "BookingIn":
        if self.ends_at <= self.starts_at:
            raise ValueError("End must be after start")
        return self
class BookingOut(BaseModel):
    id: int
    room_id: int
    starts_at: datetime
    ends_at: datetime

app = FastAPI()

@app.get("/api/rooms", response_model=list[RoomSchema])
def list_rooms(db: Session = Depends(get_db)):
    return db.scalars(select(Room).order_by(Room.id)).all()

@app.get("/api/rooms/{room_id}", response_model=RoomSchema)
def read_room(room_id: int, db: Session = Depends(get_db)):
    room = db.get(Room, room_id)
    if room is None:
        raise HTTPException(status_code=404, detail="Room not found")
    return room

@app.post("/api/rooms", response_model=RoomSchema, status_code=201)
def create_room(data: RoomIn, db: Session = Depends(get_db)):
    room = Room(name=data.name, capacity=data.capacity)
    db.add(room)
    db.commit()
    db.refresh(room)
    return room

@app.put("/api/rooms/{room_id}", response_model=RoomSchema)
def replace_room(room_id: int, data: RoomIn, db: Session = Depends(get_db)):
    room = db.get(Room, room_id)
    if room is None:
        raise HTTPException(status_code=404, detail="Room not found")
    room.name, room.capacity = data.name, data.capacity
    db.commit()
    db.refresh(room)
    return room

@app.delete("/api/rooms/{room_id}", status_code=204)
def delete_room(room_id: int, db: Session = Depends(get_db)):
    room = db.get(Room, room_id)
    if room is None:
        raise HTTPException(status_code=404, detail="Room not found")
    if db.scalars(select(Booking).where(Booking.room_id == room_id)).first() is not None:
        raise HTTPException(status_code=409, detail="Room has bookings")
    db.delete(room)
    db.commit()

@app.patch("/api/rooms/{room_id}", response_model=RoomSchema)
def patch_room(room_id: int, data: RoomPatch, db: Session = Depends(get_db)):
    room = db.get(Room, room_id)
    if room is None:
        raise HTTPException(status_code=404, detail="Room not found")
    for field in data.model_fields_set:
        if getattr(data, field) is None:
            raise HTTPException(status_code=422, detail="Fields cannot be null")
    for field in data.model_fields_set:
        setattr(room, field, getattr(data, field))
    db.commit()
    db.refresh(room)
    return room
def find_conflict(db: Session, room_id: int, new_start: datetime, new_end: datetime) -> Booking | None:
    start_utc = to_db_utc(new_start)
    end_utc = to_db_utc(new_end)
    statement = select(Booking).where(
        Booking.room_id == room_id,
        Booking.starts_at < end_utc,
        Booking.ends_at > start_utc,
    )
    return db.scalars(statement).first()
def public_booking(booking: Booking):
    return BookingOut(
        id=booking.id, room_id=booking.room_id,
        starts_at=booking.starts_at.replace(tzinfo=timezone.utc),
        ends_at=booking.ends_at.replace(tzinfo=timezone.utc),
    )

@app.post("/api/rooms/{room_id}/bookings", status_code=201, response_model=BookingOut)
def create_booking(room_id: int, data: BookingIn, db: Session = Depends(get_db)):
    if db.get(Room, room_id) is None:
        raise HTTPException(status_code=404, detail="Room not found")
    if find_conflict(db, room_id, data.starts_at, data.ends_at) is not None:
        raise HTTPException(status_code=409, detail="Booking conflict")
    booking = Booking(room_id=room_id, starts_at=to_db_utc(data.starts_at), ends_at=to_db_utc(data.ends_at))
    db.add(booking)
    db.commit()
    db.refresh(booking)
    return public_booking(booking)

@app.get("/api/rooms/{room_id}/bookings", response_model=list[BookingOut])
def list_bookings(room_id: int, db: Session = Depends(get_db)):
    if db.get(Room, room_id) is None:
        raise HTTPException(status_code=404, detail="Room not found")
    bookings = db.scalars(select(Booking).where(Booking.room_id == room_id).order_by(Booking.id)).all()
    return [public_booking(booking) for booking in bookings]

Следующие два теста также уже предоставлены. Они будут запущены настоящим pytest с вашей fixture; их assertions не являются TODO. Служебная переменная _roomly_force_failure только по отдельному запросу runner вызывает обычный assertfailure в первом тесте.

def test_create_room(client):
    response = client.post("/api/rooms", json={"name": "Альфа", "capacity": 4})
    assert response.status_code == 201
    room = response.json()
    assert type(room["id"]) is int
    assert room["name"] == "Альфа" and room["capacity"] == 4
    response = client.get("/api/rooms")
    assert response.status_code == 200 and response.json() == [room]
    # Служебный запрос этого сбоя проверяет finally после обычного падения теста.
    if _roomly_force_failure:
        assert False, "Intentional assertion to exercise fixture cleanup"

def test_empty_rooms(client):
    response = client.get("/api/rooms")
    assert response.status_code == 200 and response.json() == []

Runner создаёт принадлежащую ему временную папку и запускает настоящий pytest, включая tmp_path и fixture teardown. Наблюдение записывает используемые Session, подключения, FK, dispose и удаление файла, не выполняя эти действия вместо fixture. После teardown сохраняется отчёт; затем runner аварийно освобождает утёкшие ресурсы неисправных вариантов только для безопасного завершения проверки. Эта поздняя уборка не исправляет оценённый отчёт. Ранее установлены подмена get_db и посторонняя подмена: во время теста первую нужно заменить, после теста обе восстановить. Скрытый grader получает JSON-наблюдения; писать runner, наблюдатели или готовые тесты не нужно.

Что нужно сделать

Определите pytest fixture с именем client и параметром tmp_path, с обычным scope="function". Создайте собственный SQLite-файл внутри tmp_path, engine через готовую create_db_engine и таблицы Base именно в этом engine. Фабрика тестовых Session должна быть связана с ним. Подмените в app.dependency_overrides тот же объект get_db, который используют обработчики. Подмена открывает новую Session на каждый запрос, выдаёт её через yield и гарантированно закрывает; запросы одного теста видят один файл.

Сохраните прежние overrides, откройте настоящий TestClient(app) через with и передайте его тесту через yield. В finally после выхода из клиентского контекста восстановите прежний словарь, вызовите dispose тестового engine и удалите только собственный файл. Файл может называться test.db или иначе. На обычной базе нельзя выполнять SQL, читать её строки, вызывать init_db, очищать или удалять её.

Оба готовых теста должны проходить в любом порядке и повторно. При отдельном обычном падении assert в первом тесте teardown обязан так же восстановить overrides и освободить ресурсы. Не изменяйте готовый API и тесты. Принудительное уничтожение процесса, сеть, новый алгоритм бронирования и управление базой вне tmp_path не входят в задачу.

Ввод и вывод

stdin не используется. Печатать через print в stdout ничего не нужно. После вашего кода runner обнаруживает верхнеуровневую fixture client и готовые тесты настоящим pytest. stdin не используется; print/stdout не требуется. Функцию fixture не вызывают вручную — её жизненным циклом управляет pytest. Не запускайте pytest.main или Uvicorn самостоятельно.

Первый тест создаёт комнату и затем видит её в GET. Второй тест видит пустой каталог, даже если идёт после первого. После каждого теста его файл удалён, а старые подмены зависимостей восстановлены.