Перейти к содержанию

Тестирование AI API

Шаг 1

До этого момента мы проверяли проект вручную. Запускали FastAPI, открывали Swagger UI или отправляли запросы через curl и самостоятельно сравнивали полученный результат с ожиданиями. Это было необходимо, так мы наглядно увидели, как пользователь создаёт сессию, получает потоковый ответ и загружает файл.

Однако по мере роста проекта ручная проверка становится всё сложнее. Представьте, что мы немного изменили ChatService. После этого нужно заново проверить создание сессии, обычный ответ, SSE, очистку, удаление, подтверждение и отмену действия. Затем то же самое придётся повторить для анализа обращений и файлов.

При этом, очень легко какой-нибудь сценарий забыть. Ещё сложнее запомнить все детали, а какой HTTP-статус должен вернуться, должен ли сохраниться ход в истории и удалился ли временный файл после ошибки. Поэтому нам нужны автоматические тесты.

Что такое автоматический тест?

Автоматический тест — это обычный Python-код, который сам подготавливает сценарий, вызывает нужную часть проекта и сравнивает результат с заранее записанными ожиданиями. Если все ожидания выполнены, тест проходит. Если хотя бы одно не выполнено, тест завершается ошибкой и показывает, где фактическое поведение отличилось от ожидаемого.

Обычно внутри теста можно увидеть три части. Сначала мы создаём объекты и входные данные. Затем вызываем проверяемый метод или отправляем HTTP-запрос. После этого записываем ожидания с помощью assert. Эту схему часто кратко описывают как «подготовка → действие → проверка». Мы будем использовать её во всех следующих шагах, где и напишем тесты.

Наше приложение состоит из двух очень разных частей. Первая — обычный детерминированный Python-код. Слово «детерминированный» означает, что при одних и тех же исходных данных мы можем ожидать один и тот же результат. Например FileValidator должен всегда отклонять неподдерживаемое расширение, HTTP-эндпоинт — возвращать заранее определённый статус, а очистка сессии — удалять историю. Такие факты можно проверять точными assert.

Вторая часть — ответ языковой модели. Он не является полностью предсказуемым, ведь на один вопрос модель может дать несколько разных, но одинаково полезных ответов. Если автотест будет требовать от реальной модели одну точную фразу, он может сегодня пройти, а завтра завершиться ошибкой из-за другого порядка слов. Такая проверка будет нестабильной. Поэтому в автотестах мы не будем обращаться к ProxyAPI, а мы заменим модель управляемым объектом, который вернёт заранее известный результат. После этого мы точно проверим свой код, а именно как он собрал запрос, разобрал ответ, изменил историю и сформировал HTTP-результат. Такие тесты будут быстрыми, бесплатными и повторяемыми.

В этом разделе мы будем двигаться от простого к сложному. Сначала отдельно проверим временное хранилище и валидатор файлов. Затем заменим LLMClient fake-объектом и проверим SupportRequestAnalyzer. После этого подключим TestClient и проверим полные HTTP-цепочки: обычный ответ, ошибку, чат-сесии, SSE и загрузку файла. В конце одна команда запустит все сценарии за несколько секунд.

Напомню, что ещё во втором модуле мы создали evals для проверки качества и поведения ответов модели. Сейчас мы будем заниматься другой задачей: автотесты проверят работу обычного Python-кода, сервисов и HTTP API. То есть evals отвечают на вопрос «насколько хорошо отвечает модель?», а тесты — «правильно ли работает наше приложение?».

Устанавливаем pytest

Для автотестов будем использовать pytest. Эта библиотека находит тестовые файлы и функции по их именам, запускает их и формирует общий отчёт. Если какой-то assert не выполнился, pytest покажет имя упавшего теста и значения, которые не совпали.

Убедитесь, что виртуальное окружение проекта активно, и установите пакет:

pip install pytest

После установки обновите requirements.txt тем же способом, который мы уже использовали:

pip freeze > requirements.txt

В корне проекта создайте каталог tests, а внутри него пустой __init__.py. Каталог будет хранить автотесты, а evals останется отдельным каталогом для проверок реальной модели:

ai-support-desk/
├── app/
├── evals/
├── tests/
│   └── __init__.py
├── main.py
└── requirements.txt

В этом шаге мы определили, какие проблемы решают автотесты, отделили их от evals и подготовили каталог tests. В следующем шаге напишем первые модульные тесты для UploadedFileStorage и FileValidator. На их примере разберём, как pytest находит тесты, откуда берётся временный каталог и как проверять ожидаемые ошибки.

Шаг 2

Начнём с кода, который не обращается ни к FastAPI, ни к ProxyAPI. Класс UploadedFileStorage получает поток байтов, сохраняет его во временном каталоге, следит за размером файла, а после обработки удаляет созданные данные. Здесь нет случайного ответа модели, поэтому результат каждого действия заранее известен и его удобно проверять с помощью обычных assert.

В проекте есть и другие части, которые можно проверять отдельно, например извлечение текста из документов, подготовку контекста, разбиение больших файлов и выполнение инструментов. Однако если подробно тестировать каждый созданный класс, раздел получится слишком большим. Поэтому на примере UploadedFileStorage и FileValidator мы сначала разберём основные возможности pytest, а затем перейдём к главным сценариям нашего AI-бэкенда.

Тест, который проверяет один класс или одну функцию отдельно от остального приложения, называют модульным тестом, или unit-тестом. Такой тест подготавливает входные данные, выполняет небольшое действие и сравнивает полученный результат с ожидаемым. Если проверка не пройдёт, нам будет проще понять, в какой части кода возникла проблема.

В этом шаге мы напишем пять тестов, они проверят успешное сохранение файла, обработку пустой и слишком большой загрузки, безопасную работу с именем и отклонение неподдерживаемого формата.

Создайте файл tests/test_upload_storage.py и добавьте импорты:

from io import BytesIO
from pathlib import Path

import pytest

from app.files.upload_storage import (
    UploadedFileEmptyError,
    UploadedFileSizeError,
    UploadedFileStorage,
)
from app.files.validation import (
    FileTypeError,
    FileValidationPolicy,
    FileValidator,
)

BytesIO позволяет создать поток байтов прямо в памяти. Для UploadedFileStorage такой объект выглядит как загруженный файл, хотя нам не пришлось заранее создавать его на диске. Path понадобится для аннотации временного каталога, а pytest предоставит средства запуска тестов и проверки ожидаемых исключений.

Добавьте первый тест:

def test_storage_creates_and_removes_file(tmp_path: Path) -> None:
    storage = UploadedFileStorage(root=tmp_path, max_size_bytes=100)
    source = BytesIO(b"hello")

    with storage.stage(source, "message.txt") as path:
        assert path.exists()
        assert path.name == "message.txt"
        assert path.read_bytes() == b"hello"

    assert not path.exists()

pytest найдёт этот тест автоматически, потому что имя файла и имя функции начинаются с test_. Обратите внимание на параметр tmp_path в аргументах тестовой функции, это встроенная фикстура pytest, которая предоставляет тесту временную папку. По имени параметра pytest понимает, что тесту нужен временный каталог, автоматически создаёт его перед запуском и передаёт путь в виде объекта Path. Внутри этого каталога тест может безопасно создавать файлы, не затрагивая настоящий проект. После завершения проверки pytest самостоятельно удалит временные данные.

В начале теста создаётся хранилище с лимитом в 100 байт и поток, содержащий слово hello. Внутри with проверяем, что файл действительно появился, получил ожидаемое имя и содержит исходные байты. Последний assert расположен уже после блока with. К этому моменту контекстный менеджер должен удалить временный файл. Так один тест проверяет полный жизненный цикл успешной загрузки: файл не просто создаётся, но и обязательно удаляется после использования.

