LatornLatorn
Войти

Задача курса

Тестирование ошибок FastAPI: отказ в бронировании

Курс «FastAPI для начинающих: API с базой данных и тестами» · урок «Сценарии ошибок и регрессии контракта»

Условие

Проверка отказов при бронировании

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

После отказа из-за пересечения первая бронь должна сохраниться, а новая запись не должна появиться. Если проверку конфликта в приложении отключить, соответствующий тест должен упасть. После восстановления приложения тот же тест должен пройти без ослабления ожиданий.

Что уже дано

Даны тот же самостоятельный API, безопасная fixture client и два теста комнат. Они исполняются перед вашим кодом и не являются TODO. В редакторе вы добавляете только тест бронирования; не импортируйте отсутствующие файлы main/conftest. Полный готовый 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]

Готовая fixture:

@pytest.fixture
def client(tmp_path):
    database_path = tmp_path / "test.db"
    test_engine = create_db_engine(f"sqlite:///{database_path.as_posix()}")
    previous_overrides = dict(app.dependency_overrides)
    try:
        Base.metadata.create_all(bind=test_engine)
        TestingSession = sessionmaker(bind=test_engine, autocommit=False, autoflush=False)
        def override_get_db():
            with TestingSession() as db:
                yield db
        app.dependency_overrides[get_db] = override_get_db
        with TestClient(app) as test_client:
            yield test_client
    finally:
        app.dependency_overrides.clear()
        app.dependency_overrides.update(previous_overrides)
        test_engine.dispose()
        database_path.unlink(missing_ok=True)

Два готовых теста (служебный force_failure здесь выключен):

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 через собственный test_submission.py. Parametrize, fixture и assertions выполняются самим pytest. Для проверки регрессии controller передаёт runner синтаксически корректный заменяющий код только готового обработчика либо изменение одного HTTP-ответа; тесты не переписываются. Вариант с if False сохраняет тело условного блока. После каждого запуска исходное поведение восстанавливается, файлы тестовых баз удаляются. Скрытый grader получает реальные pytest-отчёты и HTTP-наблюдения. Самостоятельно реализовывать runner, мутации или fixture не нужно.

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

Напишите одну тестовую функцию с @pytest.mark.parametrize для трёх случаев; имя может отличаться от test_booking_errors. Каждый случай использует готовую client и независимую подготовку.

  1. Не создавайте комнату, отправьте корректные aware-границы на отсутствующий id: ожидаются404 и detail="Room not found".
  2. Создайте комнату через POST /api/rooms и используйте полученный id. В запросе бронирования starts_at задайте без пояса, ends_at — с поясом: ожидаются422, detail как список и loc ["body","starts_at"]. Не привязывайтесь к тексту msg.
  3. Создайте комнату, сохраните первую корректную бронь (201), повторите тот же запрос: ожидаются409 и detail="Booking conflict". Затем GET по этому же пути должен дать200 и ровно первую публичную бронь, без изменения её полей.

Проверяйте точный статус каждого отказа, заявленный detail или loc и постусловие после409 обычными assert. Подготовку можно вынести в свой helper или fixture. Вместе с двумя готовыми тестами pytest должен собрать пять проверок; все проходят на исправном приложении. При отключённом конфликте должен упасть assert соответствующего случая на201 вместо409; другие случаи остаются успешными. Другие проверяемые регрессии — неверный status/detail/loc, добавленная, изменённая или удалённая запись после409. Ошибка импорта, fixture или синтаксиса не считается обнаруженным HTTP-дефектом. Не меняйте готовый API, готовую fixture и тесты комнат, не ослабляйте ожидания и не используйте skip/xfail.

Ввод и вывод

stdin не используется. Печатать через print в stdout ничего не нужно. После вашего кода проверка собирает верхнеуровневую параметризованную функцию и готовые тесты через pytest. stdin не используется; print/stdout не требуется. Запуск pytest, подготовку временных модулей и откат мутаций выполняет runner, не ваш тест.

Исправный API даёт пять passed. При замене условия конфликта на if False одна проверка получает201 вместо409 и падает на assert; остальные четыре проходят. После восстановления исходного обработчика снова проходят все пять.

Попробуйте решить

РешениеPython
Без регистрации · результат не сохраняется
Тестирование ошибок FastAPI: отказ в бронировании