Testing
Both Python and TypeScript SDKs ship a MockTransport for deterministic unit testing without a live server.
MockTransport that records requests and returns canned responses — no server required.
MockTransport
from zaby import Zaby
from zaby.mocks import MockTransport
transport = MockTransport()
zaby = Zaby(api_key="test-key", transport=transport)
import { Zaby } from "@zaby-ai/sdk";
import { MockTransport } from "@zaby-ai/sdk/testing";
const transport = new MockTransport([
{ method: "GET", path: "/health", json: { status: "ok" } },
]);
const zaby = new Zaby({ apiKey: "test-key", transport });
Set Responses
transport.set_response(
method="POST",
path="/api/v1/provisioning/agentic-os/agents",
status=200,
body={"id": "test-agent-id", "status": "DRAFT"},
)
import { MockTransport } from "@zaby-ai/sdk/testing";
const transport = new MockTransport([
{
method: "POST",
path: "/api/v1/provisioning/agentic-os/agents",
status: 200,
json: { id: "test-agent-id", status: "DRAFT" },
},
]);
Assert Requests
sent = transport.requests[0]
assert sent.method == "POST"
assert sent.path == "/api/v1/provisioning/agentic-os/agents"
assert sent.body == {"name": "My Agent", "slug": "my-agent"}
const sent = transport.requests[0];
console.assert(sent.method === "POST");
console.assert(sent.path === "/api/v1/provisioning/agentic-os/agents");
console.assert(sent.json?.name === "My Agent");
Reset Between Tests
// Create a fresh MockTransport per test
const transport = new MockTransport([
{ method: "GET", path: "/health", json: { status: "ok" } },
]);
Complete Test Example
import pytest
from zaby import Zaby
from zaby.mocks import MockTransport
@pytest.mark.asyncio
async def test_create_agent():
transport = MockTransport()
zaby = Zaby(api_key="test-key", transport=transport)
transport.set_response(
"POST",
"/api/v1/provisioning/agentic-os/agents",
200,
{"id": "agent_123", "name": "My Agent", "status": "DRAFT"},
)
result = await zaby.agents.create({
"name": "My Agent",
"slug": "my-agent",
})
assert result["id"] == "agent_123"
assert result["status"] == "DRAFT"
assert transport.requests[0].body["name"] == "My Agent"
import { describe, expect, it } from "vitest";
import { Zaby } from "@zaby-ai/sdk";
import { MockTransport } from "@zaby-ai/sdk/testing";
describe("create agent", () => {
it("sends the expected request and returns the mock response", async () => {
const transport = new MockTransport([
{
method: "POST",
path: "/api/v1/provisioning/agentic-os/agents",
status: 200,
json: { id: "agent_123", name: "My Agent", status: "DRAFT" },
},
]);
const zaby = new Zaby({ apiKey: "test-key", transport });
const result = await zaby.agents.create({
name: "My Agent",
slug: "my-agent",
});
expect(result.id).toBe("agent_123");
expect(result.status).toBe("DRAFT");
expect(transport.requests[0]?.json).toMatchObject({ name: "My Agent" });
});
});
Testing Streaming
Mock SSE events for testing ZabyRuntime.stream():
import pytest
from zaby import ZabyRuntime
from zaby.mocks import MockTransport
@pytest.mark.asyncio
async def test_stream():
transport = MockTransport()
runtime = ZabyRuntime(token="test-token", transport=transport)
transport.set_sse(
"GET",
"/api/v1/provisioning/managed-agents/runs/run_1/stream",
[
{"event": "text", "data": {"content": "Hello"}},
{"event": "RunFinished", "data": {"runId": "run_1"}},
],
)
messages = []
async for event in runtime.runs.stream("run_1"):
messages.append(event)
assert len(messages) == 2
assert messages[0].event == "text"
assert messages[0].data["content"] == "Hello"
import { describe, expect, it } from "vitest";
import { ZabyRuntime } from "@zaby-ai/sdk/runtime";
import { MockTransport } from "@zaby-ai/sdk/testing";
describe("stream", () => {
it("yields SSE events from the mock transport", async () => {
const stream = [
'event: text\ndata: {"content":"Hello"}\n\n',
'event: RunFinished\ndata: {"runId":"run_1"}\n\n',
].join("");
const transport = new MockTransport([
{
method: "GET",
path: "/api/v1/agent-runtime/runs/run_1/aiui",
body: stream,
headers: { "content-type": "text/event-stream" },
},
]);
const runtime = new ZabyRuntime({ token: "test-token", transport });
const messages = [];
for await (const event of runtime.runs.stream("run_1")) {
messages.push(event);
}
expect(messages).toHaveLength(2);
expect(messages[0]?.event).toBe("text");
expect(messages[0]?.data).toMatchObject({ content: "Hello" });
});
});
Best Practices
- Reset between tests — Call
transport.reset() in Python, or create a fresh MockTransport per test in TypeScript.
- Assert request paths — Verify the SDK is hitting the correct API endpoint.
- Mock error responses — Test your error handling by returning 4xx/5xx statuses.
- Use async test runners — All SDK methods are async. Mark Python tests with
@pytest.mark.asyncio; use async/await in Vitest for TypeScript.
