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

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

avito.testing — публичное пространство тестовых утилит для consumer-side тестов без реального HTTP.

Fake transport

Публичные тестовые утилиты SDK.

AsyncFakeTransport 

AsyncFakeTransport(
    *,
    base_url: str = "https://api.avito.ru",
    fanout_recorder: FanoutPeakRecorder | None = None,
)

Deterministic async fake transport for SDK contract tests.

Initialize AsyncFakeTransport.

add 

add(
    method: str, path: str, *responses: RouteResponder
) -> AsyncFakeTransport

Регистрирует один или несколько ответов для HTTP-маршрута.

add_json 

add_json(
    method: str,
    path: str,
    payload: JsonValue,
    *,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
) -> AsyncFakeTransport

Регистрирует JSON-ответ для HTTP-маршрута.

build 

build(
    *,
    retry_policy: RetryPolicy | None = None,
    user_id: int | None = None,
    authenticated: bool = False,
    auth_settings: AuthSettings | None = None,
) -> AsyncTransport

Создаёт низкоуровневый AsyncTransport поверх fake transport.

as_client 

as_client(
    *,
    user_id: int | None = None,
    retry_policy: RetryPolicy | None = None,
    authenticated: bool = False,
    auth_settings: AuthSettings | None = None,
) -> AsyncAvitoClient

Создает публичный AsyncAvitoClient поверх fake transport.

count 

count(
    *, method: str | None = None, path: str | None = None
) -> int

Возвращает число перехваченных запросов с опциональной фильтрацией.

last 

last(
    *, method: str | None = None, path: str | None = None
) -> RecordedRequest

Возвращает последний перехваченный запрос с опциональной фильтрацией.

FanoutPeakRecorder 

FanoutPeakRecorder()

Считает пик одновременно выполняющихся async fake-запросов.

Initialize FanoutPeakRecorder.

enter async 

enter() -> None

Record fan-out enter event.

exit async 

exit() -> None

Record fan-out exit event.

AsyncSwaggerFakeTransport 

AsyncSwaggerFakeTransport(
    *,
    registry: SwaggerRegistry,
    base_url: str = "https://api.avito.ru",
)

Bases: AsyncFakeTransport

Async fake transport that registers routes by Swagger operation key.

Initialize AsyncSwaggerFakeTransport.

add_operation 

add_operation(
    operation_key: str,
    payload: JsonValue,
    *,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
) -> AsyncSwaggerFakeTransport

Register response for one Swagger operation key.

add_success_operation 

add_success_operation(
    operation_key: str,
    *,
    payload: JsonValue | None = None,
    headers: Mapping[str, str] | None = None,
) -> AsyncSwaggerFakeTransport

Register a deterministic success response for one Swagger operation.

operation 

operation(operation_key: str) -> SwaggerOperation

Return operation by key or raise an assertion error.

invoke_binding async 

invoke_binding(
    binding: DiscoveredSwaggerBinding,
    *,
    client: AsyncAvitoClient | None = None,
) -> object

Build and invoke async SDK call from discovered Swagger binding metadata.

FakeTransport 

FakeTransport(*, base_url: str = 'https://api.avito.ru')

Deterministic fake transport for SDK contract tests.

add 

add(
    method: str, path: str, *responses: RouteResponder
) -> FakeTransport

Регистрирует один или несколько ответов для HTTP-маршрута.

add_json 

add_json(
    method: str,
    path: str,
    payload: JsonValue,
    *,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
) -> FakeTransport

Регистрирует JSON-ответ для HTTP-маршрута.

build 

build(
    *,
    retry_policy: RetryPolicy | None = None,
    user_id: int | None = None,
) -> Transport

Создаёт низкоуровневый Transport поверх fake transport (internal helper).

as_client 

as_client(
    *,
    user_id: int | None = None,
    retry_policy: RetryPolicy | None = None,
) -> AvitoClient

Создает публичный AvitoClient поверх fake transport без реального HTTP.

count 

count(
    *, method: str | None = None, path: str | None = None
) -> int

Возвращает число перехваченных запросов с опциональной фильтрацией.

last 

last(
    *, method: str | None = None, path: str | None = None
) -> RecordedRequest

Возвращает последний перехваченный запрос с опциональной фильтрацией.

RecordedRequest dataclass 

RecordedRequest(
    method: str,
    path: str,
    params: dict[str, str],
    headers: dict[str, str],
    json_body: JsonValue,
    content: bytes,
)

Зафиксированный HTTP-запрос, перехваченный FakeTransport.

SwaggerFakeTransport 

SwaggerFakeTransport(
    *,
    registry: SwaggerRegistry,
    base_url: str = "https://api.avito.ru",
)

Bases: FakeTransport

Fake transport that validates requests against local Swagger operations.

add_operation 

add_operation(
    operation_key: str,
    payload: JsonValue,
    *,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
) -> SwaggerFakeTransport

Register response for one Swagger operation key.

add_success_operation 

add_success_operation(
    operation_key: str,
    *,
    payload: JsonValue | None = None,
    headers: Mapping[str, str] | None = None,
) -> SwaggerFakeTransport

Register a deterministic success response for one Swagger operation.

operation 

operation(operation_key: str) -> SwaggerOperation

Return operation metadata by canonical key.

invoke_binding 

invoke_binding(
    binding: DiscoveredSwaggerBinding,
    *,
    client: AvitoClient | None = None,
) -> object

Build and invoke SDK call from discovered Swagger binding metadata.

SwaggerRoute dataclass 

SwaggerRoute(
    operation: SwaggerOperation,
    payload: JsonValue,
    status_code: int,
    headers: Mapping[str, str],
)

Registered fake route bound to one Swagger operation.

json_response 

json_response(
    payload: JsonValue,
    *,
    status_code: int = 200,
    headers: Mapping[str, str] | None = None,
) -> httpx.Response

Создаёт httpx.Response с JSON-телом для использования в FakeTransport.

route_sequence 

route_sequence(
    *responses: RouteResponder,
) -> Iterable[RouteResponder]

Упаковывает несколько ответов в последовательность для FakeTransport.add().

error_payload 

error_payload(status_code: int) -> JsonValue

Build deterministic JSON error payload for contract tests.

generate_schema_value 

generate_schema_value(schema: SwaggerSchema) -> JsonValue

Build deterministic JSON value that covers all fields in a Swagger schema.

validate_schema_value 

validate_schema_value(
    value: object, schema: SwaggerSchema, *, path: str
) -> None

Validate JSON value keys and types against a normalized Swagger schema.

Контракт

  • FakeTransport записывает выполненные запросы и отдаёт заранее заданные ответы.
  • json_response() создаёт JSON-ответ для маршрута.
  • route_sequence() задаёт последовательность ответов для retry и stateful-сценариев.
  • FakeTransport.as_client() создаёт полностью инициализированный AvitoClient поверх fake transport без реального HTTP.
  • AsyncFakeTransport.as_client() создаёт полностью инициализированный AsyncAvitoClient поверх httpx.AsyncClient и httpx.MockTransport.
  • FanoutPeakRecorder измеряет пик одновременно выполняющихся async fake-запросов в consumer-side тестах агрегаторов.
  • RecordedRequest позволяет проверять method, path, query params, headers и JSON body.

Пользовательские тесты должны работать через публичные утилиты avito.testing, а не через приватные поля AvitoClient.