Metadata-Version: 2.4
Name: mimiko
Version: 0.1.0
Summary: Remote WebSocket Mock built on top of jj
Home-page: https://github.com/teka1905/mimiko
Author: Alexey Dolmatov
Author-email: teka1905@gmail.com
License: Apache-2.0
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.8
Description-Content-Type: text/markdown
License-File: LICENSE
License-File: NOTICE
Requires-Dist: jj<3,>=2.14
Requires-Dist: aiohttp<4,>=3.5
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: license
Dynamic: license-file
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# 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`](https://pypi.python.org/pypi/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

```bash
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

```bash
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.

  ```bash
  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:

  ```bash
  # 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.

```python
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):

```python
(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:

```python
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:

```python
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`:

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

### Low-level API

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

```python
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`.
