Python Testing SDK (digitorn.testing)

The digitorn.testing package is the canonical client library for writing live integration tests against a running Digitorn daemon. It is shipped inside the digitorn PyPI package; no separate install.

The SDK is a library, not a test runner. You write your own scenarios under tools/live_tests/<feature>_scenarios (or any other location) and import the SDK as a consumer would.

Public surface

from digitorn.testing import (
    DevClient,
    AppHandle,
    LiveEvent,
    LiveEventStream,
    SessionHandle,
    TurnResult,
    assertions,
)

Symbol	Purpose
DevClient	Top-level client. Carries auth, talks REST + Socket.IO to the daemon.
SessionHandle	Lightweight value object identifying (app_id, session_id, daemon_url, workspace). Constructed locally.
AppHandle	Same shape for apps.
LiveEvent	One Socket.IO event envelope as a dataclass with seq, type, payload.
LiveEventStream	A live tap on a session's Socket.IO stream. Buffer + helpers.
TurnResult	Final outcome of an agent turn.
assertions	Sub-module with sort_by_seq, event_order, ... helpers.

Connecting

from digitorn.testing import DevClient

client = DevClient(
    daemon_url=None,                      # default: $DIGITORN_DEV_DAEMON_URL → $DIGITORN_DAEMON_URL → http://127.0.0.1:8000
    auto_approve=True,                    # auto-approves every pending capability prompt
    timeout=3600.0,                       # default request timeout (1 h)
    token=None,                           # explicit JWT - none = anonymous; pair with .login() / .register()
)

The constructor does not auto-load a token from disk. Pass token= explicitly, or call client.login(email, password) / DevClient.with_user(email, password) to obtain one. For scripted CI, set DIGITORN_DAEMON_URL and pass token=....

Two flow shapes

One-shot chat (REST polling)

result = client.chat(app_id="my-app", message="Hello!")
print(result.content)

chat is the simple synchronous loop. Internally it creates a session, POSTs the message, polls GET /sessions/{sid}, and auto-approves pending prompts. Returns a TurnResult after message_done.

Use this when you don't care about per-event hooks.

Live event stream (Socket.IO)

import uuid
from digitorn.testing import DevClient, SessionHandle

client = DevClient(daemon_url="http://127.0.0.1:8000", auto_approve=True)
session = SessionHandle(
    session_id=f"test-{uuid.uuid4().hex[:8]}",
    app_id="my-app",
    daemon_url="http://127.0.0.1:8000",
    workspace="",
)

stream = client.send_live(session, "ping", total_timeout=60.0)
try:
    events = stream.events()
    for ev in events:
        print(ev["type"], ev.get("payload", {}))
finally:
    stream.stop(timeout=2.0)

send_live POSTs the message AND opens a live Socket.IO stream in one call. It waits for message_done (or total_timeout) before returning the stream so stream.events is complete.

This is the path you want for per-event assertions (tool calls, hooks fired, intermediate states).

Manual control

When you need finer control (e.g. multiple POSTs on one stream):

# Step 1 - POST without waiting
client.post_message_raw(session, "ping")

# Step 2 - wait for the daemon to register the session
client.wait_for_session(session, timeout=20.0)

# Step 3 - open the stream without waiting (already waited)
from digitorn.testing.events import LiveEventStream
stream = LiveEventStream(
    daemon_url=client.daemon_url,
    token=client._get_auth_token(),
    app_id=session.app_id,
    session_id=session.session_id,
)
stream.start()

# Step 4 - wait for an arbitrary event type
done = stream.wait_for("message_done", timeout=45.0)

Assertions

from digitorn.testing import assertions

events = assertions.sort_by_seq(stream.events())

ok, detail = assertions.event_order(
    events, ["user_message", "message_started", "tool_start", "tool_call", "message_done"]
)
assert ok, detail

assertions.sort_by_seq orders by the envelope's monotonic seq field (Socket.IO does NOT guarantee delivery order across namespaces, but seq does within a session).

assertions.event_order checks that the listed event types appear in the given order. Returns (ok, detail).

Other helpers worth knowing:

Helper	Purpose
assertions.contains_text	The buffered out_token content concatenated includes a substring.
assertions.tool_called	A tool_call with the given name was emitted.
assertions.no_errors	No error-typed event landed.

Approvals

auto_approve=True (the default in tests) makes the client poll GET (apps API) every second and resolve every pending request. For tests of approval flows themselves, pass auto_approve=False and call:

pending = client.list_pending_approvals(app_id)
client.resolve_approval(app_id, request_id, approved=True)

Auth

Pass a JWT explicitly via token= to the constructor, or use the helpers:

client = DevClient.with_token(token)
client = DevClient.with_user(email, password)              # login
client = DevClient.with_user(email, password, register_if_missing=True)

You can also call client.login(email, password) or client.register(email, password) on an existing anonymous client; both return the token dict and store the access token on the instance. In CI, mint a short-lived token via the auth API and pass it explicitly.

Where scenarios live

By convention, live test scenarios go under tools/live_tests/<feature>_scenarios in the source repo. Each scenario function returns a tuple (ok: bool, detail: str, artifacts: dict). The runner in drives them.

The digitorn.testing package itself contains no scenarios. Adding a scenario inside this package is a violation of the SDK's design - it is meant to stay a thin, reusable library.