Responses & serializers¶

A response is what the mock server returns for a handler. You register one with service.register(name, response). All responses derive from BaseMockResponse, so you can write your own too.

JsonResponse¶

JsonResponse serializes any object to a JSON body:

from asyncly.srvmocker import JsonResponse

service.register("fact", JsonResponse({"fact": "Meow.", "length": 5}))
service.register("missing", JsonResponse({"error": "not found"}, status=404))

RawResponse¶

RawResponse returns arbitrary bytes with arbitrary headers — ideal for testing client behavior on malformed payloads, unexpected content types, or empty bodies:

from asyncly.srvmocker import RawResponse

service.register(
    "broken_json",
    RawResponse(
        body=b'{"truncated":',
        status=200,
        headers={"Content-Type": "application/json"},
    ),
)

SequenceResponse¶

SequenceResponse returns a different response on each call, in order — useful for testing retries and pagination. The on_exhausted argument controls what happens after the last response:

from asyncly.srvmocker import JsonResponse, RawResponse, SequenceResponse

service.register(
    "flaky",
    SequenceResponse(
        [
            RawResponse(status=503),
            RawResponse(status=503),
            JsonResponse({"ok": True}),
        ],
        on_exhausted="last",  # "raise" (default), "cycle", or "last"
    ),
)

"raise" (default) — raise SequenceExhausted on the next call (surfaces to the client as a 500).
"cycle" — loop back to the first response.
"last" — keep returning the last response forever.

Constructing with an empty list raises ValueError eagerly.

LatencyResponse¶

LatencyResponse wraps another response and delays it — the tool for testing timeouts:

from asyncly.srvmocker import JsonResponse
from asyncly.srvmocker.responses.timeout import LatencyResponse

service.register(
    "slow",
    LatencyResponse(wrapped=JsonResponse({"ok": True}), latency=1.5),
)

import asyncio
import pytest

with pytest.raises(asyncio.TimeoutError):
    await client.fetch_fact(timeout=0.1)

Other formats¶

For non-JSON wire formats, register the matching response (each takes the same body, status, headers arguments as JsonResponse). These pull their serializer from a third-party package — only msgpack ships as an asyncly extra; install TOML and YAML support yourself:

Response	Format	Install
`MsgpackResponse`	msgpack	`pip install "asyncly[msgspec]"`
`TomlResponse`	TOML	`pip install toml`
`YamlResponse`	YAML	`pip install pyyaml`

ContentResponse and serializers¶

ContentResponse is the general form: a body plus an explicit Serializer that pairs a dumps callable with a content type. The format-specific responses above are thin wrappers over it. Use it to plug in a custom serializer:

from asyncly.srvmocker import ContentResponse
from asyncly.srvmocker.serialization.json import JsonSerializer

service.register(
    "custom",
    ContentResponse(body={"ok": True}, serializer=JsonSerializer),
)

Built-in serializers: JsonSerializer, MsgpackSerializer, TomlSerializer, YamlSerializer. See the serializers reference.