Теперь рассмотрим ситуации, в которых загрузку нельзя продолжать. Пустой файл не содержит данных для анализа, а слишком большой способен превысить установленное ограничение. В обоих случаях UploadedFileStorage должен остановить работу и вернуть понятную ошибку.

Добавьте ещё два теста:

def test_storage_rejects_empty_file(tmp_path: Path) -> None:
    storage = UploadedFileStorage(root=tmp_path, max_size_bytes=100)

    with pytest.raises(UploadedFileEmptyError, match="Пустой файл"):
        with storage.stage(BytesIO(b""), "empty.txt"):
            pass

def test_storage_rejects_large_file_and_cleans_directory(tmp_path: Path) -> None:
    storage = UploadedFileStorage(root=tmp_path, max_size_bytes=3)

    with pytest.raises(UploadedFileSizeError, match="превышает"):
        with storage.stage(BytesIO(b"1234"), "large.txt"):
            pass

    assert list(tmp_path.iterdir()) == []

В первом тесте передаётся поток без единого байта. Конструкция pytest.raises() сообщает, что внутри блока мы ожидаем исключение UploadedFileEmptyError. Если исключение не появится или будет иметь другой тип, тест не пройдёт. Параметр match дополнительно проверяет, что в сообщении об ошибке есть слова «Пустой файл».

В следующем тесте лимит равен трём байтам, а мы пытаемся загрузить четыре. Ожидаем UploadedFileSizeError. При этом часть данных могла быть записана до обнаружения превышения, поэтому после ошибки отдельно проверяем содержимое tmp_path. Пустой список означает, что временный каталог загрузки был удалён и на диске не осталось незавершённого файла.

Следующий сценарий связан с именем, ведь имя загружаемого файла приходит от клиента, поэтому доверять ему нельзя. Вместо обычного report.txt клиент может передать строку ../../report.txt, содержащую части пути. А хранилище должно отбросить каталоги и оставить только имя файла.

Мы не будем отдельно вызывать внутренний метод _safe_name(). Проверим его работу через публичный метод stage(), которым пользуется остальное приложение:

def test_storage_keeps_only_file_name(tmp_path: Path) -> None:
    storage = UploadedFileStorage(root=tmp_path, max_size_bytes=100)

    with storage.stage(BytesIO(b"text"), "../../report.txt") as path:
        assert path.name == "report.txt"
        assert path.parent.parent == tmp_path

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

Осталось проверить неподдерживаемый формат. Здесь важно вспомнить разделение обязанностей. UploadedFileStorage только безопасно записывает полученные байты и контролирует их количество, но он не решает, подходит ли файл для анализа. Эту проверку выполняет FileValidator.

Создадим во временном каталоге файл data.json и передадим его валидатору. В нашем приложении такой формат не поддерживается, поэтому ожидаем FileTypeError:

def test_validator_rejects_unknown_extension(tmp_path: Path) -> None:
    file_path = tmp_path / "data.json"
    file_path.write_text("{}", encoding="utf-8")
    validator = FileValidator(FileValidationPolicy(allowed_root=tmp_path))

    with pytest.raises(FileTypeError, match="не поддерживается"):
        validator.validate(file_path)

Этот тест проверяет простое правило, что файл с неизвестным расширением не должен попасть к экстрактору и далее к модели.

Теперь давайте запустим тесты:

pytest tests/test_upload_storage.py

В отчете мы увидим следующее:

(.venv) ~/PycharmProjects/AI_Course_Developer/ai-support-desk git:[master] pytest tests/test_upload_storage.py
=================================== test session starts ===================================
platform darwin -- Python 3.13.7, pytest-9.1.1, pluggy-1.6.0
rootdir: /Users/permin0ff/PycharmProjects/AI_Course_Developer/ai-support-desk
plugins: anyio-4.14.2
collected 5 items

tests/test_upload_storage.py .....                                                  [100%]

==================================== 5 passed in 0.02s ====================================

В строке 5 passed мы видим, что все пять тестов завершились успешно. Значит, сохранение, ограничение размера, обработка имени и удаление временных файлов работают ожидаемым образом. В следующем шаге перейдём к тестированию SupportRequestAnalyzer и научимся проверять AI-сервис без настоящих запросов к ProxyAPI.

Шаг 3

В прошлом шаге мы проверяли обычные Python-классы, результат работы которых полностью известен заранее. Теперь перейдём к SupportRequestAnalyzer. Этот сервис обращается к модели, получает структурированный ответ, преобразует его в Pydantic-модель и дополнительно проверяет смысл полученных данных.

Если во время каждого теста действительно обращаться к ProxyAPI, результат будет зависеть от интернета, доступности провайдера, баланса и ответа выбранной модели. Кроме того, один и тот же запрос может получить немного разные формулировки. В таком случае будет сложно понять, что именно сломалось: наш код или внешний сервис.

Поэтому настоящий LLMClient в этих тестах заменим небольшим fake-объектом. Fake — это простая тестовая реализация настоящего объекта, она содержит нужный метод, но не отправляет запрос в сеть, а возвращает подготовленный нами LLMResult. Благодаря этому мы заранее знаем, какой ответ получит анализатор.

В этом шаге мы напишем три теста, и первым делом передадим правильный JSON и проверим успешный анализ. Затем вернём строку, которую невозможно разобрать как JSON. А в последнем тесте создадим более интересный сценарий, где ответ будет соответствовать схеме, но содержать номер заказа, которого не было в исходном обращении. Так мы проверим исправляющий запрос и ограничение количества попыток.

Создайте tests/test_support_request_analyzer.py. В начале добавьте импорты и fake-клиент:

import json

import pytest

from app.llm.client import JsonSchema, LLMResult, Message
from app.services.support_request_analyzer import (
    InvalidSupportRequestError,
    SupportRequestAnalyzer,
)

class FakeStructuredLLMClient:
    def __init__(self, results: list[LLMResult]) -> None:
        self._results = list(results)
        self.calls = 0
        self.last_messages: list[Message] | None = None

    def generate_structured(
        self,
        messages: list[Message],
        schema_name: str,
        schema: JsonSchema,
    ) -> LLMResult:
        self.calls += 1
        self.last_messages = messages
        return self._results.pop(0)

SupportRequestAnalyzer вызывает метод generate_structured(), поэтому fake-клиент реализует метод с таким же именем и такими же параметрами. Наследоваться от настоящего LLMClient ему не требуется, ведь анализатору важно только наличие нужного метода.

В конструктор передаётся список ответов, которые fake-клиент будет возвращать по очереди. Поле calls считает обращения к модели, а last_messages сохраняет последние полученные сообщения. Эти два значения позволят проверить не только итог анализа, но и то, как сервис взаимодействовал с клиентом.

Чтобы не повторять создание LLMResult, добавьте после класса небольшую функцию:

def build_llm_result(text: str) -> LLMResult:
    return LLMResult(
        text=text,
        refusal=None,
        tool_calls=(),
        model="fake-model",
        finish_reason="stop",
        prompt_tokens=10,
        completion_tokens=5,
        total_tokens=15,
        response_id="fake-response-id",
    )

Функция build_llm_result() создаёт обычный результат клиента с тестовыми метаданными. В каждом сценарии нам останется передать только нужный текст. Значения fake-model и fake-response-id специально показывают, что ответ был подготовлен тестом, а не получен от провайдера.

Теперь проверим успешный сценарий. Представим, что модель правильно определила категорию, товар и номер заказа и вернула JSON, соответствующий нашей схеме. Мы передадим этот текст fake-клиенту, вызовем обычный метод analyze() и посмотрим, как сервис обработает результат:

def test_analyzer_parses_structured_result() -> None:
    response_text = json.dumps(
        {
            "category": "technical_issue",
            "order_id": "8041",
            "product": "кофемашина",
            "problem_description": "не включается",
            "urgency": "normal",
            "missing_fields": [],
            "confidence": 0.95,
        },
        ensure_ascii=False,
    )
    fake_client = FakeStructuredLLMClient([build_llm_result(response_text)])
    analyzer = SupportRequestAnalyzer(fake_client)

    result = analyzer.analyze("Заказ 8041, кофемашина не включается.")

    assert result.analysis.order_id == "8041"
    assert result.analysis.category == "technical_issue"
    assert result.repair_attempts == 0
    assert fake_client.calls == 1
    assert fake_client.last_messages is not None
    assert "8041" in fake_client.last_messages[-1]["content"]

После вызова проверяем полученный объект result. В нём должны находиться номер заказа 8041 и категория technical_issue. Значение repair_attempts == 0 означает, что первый ответ прошёл все проверки и исправляющий запрос не понадобился. А fake_client.calls == 1 подтверждает, что анализатор обратился к клиенту только один раз.

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

Но провайдер может вернуть ответ, который невозможно преобразовать в нужную модель данных. Во втором тесте fake-клиент вместо JSON вернёт строку not-json. SupportRequestAnalyzer не должен передавать такой результат дальше. Мы ожидаем понятное прикладное исключение InvalidSupportRequestError:

def test_analyzer_rejects_invalid_json() -> None:
    fake_client = FakeStructuredLLMClient([build_llm_result("not-json")])
    analyzer = SupportRequestAnalyzer(fake_client)

    with pytest.raises(InvalidSupportRequestError, match="не соответствует контракту"):
        analyzer.analyze("Заказ 8041 не работает.")

    assert fake_client.calls == 1

pytest.raises() подтверждает, что ошибочный ответ был остановлен внутри сервиса. Дополнительно проверяем количество обращений: анализатор получил один ответ и не пытался бесконечно исправлять данные, которые даже не удалось разобрать.

Строгая схема обычно не позволяет модели вернуть произвольную строку. Тем не менее наш код должен безопасно обработать нарушение ожидаемого контракта. Тест позволяет проверить эту редкую ветку без попыток специально получить неправильный ответ от настоящей модели.

В последнем сценарии JSON будет правильным по форме, но неправильным по смыслу. В обращении пользователя указан заказ 8041, а fake-клиент вернёт номер 9999. Для Pydantic оба значения являются обычными строками, поэтому такое противоречие обнаружит уже наша смысловая проверка.

Добавьте третий тест:

def test_analyzer_limits_repair_attempts() -> None:
    wrong_text = json.dumps(
        {
            "category": "technical_issue",
            "order_id": "9999",
            "product": None,
            "problem_description": "не включается",
            "urgency": "normal",
            "missing_fields": ["product"],
            "confidence": 0.7,
        },
        ensure_ascii=False,
    )
    fake_client = FakeStructuredLLMClient(
        [
            build_llm_result(wrong_text),
            build_llm_result(wrong_text),
        ]
    )
    analyzer = SupportRequestAnalyzer(fake_client)

    with pytest.raises(InvalidSupportRequestError, match="смысловую проверку"):
        analyzer.analyze("Заказ 8041 не включается.")

    assert fake_client.calls == 2

В FakeStructuredLLMClient мы заранее передаём два ответа. Первый возвращается во время обычного анализа. После обнаружения неправильного номера SupportRequestAnalyzer формирует исправляющий запрос и снова вызывает generate_structured(). Тогда fake-клиент возвращает второй подготовленный ответ. Мы намеренно оставили в нём ту же ошибку, чтобы проверить остановку после одной попытки исправления. Условие fake_client.calls == 2 подтверждает, что клиент был вызван один раз для первоначального анализа и один раз для исправления.

Теперь запустите все созданные тесты из корня проекта:

pytest

pytest автоматически найдёт тесты из прошлого и этого шага, и в итоговой строке должно появиться 8 passed: пять проверок UploadedFileStorage и FileValidator, а также три проверки SupportRequestAnalyzer.

(.venv) ~/PycharmProjects/AI_Course_Developer/ai-support-desk git:[master] pytest
======================================= test session starts ========================================
platform darwin -- Python 3.13.7, pytest-9.1.1, pluggy-1.6.0
rootdir: /Users/permin0ff/PycharmProjects/AI_Course_Developer/ai-support-desk
plugins: anyio-4.14.2
collected 8 items

tests/test_support_request_analyzer.py ...                                                   [ 37%]
tests/test_upload_storage.py .....                                                           [100%]

======================================== 8 passed in 0.62s =========================================

До этого мы тестировали отдельные Python-классы напрямую. Теперь, когда их основное поведение проверено, можно двигаться дальше. В следующем шаге мы отправим запрос через FastAPI и проверим весь путь от HTTP-запроса до ответа, который получит клиент нашего AI-бэкенда.

Шаг 4

В предыдущих шагах мы вызывали UploadedFileStorage и SupportRequestAnalyzer напрямую из Python. Но пользователь нашего AI-бэкенда обращается не к этим классам, а к HTTP-эндпоинтам. Поэтому теперь нужно проверить весь путь запроса через FastAPI.

Во время такой проверки вместе работают сразу несколько частей приложения: маршрут принимает запрос, Pydantic проверяет его тело, зависимость передаёт нужный сервис, обработчик ошибок формирует единый JSON, а middleware добавляет Request ID. Тест, который проверяет взаимодействие нескольких частей приложения, называют интеграционным.

Настоящий AI-сервис нам по-прежнему не понадобится. Мы заменим его управляемым объектом, но оставим настоящими FastAPI-приложение, маршруты, Pydantic-модели и обработчики ошибок. Благодаря этому запрос не уйдёт в ProxyAPI, а вся HTTP-логика будет выполнена так же, как при обычном запуске.

В этом шаге мы напишем четыре теста, и сначала убедимся, что API запускается и отвечает на /health. Затем проверим успешный анализ обращения, отклонение неправильного тела запроса и безопасный ответ при ошибке AI-сервиса.

Отправлять запросы будет TestClient из FastAPI. Он похож на обычный HTTP-клиент и предоставляет методы get(), post() и delete(). При этом запускать Uvicorn и открывать настоящий сетевой порт не нужно, так как запрос передаётся FastAPI-приложению прямо внутри теста.

Нескольким тестам понадобится одинаковая подготовка. Чтобы не повторять её в каждом файле, создайте tests/conftest.py:

from collections.abc import Iterator
from types import SimpleNamespace

import pytest
from fastapi import FastAPI
from fastapi.testclient import TestClient

from app.api.dependencies import get_container
from app.api.main import create_api

@pytest.fixture
def fake_container() -> SimpleNamespace:
    return SimpleNamespace()

@pytest.fixture
def application() -> Iterator[FastAPI]:
    app = create_api()
    yield app
    app.dependency_overrides.clear()

@pytest.fixture
def client(
    application: FastAPI,
    fake_container: SimpleNamespace,
    monkeypatch: pytest.MonkeyPatch,
) -> Iterator[TestClient]:
    monkeypatch.setenv("LLM_API_KEY", "test-key")
    application.dependency_overrides[get_container] = lambda: fake_container

    with TestClient(application) as test_client:
        yield test_client

Фикстура — это функция, которая заранее подготавливает объект для теста. pytest видит имя фикстуры в параметрах тестовой функции, сам вызывает её и передаёт полученный результат. В нашем случае одна фикстура создаёт пустой контейнер, вторая — FastAPI-приложение, а третья — готовый TestClient.

