# Проверьте HTTP-адаптер: семь тестов отказов, таймаут и секрет в ответе

Испытайте конкретный get_artifact из предыдущей практики. Проверьте неверные аргументы, HTTP-статусы, плохой JSON, отмену зависшего запроса и безопасную MCP-ошибку.

**В результате:** Набор unittest, который замечает ошибочную обработку 429, не допускает HTTP на плохом входе и проверяет контракт tool на настоящем SDK.

**Перед началом:** Файлы http_api_client.py и http_api_tool.py из практики HTTP API, Python 3.11 и те же закреплённые зависимости.

## 1. Разделите проверки клиента и инструмента

Прямая проверка HTTP-клиента видит адрес, заголовок, число запросов и разобранную модель. MCP-проверка видит tools/list, inputSchema, outputSchema, structuredContent и isError. Если проверять только конечный JSON, можно пропустить лишний запрос, редирект или потерянную ошибку.

Все ответы создаёт MockTransport. Он не доказывает работу реального соединения и сам не воспроизводит сетевое ожидание. Для ReadTimeout тест явно возбуждает исключение. Для общего дедлайна отдельный async handler действительно ждёт до отмены — без сети и долгого сна.

## 2. Сохраните полный набор проверок

Сохраните test_http_adapter.py рядом с клиентом и tool. Синтетическая строка SYNTHETIC_TOKEN служит меткой: она намеренно встречается в закрытом ответе или исключении, а проверка запрещает её появление в публичной ошибке. Это не настоящий ключ.

### test_http_adapter.py

