nimax

Record and replay niquests HTTP and WebSocket interactions in pytest.

nimax is a VCR-style cassette library built natively for niquests — supporting lazy responses, multiplexed connections, AsyncSession, and WebSockets. It is to niquests what betamax is to requests.

Installation

pip install nimax

Quick start

Automatic fixture

nimax registers a nimax_session pytest fixture automatically. Use it instead of niquests.Session() in your tests:

def test_my_api(nimax_session):
    resp = nimax_session.get("https://api.example.com/users")
    assert resp.status_code == 200

On the first run nimax records the real HTTP response to a cassette file under cassettes/<test_module>/<test_name>.json. Subsequent runs replay from the cassette — no network required.

Async sessions

import pytest
import niquests


async def test_async(nimax_session):
    async with niquests.AsyncSession() as session:
        with NimaxRecorder(session).use_cassette("my_cassette.json"):
            resp = await session.get("https://api.example.com/data")
            assert resp.status_code == 200

Programmatic API

import niquests
from nimax import NimaxRecorder, RecordMode


def test_programmatic(tmp_path):
    session = niquests.Session()
    cassette_path = tmp_path / "my_cassette.json"
    with NimaxRecorder(session).use_cassette(cassette_path, record_mode=RecordMode.ONCE):
        resp = session.get("https://api.example.com/users")
        assert resp.status_code == 200

Record modes

WebSocket support

WebSocket connections opened through a cassette-backed session are recorded and replayed automatically, alongside HTTP interactions, in the same cassette file:

def test_ws_echo(nimax_session):
    resp = nimax_session.get("wss://echo.example.com")
    resp.extension.send_payload('{"id": "1", "op": "ping"}')
    reply = resp.extension.next_payload()

On replay, a recv frame only releases once the real send_payload() call it depends on has actually happened — matching how a real socket can't deliver a response before its triggering request went out. By default this is tracked by position (send count) in the recorded log, which is correct as long as your live send order doesn't diverge from the recorded order.

Correlating responses by id

If your protocol embeds a correlation id in its messages (e.g. JSON-RPC-style {"id": ..., ...}) and you dispatch responses from a background reader task — so concurrent, in-flight requests can legitimately resolve out of order — pass ws_id_extractor to correlate replay by that id instead of by position:

with NimaxRecorder(session).use_cassette("my_cassette.json", ws_id_extractor="id"):
    ...

ws_id_extractor accepts:

• A dotted JSON path string, e.g. "id" or "params.id" for a nested field.
• A callable (payload: str | bytes) -> Any | None for anything else (non-JSON protocols, custom shapes).

A message whose id doesn't resolve (e.g. valid JSON with no id field) falls back to the position-based gate. A payload the extractor can't parse at all (e.g. malformed JSON when JSON was expected) raises — that means the extractor doesn't match the actual protocol, which is worth surfacing rather than silently ignoring.

Placeholders

Scrub sensitive values (tokens, API keys) from cassettes before they are written:

from nimax import Placeholder

recorder = NimaxRecorder(
    session,
    placeholders=[
        Placeholder(placeholder="<AUTH_TOKEN>", replace="Bearer secret123"),
    ],
)

Custom matchers and serializers

from nimax import BaseMatcher, NimaxRecorder


class BodyMatcher(BaseMatcher):
    name = "body"

    def match(self, recorded: dict, live: object) -> bool:
        return recorded.get("body") == live.body  # type: ignore[union-attr]


NimaxRecorder.register_matcher(BodyMatcher)

YAML cassettes are supported out of the box — use a .yaml extension for the cassette path.

Requirements

• Python ≥ 3.11
• niquests ≥ 3
• pytest ≥ 8
• PyYAML ≥ 6

License

MIT