SimpleNamespace используется как простой тестовый контейнер. Перед каждым сценарием мы сможем положить в него только нужный сервис, например request_analyzer или chat_service. Создавать настоящий ApplicationContainer со всеми объектами проекта не требуется.

Через dependency_overrides мы заменяем результат зависимости get_container. Поэтому внутри эндпоинта FastAPI передаст наш fake_container, а не контейнер с реальными AI-сервисами. Сам маршрут при этом не меняется.

monkeypatch.setenv() временно задаёт значение LLM_API_KEY. Оно нужно только для создания настроек во время запуска lifespan. Это не настоящий ключ, и в сеть он не отправляется. После завершения теста pytest автоматически восстановит окружение.

Конструкция with TestClient(application) запускает приложение перед тестом и закрывает его после. Таким образом, каждый тест получает работающий FastAPI и может отправлять ему обычные HTTP-запросы.

Теперь создайте файл tests/test_support_requests_api.py. Начнём с самого простого запроса — GET /health. Этот тест проверит, что приложение успешно создаётся, маршрут подключён и сервер возвращает ожидаемый ответ:

from types import SimpleNamespace
from unittest.mock import Mock

from fastapi.testclient import TestClient

from app.models.support_request import SupportRequestAnalysis
from app.services.support_request_analyzer import (
    SupportRequestAnalyzerError,
    SupportRequestResult,
)

def test_health(client: TestClient) -> None:
    response = client.get("/health")

    assert response.status_code == 200
    assert response.json() == {"status": "ok"}

В переменной response находится тот же результат, который получил бы обычный клиент: HTTP-статус, заголовки и тело. Мы ожидаем статус 200, а метод json() должен вернуть словарь {"status": "ok"}. Если маршрут не подключён или его ответ случайно изменится, тест не пройдёт.

Далее проверим успешный анализ обращения. В прошлой лекции мы уже протестировали внутреннюю работу SupportRequestAnalyzer, поэтому здесь заменим его объектом Mock и заранее зададим готовый SupportRequestResult. Нас интересует путь от входного JSON до HTTP-ответа:

def test_analyze_support_request_returns_contract(
    client: TestClient,
    fake_container: SimpleNamespace,
) -> None:
    analyzer = Mock()
    analyzer.analyze.return_value = SupportRequestResult(
        analysis=SupportRequestAnalysis(
            category="technical_issue",
            order_id="8041",
            product="кофемашина",
            problem_description="не включается",
            urgency="normal",
            missing_fields=[],
            confidence=0.95,
        ),
        model="fake-model",
        prompt_id="support_request_analysis",
        prompt_version="1.0.0",
        elapsed_seconds=0.01,
        prompt_tokens=10,
        completion_tokens=5,
        total_tokens=15,
        response_id="fake-response-id",
        repair_attempts=0,
    )
    fake_container.request_analyzer = analyzer

    response = client.post(
        "/support-requests/analyze", json={"text": "Заказ 8041 не включается."}
    )

    assert response.status_code == 200
    body = response.json()
    assert body["analysis"]["order_id"] == "8041"
    assert body["analysis"]["category"] == "technical_issue"
    assert body["usage"]["total_tokens"] == 15
    analyzer.analyze.assert_called_once_with("Заказ 8041 не включается.")

Mock похож на созданный ранее fake-клиент, но его ответ настраивается прямо в тесте. Через return_value мы указываем результат метода analyze(). Затем помещаем этот объект в fake_container и отправляем настоящее HTTP-обращение к /support-requests/analyze.

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

Следующий запрос будет неправильным, и вместо текста обращения мы отправим пустую строку. Pydantic должен остановить его ещё до вызова анализатора и вернуть статус 422:

def test_analyze_support_request_rejects_empty_text(
    client: TestClient,
) -> None:
    response = client.post("/support-requests/analyze", json={"text": ""})

    assert response.status_code == 422
    body = response.json()
    assert body["error"]["code"] == "validation_error"
    assert body["error"]["request_id"] == response.headers["X-Request-ID"]

Мы проверяем не только статус, но и единый код ошибки validation_error. Кроме того, Request ID в JSON должен совпадать со значением заголовка X-Request-ID. Тогда пользователь сможет сообщить идентификатор запроса, а разработчик — найти связанные с ним записи в логах.

В последнем тесте входные данные будут правильными, но сам AI-сервис завершится ошибкой. Чтобы воспроизвести этот сценарий, настроим Mock через side_effect: при вызове analyze() он выбросит SupportRequestAnalyzerError.

Эндпоинт должен преобразовать эту прикладную ошибку в статус 502 и безопасное сообщение. Технический текст исключения и значение API-ключа не должны попасть в ответ:

def test_analyze_support_request_hides_service_error(
    client: TestClient,
    fake_container: SimpleNamespace,
) -> None:
    analyzer = Mock()
    analyzer.analyze.side_effect = SupportRequestAnalyzerError("Техническая ошибка ProxyAPI")
    fake_container.request_analyzer = analyzer

    response = client.post(
        "/support-requests/analyze", json={"text": "Кофемашина не включается."}
    )

    assert response.status_code == 502
    body = response.json()
    assert body["error"]["code"] == "support_analysis_failed"
    assert body["error"]["message"] == "Не удалось проанализировать обращение."
    assert "ProxyAPI" not in response.text
    assert "test-key" not in response.text

Первые три проверки подтверждают публичный контракт ошибки: статус 502, код support_analysis_failed и понятное сообщение для пользователя. Два последних assert отдельно убеждаются, что внутреннее упоминание ProxyAPI и тестовый ключ отсутствуют в ответе.

Каждая проверка отвечает за отдельную часть HTTP-контракта. Поэтому, если тест завершится ошибкой, будет проще понять, что именно изменилось: статус, тело ответа, Request ID, переданные сервису данные или обработка исключения.

Теперь снова запустите все созданные тесты:

pytest

В итоговой строке должно появиться 12 passed. К пяти тестам временных файлов и трём тестам SupportRequestAnalyzer добавились четыре проверки FastAPI. Теперь автоматически проверяются запуск приложения, успешный анализ обращения, Pydantic-валидация, работа Request ID и безопасная обработка ошибки AI-сервиса.

До сих пор каждый проверяемый запрос был независимым, и в следующем шаге мы перейдём к чату, где состояние сохраняется между обращениями. Тем самым мы проверим изоляцию сессий, управление историей и потоковую передачу ответа через SSE.

Шаг 5

Анализ обращения из предыдущего шага не хранит состояние между запросами, так как клиент отправляет текст и получает готовый результат. Но чат работает сложнее, и сначала сервер создаёт сессию, затем сохраняет внутри неё историю разговора. Эту историю можно прочитать, очистить или удалить вместе со всей сессией.

Также нам нужно отдельно проверить потоковый эндпоинт. Обычный эндпоинт возвращает один готовый JSON после завершения всей обработки, а потоковый постепенно отправляет клиенту SSE-события и сначала приходят отдельные части ответа модели, а в конце — событие об успешном завершении или возникшей ошибке. При этом первые показанные фрагменты ещё не считаются законченным ответом. Вопрос пользователя и ответ assistant должны сохраняться в истории только после успешного завершения всего потока. Если во время генерации произойдёт ошибка, незаконченный ответ не должен попасть в историю диалога.

Настоящая модель для таких тестов не нужна. Вместо BoundedToolLoop создадим простую тестовую реализацию, которая вернёт два заранее известных фрагмента текста и итоговое событие. ChatService будет работать с ней так же, как с настоящим циклом инструментов, но результат останется полностью предсказуемым.

