ClaudeWrapper.Test (ClaudeWrapper v0.12.0)

Copy Markdown View Source

Drive a ClaudeWrapper.DuplexSession against an in-process double, with no claude binary and no network.

start_session/1 returns a live session wired to ClaudeWrapper.DuplexSession.Adapter.Test plus a stub handle you feed NDJSON events to. The session behaves exactly as it does over a real subprocess -- same send/3, same subscriber events, same Result -- so it is a faithful way to test code that drives claude_wrapper.

Each start_session/1 gets its own stub process and nothing is shared between sessions, so async: true tests are safe.

Example

{:ok, session, stub} = ClaudeWrapper.Test.start_session()

# `send/3` blocks until the turn's result, so drive it from a task.
task = Task.async(fn -> ClaudeWrapper.DuplexSession.send(session, "hi") end)

ClaudeWrapper.Test.emit(stub, [
  ClaudeWrapper.Test.text_delta("Hello")
])
ClaudeWrapper.Test.emit_result(stub, result: "Hello", session_id: "s-1")

{:ok, %ClaudeWrapper.Result{session_id: "s-1"}} = Task.await(task)

Canned replies

For the common "send a prompt, get a fixed answer" shape, queue a reply and it is emitted automatically on the next send/3:

ClaudeWrapper.Test.stub(stub, [
  ClaudeWrapper.Test.text_delta("Hi"),
  ClaudeWrapper.Test.result(result: "Hi")
])

{:ok, _result} = ClaudeWrapper.DuplexSession.send(session, "anything")

Summary

Types

line()

A line to feed the session: a map (JSON-encoded) or a raw string.

Functions

emit(stub, lines)

Feed NDJSON lines to the session now, as if the subprocess emitted them. Each line is a map (JSON-encoded) or a raw string.

emit_result(stub, attrs \\ [])

Emit a result event, ending the current turn.

exit_status(stub, code)

Simulate the subprocess exiting with code.

result(attrs \\ [])

A minimal success result event; attrs override the defaults.

start_session(opts \\ [])

Start a DuplexSession backed by the test adapter.

stub(stub, lines)

Queue a one-shot canned reply emitted automatically on the next DuplexSession.send/3. Use for the "prompt in, fixed answer out" shape.

text_delta(text, index \\ 0)

A stream_event carrying a text delta.

thinking_delta(text, index \\ 0)

A stream_event carrying an extended-thinking delta.

Types

line()

@type line() :: map() | String.t()

A line to feed the session: a map (JSON-encoded) or a raw string.

Functions

emit(stub, lines)

@spec emit(pid(), [line()] | line()) :: :ok

Feed NDJSON lines to the session now, as if the subprocess emitted them. Each line is a map (JSON-encoded) or a raw string.

emit_result(stub, attrs \\ [])

@spec emit_result(
  pid(),
  keyword()
) :: :ok

Emit a result event, ending the current turn.

attrs are merged into a minimal success result (see result/1).

exit_status(stub, code)

@spec exit_status(pid(), integer()) :: :ok

Simulate the subprocess exiting with code.

result(attrs \\ [])

@spec result(keyword()) :: map()

A minimal success result event; attrs override the defaults.

start_session(opts \\ [])

@spec start_session(keyword()) :: {:ok, pid(), pid()}

Start a DuplexSession backed by the test adapter.

Returns {:ok, session, stub} where stub is the controller handle you pass to emit/2, emit_result/2, and stub/2.

Options:

  • :config -- a ClaudeWrapper.Config.t/0 (a dummy is used by default; the test adapter never launches a binary)
  • :on_permission -- forwarded to DuplexSession.start_link/1
  • any other DuplexSession option except :adapter/:adapter_opts

stub(stub, lines)

@spec stub(pid(), [line()] | line()) :: :ok

Queue a one-shot canned reply emitted automatically on the next DuplexSession.send/3. Use for the "prompt in, fixed answer out" shape.

text_delta(text, index \\ 0)

@spec text_delta(String.t(), non_neg_integer()) :: map()

A stream_event carrying a text delta.

thinking_delta(text, index \\ 0)

@spec thinking_delta(String.t(), non_neg_integer()) :: map()

A stream_event carrying an extended-thinking delta.