mimiko

mimiko is a remote WebSocket mock library. It lets you declare WebSocket endpoints, script what the server sends on connect and in reaction to incoming messages, drive live sessions from your test, and read a bidirectional history of frames.

It is a companion package built on top of jj (a remote HTTP mock). mimiko reuses jj's matchers, resolver, remote protocol (/__jj__/register|deregister|history|reset) and server runner, and adds a WebSocket layer on top. jj itself is used unmodified, as a dependency.

Installation

pip install mimiko

This installs jj as a dependency. WebSocket support requires no extra packages beyond aiohttp (already required by jj).

Running the mock server

python -m mimiko -p 8080
# or, after install, via the console script:
mimiko -p 8080

The client talks to the server over HTTP. Point it at the server with an environment variable or the mock_url= argument. The address is resolved in this order of priority (highest first):

1. mock_url= argument to mocked_ws(...).
2. MIMIKO_REMOTE_MOCK_URL — the WebSocket-mock-specific override.
3. JJ_REMOTE_MOCK_URL — jj's shared variable (default http://localhost:8080).

Both variables are read from the environment on every call (not cached at import time), so they stay compatible with load_dotenv()-style setups where the .env is loaded after modules are imported.

Two deployment modes

• Single server (HTTP + WS). Run one mimiko server and let it serve everything. Leave MIMIKO_REMOTE_MOCK_URL unset and point JJ_REMOTE_MOCK_URL at it (or rely on the default). Behavior is identical to plain jj.

  export JJ_REMOTE_MOCK_URL=http://localhost:8080
  

• Separate WS process next to jj. Run the regular jj server for HTTP mocks and a standalone mimiko process for WebSocket mocks on a different port. Set both variables so HTTP goes to jj and WebSocket goes to mimiko without passing mock_url= in every call:

  # jj serves HTTP mocks on :5001
  export JJ_REMOTE_MOCK_URL=http://localhost:5001
  # mimiko serves WebSocket mocks on :8080
  export MIMIKO_REMOTE_MOCK_URL=http://localhost:8080
  mimiko -p 8080
  

Quick start

Scenarios are built with a small declarative builder. on_open(...) scripts the messages sent right after the connection opens; on(...) adds reactions matched against incoming messages.

import mimiko
from mimiko import WebSocketScenario, mocked_ws

scenario = (
    WebSocketScenario()
    .on_open(send=[
        {"type": "LOAD_DEFAULT_FILTERS_TICKETS_PREVIEWS"},
        {"type": "LOAD_QUEUES_INFO"},
        {"type": "LOAD_FILTERS"},
        {"type": "LOAD_TICKETS_PREVIEWS"},
    ])
    .on("SET_ACTIVE_TICKET", send={"type": "ACTIVE_TICKET_SET"})
)

async with mocked_ws("/api/wss/session/v2", scenario) as ws_mock:
    # ... run the system under test; it connects, receives the bootstrap
    #     sequence above, then sends SET_ACTIVE_TICKET ...

    # Assert on what the SUT sent to the mock (incoming frames):
    incoming = await ws_mock.wait_for_messages(1, timeout=5)
    assert incoming[0]["type"] == "SET_ACTIVE_TICKET"

Matching incoming messages

Reactions match by message type (a string) and/or a subset of payload (dict-contains, applied recursively):

(WebSocketScenario()
 .on("PING", send={"type": "PONG"})                                  # by type
 .on(where={"type": "SET_ACTIVE_TICKET", "payload": {"id": 42}},     # by type + payload subset
     send={"type": "ACTIVE_TICKET_SET"}))

The first matching reaction wins. Unmatched incoming messages are still recorded in history, but produce no reply.

Send steps, delays and policies

send= accepts a dict (sent as JSON), a str (raw text), bytes (binary), or a list mixing these with explicit step helpers:

from mimiko import send_json, send_text, send_bytes, send_malformed, close, silence

(WebSocketScenario()
 .on_open(send=[
     {"type": "READY"},
     send_json({"type": "BOOT"}, delay=0.1),   # per-step delay (seconds)
 ])
 .on("BAD", send=send_malformed("{not json"))  # send intentionally invalid JSON
 .on("QUIET", send=silence())                  # match, but send nothing (stay open)
 .on("BYE", send={"type": "CLOSING"}, close=1001))  # reply, then close with a code

Live control

You can drive a connected session from the test side at any time:

async with mocked_ws("/api/wss/session/v2", scenario) as ws_mock:
    # push a message to every connected session for this mock
    await ws_mock.push({"type": "SERVER_EVENT", "payload": {"n": 7}})
    # close every connected session
    await ws_mock.close(code=1000, reason="done")

History

Every frame is recorded in chronological order with a direction (in/out), type, payload, raw and a timestamp ts:

history = await ws_mock.fetch_history()   # List[WsHistoryItem]
for item in history:
    print(item["direction"], item["type"], item["payload"])

Keeping the socket warm (keepalive)

Browser-side heartbeats treat silence as a dead connection. If the mock says nothing after its bootstrap, the frontend closes the socket after a few missed pings and reconnects — the handshake lands on the same handler, on_open is replayed, and the bootstrap appears in the history a second time. Meanwhile the application re-initialises its stores in the middle of your test. Set the interval below the frontend's ping timeout:

scenario = (
    WebSocketScenario()
    .on_open(send=[{"type": "LOAD_TICKETS"}])
    .keepalive(every=5)                       # default frame: the text "ping"
    .keepalive(every=5, send={"type": "PING"})  # ...or your own
)

Measured against a real application: without keepalive the history holds one bootstrap at t=35s and two by t=45s; with keepalive(every=5) it stays at one through t=70s.

Two properties worth knowing:

• It does not block incoming frames. The keepalive runs as a task alongside the receive loop. Putting a delay inside on_open instead would not work — on_open is replayed before the loop starts, so a delayed chain there stalls reception.
• Its frames are not recorded in the history. They are transport plumbing, not application protocol; recording them would drown the report and break assertions over frame order.

Low-level API

The builder compiles to a packable WebSocketResponse; you can construct it directly and register it like any jj remote mock:

import jj
from mimiko import WebSocketResponse, mocked_ws

response = WebSocketResponse(
    on_open=[{"type": "READY"}],
    reactions=[{"where": {"type": "PING"}, "send": {"type": "PONG"}}],
)
async with mocked_ws(jj.match("GET", "/ws"), response):
    ...

What mimiko does not do

• It is transport-neutral / schema-agnostic. mimiko does not know your message schemas (e.g. WSTicketEvent); validate those on the test side (for example with d42 schemas). Matching is by type string and payload subset only.
• Live push/close require an open session. Sessions are tracked per handler id; if several sessions are connected, push/close fan out to all of them. A push to a mock with no live session is delivered to zero sessions.
• No protocol negotiation / subprotocols / ping-pong control frames beyond what aiohttp handles by default.
• No stateful branching logic. Reactions are stateless first-match rules; anything requiring server-side state across messages should be driven from the test via live push.
• Malformed JSON / silence are opt-in directives, not automatic behaviors.

Origin and license

mimiko is derived work built on top of jj (Apache-2.0, authored by Nikita Tsvetkov). jj is used unmodified and declared as an external dependency; all code in this repository is new and lives in the mimiko/ package. mimiko subclasses and builds upon jj's public API (WsMock(Mock), matchers, resolver, remote protocol, server runner).

mimiko is licensed under Apache-2.0 (see LICENSE). Attribution to jj is recorded in NOTICE.