Создайте tests/fakes.py:

from collections.abc import Generator

from app.llm.client import Message
from app.services.tool_loop import (
    ToolLoopError,
    ToolLoopResult,
    ToolLoopStreamCompleted,
    ToolLoopStreamEvent,
    ToolLoopTextDelta,
)

class FakeToolLoop:
    def __init__(self) -> None:
        self.calls: list[list[Message]] = []

    def stream(self, messages: list[Message]) -> Generator[ToolLoopStreamEvent, None, None]:
        self.calls.append(messages)
        parts = ("Ваш заказ ", "находится в пути.")

        for part in parts:
            yield ToolLoopTextDelta(text=part)

        yield ToolLoopStreamCompleted(
            result=ToolLoopResult(
                text="".join(parts),
                model="fake-model",
                prompt_tokens=10,
                completion_tokens=5,
                total_tokens=15,
                response_ids=("fake-response-id",),
                model_steps=1,
                tool_calls_count=0,
                tool_names=(),
                audit_records=(),
                confirmation=None,
            )
        )

class FailingAfterDeltaToolLoop:
    def stream(self, messages: list[Message]) -> Generator[ToolLoopStreamEvent, None, None]:
        yield ToolLoopTextDelta(text="Начало ответа")
        raise ToolLoopError("Искусственный сбой")

class UnusedSummaryService:
    def summarize(self, previous_summary: str | None, messages: list[Message]) -> str:
        raise AssertionError("Summary не должно вызываться")

FakeToolLoop сохраняет полученные сообщения в calls, а затем выдаёт две части: «Ваш заказ » и «находится в пути.». После них возвращается ToolLoopStreamCompleted с полным текстом и тестовыми метаданными. Так мы сможем проверить настоящий ChatService, не обращаясь к модели.

FailingAfterDeltaToolLoop предназначен для другого сценария. Он успевает передать начало ответа, а затем выбрасывает ToolLoopError. С его помощью мы проверим, что клиент получит событие об ошибке, а незавершённый ответ не сохранится в истории.

UnusedSummaryService не выполняет суммаризацию. История в тестах будет короткой и полностью поместится в установленный лимит и если ChatService неожиданно попробует создать резюме, этот объект сразу завершит тест ошибкой и покажет лишний вызов.

Теперь дополните импорты в tests/conftest.py:

from app.chat_history import SlidingWindowHistory, TokenCounter
from app.services.chat_service import ChatService
from tests.fakes import FakeToolLoop, UnusedSummaryService

В конец того же файла добавьте фикстуру, которая перед каждым тестом создаёт новый ChatService:

@pytest.fixture
def chat_service() -> ChatService:
    return ChatService(
        tool_loop=FakeToolLoop(),
        history_policy=SlidingWindowHistory(
            token_counter=TokenCounter(),
            input_token_budget=6_000,
        ),
        summary_service=UnusedSummaryService(),
    )

Фикстура создаёт отдельный ChatService перед каждым тестом. Это важно, потому что именно внутри сервиса находится словарь чат-сессий. Если использовать один объект несколько раз, данные предыдущего теста останутся в памяти и смогут повлиять на следующий сценарий.

Создайте tests/test_chat_api.py и добавьте импорты:

from types import SimpleNamespace

from fastapi.testclient import TestClient

from app.chat_history import SlidingWindowHistory, TokenCounter
from app.services.chat_service import ChatService
from tests.fakes import (
    FailingAfterDeltaToolLoop,
    UnusedSummaryService,
)

Сначала проверим изоляцию диалогов. Создадим две сессии, но сообщение отправим только в первую. После ответа запросим состояние обеих сессий. В первой должны появиться два сообщения — реплика пользователя и ответ assistant. Вторая сессия должна остаться пустой:

def test_chat_sessions_are_isolated(
    client: TestClient,
    fake_container: SimpleNamespace,
    chat_service: ChatService,
) -> None:
    fake_container.chat_service = chat_service

    first_id = client.post("/chat/sessions").json()["session_id"]
    second_id = client.post("/chat/sessions").json()["session_id"]

    response = client.post(
        f"/chat/sessions/{first_id}/messages",
        json={"message": "Где мой заказ?"},
    )

    assert response.status_code == 200
    assert response.json()["text"] == "Ваш заказ находится в пути."

    first = client.get(f"/chat/sessions/{first_id}")
    second = client.get(f"/chat/sessions/{second_id}")

    assert first.json()["history_messages"] == 2
    assert second.json()["history_messages"] == 0

Сценарий проходит через настоящие HTTP-эндпоинты и настоящий ChatService. Тестовая реализация заменяет только получение ответа модели. Проверка history_messages == 2 показывает, что законченный ход сохранился в первой сессии, а значение 0 подтверждает, что вторая сессия не изменилась.

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

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

def test_clear_and_delete_chat_session(
    client: TestClient,
    fake_container: SimpleNamespace,
    chat_service: ChatService,
) -> None:
    fake_container.chat_service = chat_service
    session_id = client.post("/chat/sessions").json()["session_id"]
    client.post(
        f"/chat/sessions/{session_id}/messages",
        json={"message": "Где мой заказ?"},
    )

    clear_response = client.delete(f"/chat/sessions/{session_id}/messages")
    assert clear_response.status_code == 204
    session = client.get(f"/chat/sessions/{session_id}")
    assert session.json()["history_messages"] == 0

    delete_response = client.delete(f"/chat/sessions/{session_id}")
    assert delete_response.status_code == 204
    assert client.get(f"/chat/sessions/{session_id}").status_code == 404

Статус 204 подтверждает, что операция выполнена и отдельного тела ответа нет. После первого DELETE чтение сессии по-прежнему работает и показывает пустую историю. После второго DELETE тот же идентификатор уже не существует. Благодаря этому тесту очистка и удаление не смогут случайно начать выполнять одно и то же действие.

Далее перейдём к потоковой передаче. Успешный SSE-ответ должен содержать два события delta с частями текста и итоговое событие completed. Только после completed законченная пара сообщений должна появиться в истории. Для отправки запроса будем использовать метод TestClient.stream(). Он открывает потоковое соединение с эндпоинтом. Метод iter_text() позволяет последовательно получить отправленные сервером SSE-события и собрать их текст для дальнейшей проверки.

def test_stream_returns_deltas_and_completed_event(
    client: TestClient,
    fake_container: SimpleNamespace,
    chat_service: ChatService,
) -> None:
    fake_container.chat_service = chat_service
    session_id = client.post("/chat/sessions").json()["session_id"]

    with client.stream(
        "POST",
        f"/chat/sessions/{session_id}/messages/stream",
        json={"message": "Где мой заказ?"},
    ) as response:
        body = "".join(response.iter_text())

    assert response.status_code == 200
    assert response.headers["content-type"].startswith(
        "text/event-stream"
    )
    assert body.count("event: delta") == 2
    assert "Ваш заказ " in body
    assert "находится в пути." in body
    assert "event: completed" in body
    session = client.get(f"/chat/sessions/{session_id}")
    assert session.json()["history_messages"] == 2

Сначала склеиваем все части, которые вернул iter_text(). Нам не важно, какими фрагментами их передал сам тестовый клиент, ведь мы проверяем именно договор нашего API: тип содержимого text/event-stream, два события delta, обе части текста и событие completed.

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

Осталось проверить ошибку после начала потока. FailingAfterDeltaToolLoop сначала передаст один фрагмент, а затем выбросит исключение. К этому моменту HTTP-ответ уже начался, поэтому заменить статус 200 на 502 невозможно. Вместо этого клиент должен получить SSE-событие error.

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

