ТЕХНИЧЕСКОЕ ЗАДАНИЕ: Реализовать simulation testing для AI-агента

1. Цель задачи

Научиться проектировать и реализовывать simulation testing окружение для AI-агента, которое позволяет имитировать работу внешних сервисов и целенаправленно внедрять сбои (injection|fault injection). Результатом станет набор автоматизированных тестов, покрывающих редкие сценарии (отказы API, таймауты, невалидные ответы, задержки), которые сложно воспроизвести в реальном окружении.

Ключевой результат Написаны и выполняются тесты, эмулирующие не менее 5 типов редких сбоев, с проверкой корректной обработки агентом (retry, fallback, логирование, корректный ответ пользователю).

2. Исходные данные

Если нет реального инструмента — симулируем:

1. Создать простого AI-агента на Python (например, на базе LangChain или просто класс с методом get_weather(city), который делает HTTP GET к api.weather.example).
2. Развернуть WireMock в Docker (docker run -p 8080:8080 wiremock/wiremock) или написать mock-сервер на FastAPI (один файл).
3. Написать декоратор inject_fault для подмены запросов в тестах (с помощью responses для requests или aioresponses для aiohttp).

3. Технологический стек

4. Этапы выполнения

Этап 1: Подготовка тестового агента (оценка 30-60 мин)

Действия

1. Если нет готового агента — написать простого агента, который вызывает одно внешнее REST API (например, GET /v1/weather?city=...).
2. Реализовать обработку ответа: извлечение температуры, возврат строки пользователю.
3. Добавить базовую логику повторных попыток (retry) при 5xx и таймаутах (максимум 3 попытки).
4. Написать конфиг с адресом внешнего сервиса (через переменные окружения).
5. Убедиться, что агент работает с реальным mock-сервером без сбоев.

# agent.py (упрощённый пример)
import asyncio
import aiohttp

class WeatherAgent:
    def __init__(self, api_url: str):
        self.api_url = api_url
        self.max_retries = 3

    async def get_weather(self, city: str) -> str:
        for attempt in range(self.max_retries):
            try:
                async with aiohttp.ClientSession() as session:
                    async with session.get(f"{self.api_url}/v1/weather", params={"city": city}) as resp:
                        if resp.status == 200:
                            data = await resp.json()
                            return f"Temperature: {data['temp']}°C"
                        elif resp.status >= 500:
                            raise aiohttp.HttpProcessingError(f"Server error {resp.status}")
            except (aiohttp.ClientError, asyncio.TimeoutError) as e:
                if attempt == self.max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)

Ожидаемый результат этапа Рабочий агент, который корректно обрабатывает успешные ответы mock-сервера.

Этап 2: Разработка mock API для внешних сервисов (1-2 часа)

Действия

1. Выбрать способ mock-сервера WireMock (рекомендуется для реалистичности) или in-process mock с aioresponses.

2. WireMock

   • Запустить WireMock в Docker.
   • Создать файлы __files/ и mappings/ для статических ответов.
   • Настроить эндпоинт /v1/weather с ответом по умолчанию:

     { "temp": 22, "condition": "clear" }
     

3. In-process mock (альтернатива):

   • Использовать aioresponses как контекстный менеджер в тестах.
   • Определить фикстуру, которая отключает реальные HTTP запросы и подменяет ответы.

4. Создать несколько профилей ответов

   • Успешный (200, нормальные данные)
   • Пустой ответ (200, битый JSON)
   • Неверный статус (404, 429, 503)
   • Задержка (delay 5 сек – имитация таймаута)

5. Добавить конфигурацию mock-сервера в тестовую фикстуру pytest.

# conftest.py (пример с aioresponses)
import pytest
from aioresponses import aioresponses

@pytest.fixture
def mock_weather():
    with aioresponses() as m:
        m.get("http://test-api/v1/weather", status=200, body='{"temp":22}')
        yield m

Ожидаемый результат этапа Mock-сервер готов, тесты с успешным сценарием проходят.

Этап 3: Внедрение fault injection (1-1.5 часа)

Действия

1. Определить набор сбоев для тестирования

   • HTTP 500 (внутренняя ошибка сервера)
   • HTTP 429 (rate limit, с заголовком Retry-After)
   • Таймаут (задержка > допустимого таймаута агента)
   • Битая JSON строка (200, но тело не валидный JSON)
   • Пустой ответ (200, тело null)
   • Медленный ответ (задержка 3-5 сек, но без таймаута)

2. Реализовать механизм fault injection двумя способами:

   • Статический в каждом тесте задать конкретный сбой.
   • Динамический декоратор / фикстура, которая с вероятностью X подменяет ответ.

3. Сценарий статического fault injection через aioresponses:

@pytest.mark.parametrize("fault_type", ["timeout", "500", "429", "bad_json", "empty"])
async def test_fault_scenarios(agent, fault_type):
    with aioresponses() as m:
        if fault_type == "timeout":
            m.get("http://test-api/v1/weather", exception=asyncio.TimeoutError)
        elif fault_type == "500":
            m.get("http://test-api/v1/weather", status=500)
        elif fault_type == "429":
            m.get("http://test-api/v1/weather", status=429, headers={"Retry-After": "2"})
        elif fault_type == "bad_json":
            m.get("http://test-api/v1/weather", status=200, body='not json')
        elif fault_type == "empty":
            m.get("http://test-api/v1/weather", status=200, body='null')

        result = await agent.get_weather("Moscow")
        # Проверить, что агент возвращает fallback-сообщение, а не крашится
        assert "ошибка" in result.lower() or "недоступен" in result.lower()

4. Добавить логирование в агент (например, через structlog), чтобы в логах теста было видно, какие попытки были.

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

Этап 4: Написание simulation тестов и проверка устойчивости (1-2 часа)

Действия

1. Дополнить тесты проверкой:

   • Количество попыток retry (не более 3)
   • Время выполнения (при таймауте не превышает заданный таймаут)
   • Логирование каждого сбоя (через захват логов в тесте)
   • Метрики (если внедрили счётчики)

2. Написать тест на восстановление после временной ошибки (агент должен корректно обработать сервис, который сначала падает, а потом отвечает успешно).

async def test_recovery_after_fault(agent):
    with aioresponses() as m:
        # Первые два запроса – 500, третий – успех
        m.get("http://test-api/v1/weather", status=500, repeat=2)
        m.get("http://test-api/v1/weather", status=200, body='{"temp":25}', repeat=1)

        result = await agent.get_weather("Moscow")
        assert "25" in result
        # Убедиться, что было ровно 3 попытки

3. Добавить тесты на конкурирующие сценарии (если агент асинхронный): одновременные запросы с разными сбоями.

4. Оформить отчёт о покрытии редких сценариев (Markdown список).

Ожидаемый результат этапа Все тесты проходят, фиксируются логи, метрики (если есть). Отчёт о покрытии.

Этап 5: Интеграция в CI (оценка 45 мин)

Действия

1. Создать Makefile с командой make test.
2. Добавить GitHub Actions workflow .github/workflows/test.yml:
   • Установка Python и зависимостей
   • Запуск pytest --timeout=60 -v --tb=short
3. Настроить вывод отчёта (pytest-html или junit.xml).
4. Убедиться, что mock-сервер не требуется извне – все тесты используют in-process mock.

Ожидаемый результат этапа Тесты автоматически запускаются на каждый PR, результат доступен.

5. Критерии приемки (Definition of Done)

• Реализован mock API для всех внешних вызовов агента (минимум 1 эндпоинт)
• Fault injection покрывает не менее 5 типов ошибок (таймаут, 500, 429, битый JSON, пустой ответ)
• Написаны автоматизированные тесты для каждого типа ошибки (параметризованные)
• Тесты проверяют, что агент не падает, а возвращает ожидаемое fallback-сообщение
• Добавлена фикстура для захвата логов, проверяющая, что ошибки логируются
• Реализован тест на восстановление после временной ошибки (2 сбоя → успех)
• Тесты запускаются одной командой (pytest) и проходят в CI
• Все тесты не требуют доступа к внешним сервисам
• В README описано, как запустить тесты и добавить новый сценарий

6. Ожидаемый результат

Основные артефакты

• Репозиторий с кодом агента (agent.py)
• Тесты (test_*.py) и фикстуры (conftest.py)
• Конфигурация CI (.github/workflows/test.yml)
• Файл requirements.txt или pyproject.toml с зависимостями
• README.md с описанием структуры, запуска и примером добавления нового сбоя
• Отчёт о покрытии редких сценариев (таблица в Markdown)

Опционально дашборд метрик (если внедрён prometheus_client), логи тестов в stdout в формате JSON.

7. Возможные сложности и их решение

8. Бюджет времени (оценка)

Примечание Для первого исполнения заложите 8–10 часов с учётом исследования инструментов и отладки.

9. Связанные вопросы из базы знаний

10. Чек-лист самопроверки

• Я настроил mock-сервер, который возвращает все необходимые ответы (успех и сбои)
• Я внедрил fault injection для минимум 5 типов сбоев (таймаут, 500, 429, битый JSON, пустой ответ)
• Я написал параметризованные тесты, которые проверяют реакцию агента на каждый сбой (не падает, возвращает fallback)
• Я убедился, что тесты запускаются одной командой (pytest) и проходят без доступа к внешним ресурсам
• Я добавил логирование ошибок и (опционально) метрики ошибок по типу
• Я оформил README с описанием как запустить тесты и добавить новый сценарий