```python
"""Behavior checks for HTTP failures and the MCP boundary; never call a remote API."""
import asyncio
from datetime import timedelta
import unittest

import httpx
from mcp.shared.memory import create_connected_server_and_client_session

from http_api_client import AdapterError, ArtifactClient
from http_api_tool import make_server

GOOD = {"item_id": "KIT-42", "title": "Demo kit", "status": "ready"}
SYNTHETIC_TOKEN = "synthetic-secret-for-test-only"


class AdapterTests(unittest.IsolatedAsyncioTestCase):
    async def test_success_request_shape_timeout_and_client_lifetime(self):
        requests = []
        def handler(request):
            requests.append(request)
            return httpx.Response(200, json=GOOD)
        async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
            result = await ArtifactClient(http, SYNTHETIC_TOKEN).get_artifact("KIT-42")
            self.assertEqual(result.model_dump(), GOOD)
        self.assertTrue(http.is_closed)
        self.assertEqual(len(requests), 1)
        request = requests[0]
        self.assertEqual(request.method, "GET")
        self.assertEqual(str(request.url), "https://inventory.example.invalid/v1/items/KIT-42")
        self.assertEqual(request.headers["Authorization"], "Bearer " + SYNTHETIC_TOKEN)
        self.assertNotIn(SYNTHETIC_TOKEN, str(request.url))
        self.assertEqual(request.extensions["timeout"], {"connect": 2.0, "read": 5.0, "write": 5.0, "pool": 1.0})

    async def test_bad_input_never_reaches_http(self):
        requests = []
        def handler(request):
            requests.append(request)
            return httpx.Response(200, json=GOOD)
        async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
            client = ArtifactClient(http)
            for item_id in ["../secret", "https://other.invalid", "", "KIT-0", 42]:
                with self.subTest(item_id=item_id), self.assertRaisesRegex(AdapterError, "INVALID_ID"):
                    await client.get_artifact(item_id)
        self.assertEqual(requests, [])

    async def test_statuses_are_safe_and_do_not_retry_or_follow_redirects(self):
        for status, code in [(401, "AUTH"), (403, "AUTH"), (404, "NOT_FOUND"), (429, "RATE_LIMIT"), (500, "UPSTREAM"), (503, "UPSTREAM"), (302, "HTTP_STATUS"), (400, "HTTP_STATUS")]:
            requests = []
            def handler(request):
                requests.append(request)
                return httpx.Response(status, text=SYNTHETIC_TOKEN, headers={"Location": "https://other.invalid/", "Retry-After": "2"})
            with self.subTest(status=status):
                async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
                    with self.assertRaises(AdapterError) as caught:
                        await ArtifactClient(http, SYNTHETIC_TOKEN).get_artifact("KIT-42")
                self.assertEqual(caught.exception.code, code)
                self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception))
                self.assertEqual(len(requests), 1)

    async def test_timeout_and_connection_failure(self):
        for exception, code in [(httpx.ReadTimeout, "TIMEOUT"), (httpx.ConnectError, "NETWORK")]:
            def handler(request):
                raise exception(SYNTHETIC_TOKEN, request=request)
            with self.subTest(exception=exception.__name__):
                async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
                    with self.assertRaises(AdapterError) as caught:
                        await ArtifactClient(http).get_artifact("KIT-42")
                self.assertEqual(caught.exception.code, code)
                self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception))

    async def test_total_deadline_cancels_waiting_handler(self):
        cancelled = asyncio.Event()
        async def handler(request):
            try:
                await asyncio.Event().wait()
            finally:
                cancelled.set()
        async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
            with self.assertRaisesRegex(AdapterError, "TIMEOUT"):
                await ArtifactClient(http, deadline=0.03).get_artifact("KIT-42")
        self.assertTrue(cancelled.is_set())

    async def test_invalid_json_schema_and_wrong_record(self):
        responses = [httpx.Response(200, text="not JSON"), httpx.Response(200, json=[]), httpx.Response(200, json={}), httpx.Response(200, json={**GOOD, "status": "unknown"}), httpx.Response(200, json={**GOOD, "title": 42}), httpx.Response(200, json={**GOOD, "private": SYNTHETIC_TOKEN}), httpx.Response(200, json={**GOOD, "item_id": "KIT-43"})]
        for response in responses:
            with self.subTest(body=response.text):
                async with httpx.AsyncClient(transport=httpx.MockTransport(lambda request: response), trust_env=False) as http:
                    with self.assertRaises(AdapterError) as caught:
                        await ArtifactClient(http).get_artifact("KIT-42")
                self.assertEqual(caught.exception.code, "BAD_RESPONSE")
                self.assertNotIn(SYNTHETIC_TOKEN, str(caught.exception))

    async def test_mcp_schema_success_bad_input_and_safe_failure(self):
        requests = []
        def handler(request):
            requests.append(request)
            return httpx.Response(503, text=SYNTHETIC_TOKEN) if request.url.path.endswith("KIT-43") else httpx.Response(200, json=GOOD)
        async with httpx.AsyncClient(transport=httpx.MockTransport(handler), trust_env=False) as http:
            server = make_server(ArtifactClient(http, SYNTHETIC_TOKEN))
            async with create_connected_server_and_client_session(server, read_timeout_seconds=timedelta(seconds=5), raise_exceptions=True) as session:
                tool = (await session.list_tools()).tools[0]
                self.assertEqual(tool.name, "get_artifact")
                self.assertIn("item_id", tool.inputSchema["required"])
                self.assertIsNotNone(tool.outputSchema)
                success = await session.call_tool("get_artifact", {"item_id": "KIT-42"})
                self.assertFalse(success.isError)
                self.assertEqual(success.structuredContent, GOOD)
                before = len(requests)
                for arguments in [{}, {"item_id": "../secret"}, {"item_id": 42}]:
                    result = await session.call_tool("get_artifact", arguments)
                    self.assertTrue(result.isError)
                self.assertEqual(len(requests), before)
                failure = await session.call_tool("get_artifact", {"item_id": "KIT-43"})
                self.assertTrue(failure.isError)
                self.assertIsNone(failure.structuredContent)
                self.assertIn("UPSTREAM", failure.content[0].text)
                self.assertNotIn(SYNTHETIC_TOKEN, failure.model_dump_json())


if __name__ == "__main__":
    unittest.main(verbosity=2)
```

```powershell
.\.venv\Scripts\python.exe -m unittest -v test_http_adapter
$LASTEXITCODE
```

```text
test_bad_input_never_reaches_http ... ok
test_invalid_json_schema_and_wrong_record ... ok
test_mcp_schema_success_bad_input_and_safe_failure ... ok
test_statuses_are_safe_and_do_not_retry_or_follow_redirects ... ok
test_success_request_shape_timeout_and_client_lifetime ... ok
test_timeout_and_connection_failure ... ok
test_total_deadline_cancels_waiting_handler ... ok
Ran 7 tests
OK
0
```

В unittest имена будут дополнены названием класса, а время запуска зависит от компьютера. Важны семь ok, итог OK и код 0. INFO SDK/HTTPX в stderr допустимы. На авторской среде все семь прошли; stdout и stderr сохранены отдельно.

## 3. Прочитайте, что замечает каждый тест