def test_stream_error_does_not_save_history(
    client: TestClient,
    fake_container: SimpleNamespace,
) -> None:
    fake_container.chat_service = ChatService(
        tool_loop=FailingAfterDeltaToolLoop(),
        history_policy=SlidingWindowHistory(
            token_counter=TokenCounter(),
            input_token_budget=6_000,
        ),
        summary_service=UnusedSummaryService(),
    )
    session_id = client.post("/chat/sessions").json()["session_id"]

    with client.stream(
        "POST",
        f"/chat/sessions/{session_id}/messages/stream",
        json={"message": "Дайте длинный ответ."},
    ) as response:
        body = "".join(response.iter_text())

    assert response.status_code == 200
    assert "event: delta" in body
    assert "event: error" in body
    assert "chat_stream_failed" in body
    session = client.get(f"/chat/sessions/{session_id}")
    assert session.json()["history_messages"] == 0

В полученном тексте проверяем наличие первого delta, события error и программного кода chat_stream_failed. После этого запрашиваем состояние сессии. Значение history_messages == 0 подтверждает, что вопрос и незавершённый ответ не были сохранены.

Теперь запустите все созданные тесты:

pytest

В итоговой строке должно быть указано 16 passed. Это означает, что успешно прошли не только четыре новых сценария чата, но и все проверки из предыдущих лекций. Теперь тесты контролируют работу с временными файлами, анализ обращений, HTTP API, изоляцию чат-сессий и передачу ответа через SSE.

В следующем шаге проверим последний важный сценарий нашего AI-бэкенда — загрузку документа через HTTP-эндпоинт. И убедимся, что файл передаётся в DocumentAIService, ошибки получают правильные HTTP-статусы, а временные данные удаляются после обработки.

Шаг 6

В прошлом разделе мы вручную загружали документы через Swagger UI и смотрели ответы AI-бэкенда. Теперь автоматизируем эту проверку, где TestClient отправит файл и вопрос в формате multipart/form-data, эндпоинт сохранит полученные байты во временном каталоге и передаст путь в DocumentAIService.

Настоящий DocumentAIService здесь использовать не будем. Иначе один тест одновременно запускал бы извлечение текста, подготовку контекста и обращение к модели. Мы заменим только AI-сервис, а UploadedFileStorage оставим настоящим. Так сможем проверить именно HTTP-загрузку и удаление временных данных.

Создайте tests/test_files_api.py. Сначала добавьте импорты и fake-сервис:

from pathlib import Path
from types import SimpleNamespace
from unittest.mock import Mock

from fastapi.testclient import TestClient

from app.files.upload_storage import UploadedFileStorage
from app.files.validation import FileKind
from app.models.document_analysis import (
    DocumentAnalysis,
    DocumentFact,
)
from app.services.document_ai_service import (
    DocumentAIInputError,
    DocumentAIRequest,
)
from app.services.document_content_analyzer import DocumentAIResult

class FakeDocumentAIService:
    def __init__(self) -> None:
        self.received_request: DocumentAIRequest | None = None
        self.received_text: str | None = None

    def analyze(self, request: DocumentAIRequest) -> DocumentAIResult:
        self.received_request = request
        self.received_text = request.source_path.read_text(encoding="utf-8")

        return DocumentAIResult(
            analysis=DocumentAnalysis(
                summary="Краткое резюме.",
                answer="Ответ найден в документе.",
                key_facts=[
                    DocumentFact(
                        text="Факт из документа.",
                        reference="document",
                    )
                ],
                limitations=[],
                requires_human_review=False,
            ),
            source_name=request.source_path.name,
            source_kind=FileKind.TEXT,
            warnings=(),
            model="fake-model",
            prompt_id="document_analysis",
            prompt_version="1.0.0",
            elapsed_seconds=0.01,
            prompt_tokens=10,
            completion_tokens=5,
            total_tokens=15,
            response_id="fake-response-id",
        )

FakeDocumentAIService получает обычный DocumentAIRequest и читает файл по переданному source_path. Если эндпоинт не сохранил загрузку, чтение сразу завершится ошибкой. Полученный запрос и прочитанный текст сохраняются в полях received_request и received_text, чтобы позднее тест мог их проверить.

После чтения fake-сервис возвращает готовый DocumentAIResult с тестовыми значениями. Модель при этом не запускается. Нам нужен полноценный результат только для того, чтобы проверить, как эндпоинт преобразует его в публичный JSON.

Сначала проверим успешную загрузку. Отправим небольшой текстовый файл и вопрос. Fake-сервис должен прочитать его содержимое, а клиент — получить статус 200 и результат анализа. После завершения запроса временный файл должен быть удалён:

def test_file_upload_returns_analysis_and_removes_file(
    client: TestClient,
    fake_container: SimpleNamespace,
    tmp_path: Path,
) -> None:
    service = FakeDocumentAIService()
    fake_container.upload_storage = UploadedFileStorage(root=tmp_path, max_size_bytes=1_000)
    fake_container.document_ai_service = service

    response = client.post(
        "/files/analyze",
        files={
            "file": (
                "request.txt",
                b"Order 8041 is damaged.",
                "text/plain",
            )
        },
        data={"question": "Что случилось?"},
    )

    assert response.status_code == 200
    body = response.json()
    assert body["source_name"] == "request.txt"
    assert body["source_kind"] == "text"
    assert body["analysis"]["answer"] == "Ответ найден в документе."
    assert service.received_text == "Order 8041 is damaged."
    assert service.received_request is not None
    assert service.received_request.declared_media_type == "text/plain"
    assert not service.received_request.source_path.exists()

Параметр files содержит имя, байты и MIME type файла, а в data передаётся обычное поле формы с вопросом. Вместе они образуют запрос multipart/form-data, такой же, как при загрузке через Swagger UI.

В ответе проверяем имя, формат и текст результата. Затем смотрим, что fake-сервис получил исходное содержимое и MIME type text/plain. Последний assert выполняется уже после завершения эндпоинта и подтверждает, что временный путь больше не существует.

Следующий тест проверит файл, превышающий ограничение. Установим лимит в три байта и отправим четыре. Хранилище должно остановить загрузку, удалить временные данные и вернуть статус 413. Метод analyze() при этом вызываться не должен:

def test_large_file_returns_413_before_analysis(
    client: TestClient,
    fake_container: SimpleNamespace,
    tmp_path: Path,
) -> None:
    service = Mock()
    fake_container.upload_storage = UploadedFileStorage(root=tmp_path, max_size_bytes=3)
    fake_container.document_ai_service = service

    response = client.post(
        "/files/analyze",
        files={"file": ("large.txt", b"1234", "text/plain")},
        data={"question": "Что в файле?"},
    )

    assert response.status_code == 413
    assert response.json()["error"]["code"] == "file_too_large"
    service.analyze.assert_not_called()
    assert list(tmp_path.iterdir()) == []

Для этого сценария Mock не требуется готовый результат, потому что до анализа выполнение доходить не должно. Проверка assert_not_called() подтверждает, что слишком большой файл не был передан AI-сервису. Пустой список содержимого tmp_path показывает, что после ошибки на диске не осталось временного каталога.

В последнем тесте загрузка пройдёт, но дальнейшая проверка файла завершится DocumentAIInputError. Мы настроим Mock так, чтобы он выбросил это исключение, а затем проверим, что эндпоинт вернул статус 422 и программный код invalid_file:

def test_invalid_file_returns_422(
    client: TestClient,
    fake_container: SimpleNamespace,
    tmp_path: Path,
) -> None:
    service = Mock()
    service.analyze.side_effect = DocumentAIInputError("Формат .json не поддерживается.")
    fake_container.upload_storage = UploadedFileStorage(root=tmp_path, max_size_bytes=1_000)
    fake_container.document_ai_service = service

    response = client.post(
        "/files/analyze",
        files={
            "file": ("data.json", b"{}", "application/json")
        },
        data={"question": "Что в файле?"},
    )

    assert response.status_code == 422
    assert response.json()["error"]["code"] == "invalid_file"
    service.analyze.assert_called_once()

Сам FileValidator мы уже проверили в лекции 5.6.2. Здесь нас интересует поведение HTTP-слоя: прикладная ошибка должна превратиться в понятный клиенту ответ. Проверка assert_called_once() дополнительно подтверждает, что AI-сервис действительно был вызван, то есть ошибка возникла после сохранения загрузки.

Теперь код последнего тестового файла готов. Сначала запустите только три проверки загрузки документов:

pytest -q tests/test_files_api.py

В итоговой строке должно появиться 3 passed. Это означает, что файл успешно проходит от HTTP-запроса до AI-сервиса, превышение размера останавливает обработку, прикладная ошибка получает правильный HTTP-статус, а временные данные удаляются.

После этого запустите все тесты проекта одной командой:

pytest -q

И мы получаем краткий вывод, где в результате должно быть указано:

19 passed

А каждая точка в кратком отчёте обозначает один успешно выполненный тест. Если какая-либо проверка не пройдёт, pytest покажет имя теста, строку с ошибкой и значения, которые не совпали с ожиданиями.

В зависимости от установленных версий FastAPI и Starlette рядом с результатом может появиться предупреждение, связанное с TestClient. Предупреждение не означает, что тест завершился ошибкой. Если в конце указано 19 passed, все наши сценарии выполнены успешно.

Но наш набор из 19 тестов нельзя считать полным покрытием всего проекта. Я намеренно выбрал основные сценарии, чтобы познакомить вас с тестированием AI-бэкенда и не превращать раздел в отдельный большой курс по pytest.

Например, мы пока не проверили:

  • суммаризацию длинной истории и точное соблюдение бюджета
  • выбор и выполнение каждого зарегистрированного инструмента
  • HTTP-операции подтверждения и отмены подготовленного действия
  • все варианты извлечения данных из PDF, DOCX, Markdown, таблиц и изображений
  • сравнение нескольких документов
  • одновременную обработку большого количества запросов
  • поведение приложения при настоящем отключении клиента во время SSE-потока

Если хотите дополнительно потренироваться, попробуйте самостоятельно добавить несколько тестов. Например можно проверить получение неизвестной чат-сессии со статусом 404, повторное подтверждение действия со статусом 409, отклонение пустого сообщения, отмену подготовленной заявки или обработку неподдерживаемого типа файла. Для этого не обязательно писать всё с нуля, а используйте уже созданные фикстуры, fake-объекты и TestClient.

Важно понимать, что автотесты не отвечают на вопрос, насколько хорошо настоящая модель понимает пользователя. В тестах мы намеренно заменяли модель заранее подготовленными результатами, поэтому они проверяют наш Python-код, сервисы, состояние сессий и HTTP-контракты.

Качество ответов модели проверяется отдельно с помощью evals и ручных сценариев через ProxyAPI. Таким образом, эти два подхода дополняют друг друга:

В этом разделе мы проверили сохранение и удаление временных файлов, ограничения загрузки, разбор структурированного ответа, исправляющие запросы, HTTP-статусы, безопасные сообщения об ошибках, изоляцию чат-сессий и формат SSE. Теперь большая часть важных ошибок в обычном коде обнаруживается за несколько секунд и без расходования баланса ProxyAPI.

На этом мы завершили базовое тестирование AI-бэкенда, а вместе с ним и пятый модуль. В шестом модуле мы не будем добавлять новые крупные AI-функции, а вместо этого подготовим созданный AI Backend к более надёжной работе и добавим ресурсные ограничения, продолжим работу с безопасностью и сделаем поведение приложения наблюдаемым.

Шаг 7

Зачем в автоматических тестах AI-сервиса использовать фейковый LLM-клиент?

Правильный ответ
  • ✅ Чтобы получать предсказуемые ответы без сети и расходов

Шаг 8

Для чего используется application.dependency_overrides в тестах FastAPI?

Правильный ответ
  • ✅ Чтобы заменить настоящую зависимость контролируемой тестовой реализацией

Шаг 9

Что важно проверить в тесте ошибки AI-эндпоинта?

Выберите все правильные варианты.

Правильный ответ
  • ✅ Отсутствие внутреннего сообщения провайдера и секретов
  • ✅ Стабильный публичный код ошибки
  • ✅ Ожидаемый HTTP-статус
  • ✅ Наличие Request ID согласно контракту

Шаг 10

Что должен подтвердить тест прерванного или аварийно завершившегося SSE-потока?

Правильный ответ
  • ✅ В потоке есть error, а незавершённая пара сообщений не сохранена в истории

Шаг 11

Фейковый клиент структурированных ответов

При тестировании AI-сервиса настоящий LLM-клиент заменяют предсказуемым объектом. Такой объект не обращается к сети: он по очереди возвращает заранее подготовленные результаты и сохраняет аргументы каждого вызова. Благодаря этому можно отдельно проверить логику сервиса.

Задача

Реализуйте класс FakeStructuredLLMClient с конструктором и методом generate_structured(messages, schema_name, schema).

Требования

  • Не изменяйте имя класса, имя метода и имена их аргументов.
  • В конструкторе создайте новый список из последовательности results. Дальнейшие изменения очереди не должны изменять исходную последовательность.
  • В конструкторе также создайте публичный пустой список calls.
  • При каждом вызове generate_structured(...) сначала добавьте в calls словарь ровно с ключами messages, schema_name и schema. Значениями должны быть переданные аргументы.
  • После записи вызова удалите первый элемент внутренней очереди результатов и верните его.
  • Результаты должны возвращаться в том же порядке, в котором они были переданы конструктору.
  • Если очередь пуста, не подменяйте ошибку своим значением и наружу должен выйти обычный IndexError от операции со списком.
Правильный ответ 
class FakeStructuredLLMClient:
    def __init__(self, results):
        self.results = list(results)
        self.calls = []

    def generate_structured(self, messages, schema_name, schema):
        self.calls.append(
            {
                "messages": messages,
                "schema_name": schema_name,
                "schema": schema,
            }
        )

        return self.results.pop(0)

Шаг 12

Тест эндпоинта проверки состояния

В заготовке нужно написать тест для уже созданного эндпоинта GET /health. Эндпоинт считается исправным только тогда, когда возвращает статус 200 и точное JSON-тело {"status": "ok"}. Объект client работает так же, как TestClient из FastAPI.

Задача

Допишите тестовую функцию test_health(client): выполните запрос к эндпоинту и с помощью assert проверьте оба условия контракта.

Требования

  • Не изменяйте имя функции и аргумент client.
  • Ровно один раз вызовите client.get("/health") и сохраните полученный ответ в локальной переменной.
  • Одним выражением assert проверьте, что response.status_code == 200.
  • Отдельным выражением assert проверьте, что response.json() == {"status": "ok"}.
  • Не возвращайте True или False. Для корректного ответа функция должна просто завершиться, а для некорректного — выбросить AssertionError.
Правильный ответ 
def test_health(client) -> None:
    response = client.get("/health")

    assert response.status_code == 200
    assert response.json() == {"status": "ok"}

Шаг 13

Тест безопасной ошибки AI-эндпоинта