| Тест | Что проверяет | Какой дефект поймает |
| --- | --- | --- |
| success_request_shape… | GET, точный адрес, Authorization, настройки timeout, закрытие клиента | Токен в URL, неверный endpoint или забытое закрытие |
| bad_input… | Нет запросов при неверном item_id | Сетевой запрос до валидации |
| statuses… | 401/403/404/429/5xx/302/400 и один запрос | Лишние повторы, переход по Location, неверный код отказа |
| timeout_and_connection… | ReadTimeout и ConnectError | Потерянный отказ или утечка сырого исключения |
| total_deadline… | Реальная отмена ожидания через 0.03 s | Зависший обработчик без общего предела |
| invalid_json… | Мусор, массив, пропущенные поля, неверные типы, лишнее поле и другой id | Передача непроверенного ответа |
| mcp_schema… | Настоящие list/call, плохие аргументы, isError и безопасный результат | Неверная регистрация или успешный ответ вместо ошибки |

Тест settings timeout смотрит на параметры, переданные HTTPX; он не измеряет поведение сети. Тест дедлайна проверяет asyncio.timeout в этой программе. Проверка секрета относится к сообщению AdapterError и результату MCP; она не является аудитом всех логов, памяти или окружения процесса.

## 4. Убедитесь, что тест способен упасть

Сделайте отдельную копию новой учебной папки. Только в копии клиента замените код RATE_LIMIT для 429 на AUTH. Запустите unittest: test_statuses_are_safe_and_do_not_retry_or_follow_redirects должен упасть на сравнении кода, а команда завершиться ненулевым кодом. Верните RATE_LIMIT и повторите запуск.

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

## 5. Найдите слой проблемы

| Симптом | Что проверить сначала |
| --- | --- |
| ModuleNotFoundError | Все четыре .py и requirements в одной папке; запускается интерпретатор своего .venv |
| Нет inputSchema / outputSchema | Версия mcp==1.30.0, аннотации аргумента и модели Artifact |
| INVALID_ID не блокирует запрос | Прямую валидацию клиента и передачу аргументов tool |
| BAD_RESPONSE на правильной записи | Строгие типы, неизвестные поля и совпадение item_id |
| Тест deadline не заканчивается | Контекст asyncio.timeout вокруг await запроса; не увеличивать предел вслепую |
| Синтетический секрет появился в ошибке | Не возвращается ли response.text или repr исключения HTTPX/Pydantic |

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

```powershell
.\.venv\Scripts\python.exe -m unittest -v test_http_adapter.AdapterTests.test_statuses_are_safe_and_do_not_retry_or_follow_redirects
```

## 6. Подготовьте адаптацию под свой контракт

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

Для retry сначала выберите ограниченный план повторов и поведение при 429/Retry-After. Не добавляйте while True. Для записи нужны отдельные правила повторной операции и проверка идемпотентности; текущий GET-пример их не обещает.

Следующий полезный результат — собственная таблица: «изменение контракта → какой тест падает». Если изменение полей, прав или ошибки не замечает ни одна проверка, добавьте соответствующий сценарий до расширения tool.

## Версия и документация

Python 3.11; mcp==1.30.0; httpx==0.28.1; pydantic==2.13.5. Проверка выполняет MCP-вызовы в памяти и HTTP через MockTransport. Не проверены внешний API, удалённый MCP и установка новой среды. Документация прочитана 12 октября 2026 года.

- [MCP: схема, результат и ошибки инструмента](https://modelcontextprotocol.io/specification/2025-11-25/server/tools) · Протокол 2025-11-25

- [Python SDK v1: тестирование в памяти](https://py.sdk.modelcontextprotocol.io/v1/testing/) · Пример проверен на mcp==1.30.0

- [HTTPX: MockTransport](https://www.python-httpx.org/advanced/transports/#mock-transports) · Пример проверен на httpx==0.28.1

- [HTTPX: connect, read, write и pool timeout](https://www.python-httpx.org/advanced/timeouts/) · Документация сверена 12.10.2026

- [HTTPX: исключения запроса](https://www.python-httpx.org/exceptions/) · Документация сверена 12.10.2026

- [Python 3.11: IsolatedAsyncioTestCase](https://docs.python.org/3.11/library/unittest.html#unittest.IsolatedAsyncioTestCase) · Python 3.11

- [Python 3.11: asyncio.timeout](https://docs.python.org/3.11/library/asyncio-task.html#asyncio.timeout) · Python 3.11