Эндпоинт анализа может получить ошибку от AI-провайдера, но клиент API не должен видеть внутренние детали, имя провайдера или секретный ключ. Вам нужно написать тест, который отправляет одно обращение и проверяет публичный контракт ответа с ошибкой. Объект client имеет интерфейс FastAPI TestClient.

Задача

Реализуйте тестовую функцию test_safe_analysis_error(client). Она должна проверить статус, структуру JSON, идентификатор запроса и отсутствие двух секретных подстрок в тексте ответа.

Требования

  • Не изменяйте имя функции и аргумент client.
  • Ровно один раз отправьте POST-запрос на "/support-requests/analyze".
  • Передайте аргумент json={"text": "Кофемашина не включается."}.
  • С помощью assert проверьте, что response.status_code == 502.
  • Вызовите response.json() ровно один раз и сохраните результат в переменной body.
  • Проверьте, что body["error"]["code"] равно "support_analysis_failed".
  • Проверьте, что body["error"]["message"] равно "Не удалось проанализировать обращение.".
  • Проверьте, что body["error"]["request_id"] равно значению заголовка response.headers["X-Request-ID"].
  • Отдельно проверьте, что строки "ProxyAPI" и "test-key" не встречаются в response.text.
Правильный ответ 
def test_safe_analysis_error(client) -> None:
    response = client.post(
        "/support-requests/analyze",
        json={"text": "Кофемашина не включается."},
    )

    assert response.status_code == 502

    body = response.json()

    assert body["error"]["code"] == "support_analysis_failed"
    assert body["error"]["message"] == "Не удалось проанализировать обращение."
    assert body["error"]["request_id"] == response.headers["X-Request-ID"]

    assert "ProxyAPI" not in response.text
    assert "test-key" not in response.text

Шаг 14

Тест эндпоинта анализа файла

Эндпоинт POST /files/analyze принимает файл и дополнительные поля как multipart-запрос. Вам нужно написать тест, который отправляет небольшой текстовый файл, задаёт вопрос и проверяет основные поля публичного ответа. Файл передаётся прямо в аргументе files, поэтому создавать его на диске не требуется.

Задача

Реализуйте тестовую функцию test_file_analysis(client). Выполните один запрос через объект client, затем проверьте статус и три значения в JSON-ответе.

Требования

  • Не изменяйте имя функции и аргумент client.
  • Ровно один раз вызовите client.post("/files/analyze", ...).
  • Передайте аргумент files={"file": ("request.txt", b"Order 8041 is damaged.", "text/plain")}.
  • Передайте аргумент data={"question": "Что случилось?", "require_human_review": "false"}.
  • Сохраните ответ и проверьте с помощью assert, что его статус равен 200.
  • Получите JSON-тело ответа и сохраните его в локальной переменной.
  • Проверьте, что body["source_name"] == "request.txt".
  • Проверьте, что body["source_kind"] == "text".
  • Проверьте, что body["analysis"]["answer"] == "Ответ найден в документе.".
  • Для корректного ответа функция должна завершиться без исключения; возвращать логическое значение не нужно.
Правильный ответ 
def test_file_analysis(client) -> None:
    response = client.post(
        "/files/analyze",
        files={
            "file": (
                "request.txt",
                b"Order 8041 is damaged.",
                "text/plain",
            )
        },
        data={
            "question": "Что случилось?",
            "require_human_review": "false",
        },
    )

    assert response.status_code == 200

    body = response.json()

    assert body["source_name"] == "request.txt"
    assert body["source_kind"] == "text"
    assert body["analysis"]["answer"] == "Ответ найден в документе."

Шаг 15

Тест SSE-потока с ответом модели

Эндпоинт потокового чата возвращает несколько SSE-событий в одном HTTP-ответе. Тест не должен зависеть от количества фрагментов текста, потому что модель может разбивать ответ по-разному. Нужно проверить общий контракт: правильный статус и тип содержимого, наличие хотя бы одного события delta, последующее событие completed и отсутствие события error.

Задача

Реализуйте тестовую функцию test_sse_stream(client, session_id). Откройте потоковый запрос как контекстный менеджер, соберите все полученные части в одну строку и проверьте последовательность событий.

Требования

  • Не изменяйте имя функции и аргументы client и session_id.
  • Откройте результат client.stream("POST", f"/chat/sessions/{session_id}/messages/stream", json={"message": "Где мой заказ?"}) с помощью with.
  • Внутри блока with проверьте, что response.status_code == 200.
  • Проверьте, что значение response.headers["content-type"] начинается со строки "text/event-stream". После неё могут находиться дополнительные параметры, например кодировка.
  • Внутри того же блока соберите всё тело выражением "".join(response.iter_text()) и сохраните полученную строку.
  • После чтения проверьте, что в строке есть подстроки "event: delta" и "event: completed".
  • Проверьте, что первое вхождение "event: completed" расположено позже первого вхождения "event: delta".
  • Проверьте, что подстроки "event: error" в теле нет.
  • Не проверяйте точное количество событий delta и конкретное разбиение ответа на фрагменты.
Правильный ответ 
def test_sse_stream(client, session_id: str) -> None:
    with client.stream(
        "POST",
        f"/chat/sessions/{session_id}/messages/stream",
        json={"message": "Где мой заказ?"},
    ) as response:
        assert response.status_code == 200
        assert response.headers["content-type"].startswith(
            "text/event-stream"
        )

        body = "".join(response.iter_text())

    assert "event: delta" in body
    assert "event: completed" in body
    assert body.find("event: completed") > body.find("event: delta")
    assert "event: error" not in body

Шаг 16

Тест очистки и удаления чат-сессии

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

Задача

Реализуйте тестовую функцию test_clear_and_delete_chat_session(client). В одном тестовом сценарии создайте сессию, добавьте в неё сообщение, очистите историю, а затем удалите всю сессию.

Требования

  • Не изменяйте имя функции и аргумент client.
  • Вызовите client.post("/chat/sessions"), сохраните ответ и проверьте, что его статус равен 201.
  • Получите JSON созданной сессии и сохраните значение ключа session_id в переменной session_id.
  • Отправьте сообщение вызовом client.post(f"/chat/sessions/{session_id}/messages", json={"message": "Где мой заказ?"}) и проверьте статус 200.
  • Очистите историю вызовом client.delete(f"/chat/sessions/{session_id}/messages") и проверьте статус 204.
  • После очистки вызовите client.get(f"/chat/sessions/{session_id}"). Проверьте статус 200 и значение response.json()["history_messages"] == 0.
  • Удалите всю сессию вызовом client.delete(f"/chat/sessions/{session_id}") и проверьте статус 204.
  • После удаления ещё раз вызовите client.get(f"/chat/sessions/{session_id}") и проверьте статус 404.
  • Соблюдайте указанный порядок запросов. Не возвращайте логическое значение: корректный сценарий должен завершиться без исключения.
Правильный ответ 
def test_clear_and_delete_chat_session(client) -> None:
    response = client.post("/chat/sessions")
    assert response.status_code == 201

    session_id = response.json()["session_id"]

    response = client.post(
        f"/chat/sessions/{session_id}/messages",
        json={"message": "Где мой заказ?"},
    )
    assert response.status_code == 200

    response = client.delete(
        f"/chat/sessions/{session_id}/messages"
    )
    assert response.status_code == 204

    response = client.get(
        f"/chat/sessions/{session_id}"
    )
    assert response.status_code == 200
    assert response.json()["history_messages"] == 0

    response = client.delete(
        f"/chat/sessions/{session_id}"
    )
    assert response.status_code == 204

    response = client.get(
        f"/chat/sessions/{session_id}"
    )
    assert response.status_code == 404