apitest

package
v0.4.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 25, 2026 License: AGPL-3.0 Imports: 41 Imported by: 0

Documentation

Overview

Package apitest is the integration test harness for the airlock API. Imported only from _test.go files, so testcontainers and other test-only deps never reach the production binary.

Index

Constants

View Source 
const AgentSubdomainSuffix = ".apitest.local"

AgentSubdomainSuffix is the suffix used when crafting subdomain requests: a request for slug "foo" hits Host "foo.apitest.local". Matches the AgentDomain wired into Setup's config.

View Source 
const EncryptionKey = "0001020304050607080910111213141516171819202122232425262728293031"

EncryptionKey is the 32-byte AES-256-GCM key for secrets.NewLocal, hex-encoded. Stable to match real config decoding.

View Source 
const JWTSecret = "apitest-jwt-secret-stable-across-runs"

JWTSecret is the shared HS256 secret used by every test in the package. Stable across runs so test JWTs can be hand-crafted if needed.

Variables

This section is empty.

Functions

func AddAgentMember 

func AddAgentMember(t *testing.T, h *Harness, agentID, userID uuid.UUID, role string)

AddAgentMember grants role on agentID to userID. role is "admin" or "user".

func AddSibling 

func AddSibling(t *testing.T, h *Harness, callerID, targetID, authorizingGranteeID uuid.UUID, maxAccess string)

AddSibling declares a caller-to-target A2A edge authorized by a grant the target already has for authorizingGranteeID.

func Available 

func Available() bool

Available reports whether PackageMain succeeded — i.e. tests can call Setup. Useful in skip helpers.

func CreateAgent 

func CreateAgent(t *testing.T, h *Harness, opts AgentOpts) uuid.UUID

CreateAgent inserts an agent row past the build pipeline:

  • status='active'
  • image_ref='apitest:stub'
  • db_password encrypted via Harness.Secrets (so EnsureRunning's decrypt step succeeds).
  • owner added as agent admin member.

func CreateUser 

func CreateUser(t *testing.T, h *Harness, name, role string) uuid.UUID

CreateUser inserts a user with a unique email derived from name + random suffix. Returns the user's UUID; pair with IssueUserToken to drive authenticated requests.

func IssueAgentToken 

func IssueAgentToken(t *testing.T, h *Harness, agentID uuid.UUID) string

IssueAgentToken mints an agent JWT at the row's live token version. Used by tests driving /api/agent/* endpoints directly.

func IssueUserToken 

func IssueUserToken(t *testing.T, h *Harness, userID uuid.UUID, email, role string) string

IssueUserToken creates an active first-party session and mints its access JWT with the harness JWT secret.

func MarshalEvent 

func MarshalEvent(eventType string, data map[string]any) []byte

MarshalEvent is exposed so tests can hand-craft a single NDJSON line without spinning up an Upstream — useful for one-off assertions against the parser.

func PackageMain 

func PackageMain(m interface{ Run() int }) int

PackageMain wires dbtest + s3test once per test binary and runs m.Run. The test package's TestMain should be: 

func TestMain(m *testing.M) { os.Exit(apitest.PackageMain(m)) }

If Docker is unreachable for either Postgres or MinIO, PackageMain returns the m.Run exit code with no harness state — tests that call Setup will skip individually.

func SkipIfUnavailable 

func SkipIfUnavailable(t *testing.T)

SkipIfUnavailable skips the test when the harness couldn't boot (no Docker / no external services). Matches the dbtest skip contract.

Types

type AgentOpts 

type AgentOpts struct {
	Name              string
	Slug              string
	OwnerID           uuid.UUID
	AllowPublicMCP    bool
	AllowPublicChat   bool
	AllowPublicRoutes bool
	// Stopped parks the agent at status='stopped' (image_ref still set) so
	// EnsureRunning refuses it — used to exercise the not-runnable paths.
	Stopped bool
}

AgentOpts configures a test agent. Zero values produce a minimum-viable row: status='active', image_ref='apitest:stub' so dispatcher.EnsureRunning skips the build path; db_password seeded with an encrypted dummy so the dispatcher's decrypt step succeeds.

type CommandHandler 

type CommandHandler func(s *Session) (exitCode int, err error)

CommandHandler implements the body of one exec channel session. It receives the parsed command line (everything the SSH client sent — `cmd arg1 'arg with space' ...`) plus a Session for I/O. Return value becomes the exit code (0 on nil error; ssh.ExitError preserved otherwise).

Use Session.Stdout / Session.Stderr to write output; close Session.Stdin before returning if the test sent stdin and you've consumed it.

type FakeContainerManager 

type FakeContainerManager struct {
	// contains filtered or unexported fields
}

FakeContainerManager implements container.ContainerManager backed by per-agent httptest.Server instances. Tests register an http.Handler per agent (typically built with Upstream); StartAgent returns the recorded endpoint so the dispatcher dials a real local HTTP server over the loopback. No Docker involvement.

MarkBusy/MarkIdle increment counters tests can read to assert the dispatcher correctly brackets every forwarded request — the reaper invariant (airlock/AGENTS.md Agent Execution).

func NewFakeContainerManager 

func NewFakeContainerManager() *FakeContainerManager

func (*FakeContainerManager) BusyCount 

func (m *FakeContainerManager) BusyCount(agentID uuid.UUID) int

BusyCount / IdleCount expose the MarkBusy/MarkIdle counters for assertion. Tests typically expect them equal at end-of-flow.

func (*FakeContainerManager) CaptureToolserverDiagnostics 

func (m *FakeContainerManager) CaptureToolserverDiagnostics(ctx context.Context, name, reason string) error

func (*FakeContainerManager) Close 

func (m *FakeContainerManager) Close()

Close shuts every registered httptest.Server down. Called from the harness teardown via t.Cleanup.

CloseClientConnections runs first so handlers parked on r.Context() unblock as soon as the underlying TCP socket goes away — otherwise srv.Close() blocks waiting for the handler to return and a failing test can hang the package until `go test -timeout` SIGQUITs it.

func (*FakeContainerManager) GetRunning 

func (m *FakeContainerManager) GetRunning(ctx context.Context, agentID uuid.UUID) (*container.Container, error)

func (*FakeContainerManager) IdleCount 

func (m *FakeContainerManager) IdleCount(agentID uuid.UUID) int

func (*FakeContainerManager) KillToolserver 

func (m *FakeContainerManager) KillToolserver(ctx context.Context, name string) error

func (*FakeContainerManager) LockSwap 

func (m *FakeContainerManager) LockSwap(agentID uuid.UUID) func()

LockSwap returns a no-op release fn — the apitest harness doesn't race builds with triggers, so swap serialisation has nothing to gate.

func (*FakeContainerManager) MarkBusy 

func (m *FakeContainerManager) MarkBusy(agentID uuid.UUID)

func (*FakeContainerManager) MarkIdle 

func (m *FakeContainerManager) MarkIdle(agentID uuid.UUID)

func (*FakeContainerManager) RegisterAgent 

func (m *FakeContainerManager) RegisterAgent(agentID uuid.UUID, h http.Handler, token string)

RegisterAgent wires a handler for the given agent. The handler must honour the dispatcher's HTTP contract:

  • POST /prompt with X-Run-ID and Bearer agentToken — stream NDJSON.
  • POST /__air/tool/{name} for user tool calls — return JSON.

token is the bearer the test expects upstream to validate against. Most tests pass a zero token and ignore it; the harness still puts it in the Container record for completeness.

func (*FakeContainerManager) RemoveImage 

func (m *FakeContainerManager) RemoveImage(ctx context.Context, imageRef string) error

func (*FakeContainerManager) RunningAgents 

func (m *FakeContainerManager) RunningAgents(ctx context.Context, agentIDs []uuid.UUID) (map[uuid.UUID]bool, error)

func (*FakeContainerManager) SetStopError 

func (m *FakeContainerManager) SetStopError(agentID uuid.UUID, err error)

SetStopError configures StopAgent to fail for an agent. Passing nil clears the failure so tests can exercise a successful retry.

func (*FakeContainerManager) StartAgent 

func (m *FakeContainerManager) StartAgent(ctx context.Context, opts container.AgentOpts) (*container.Container, error)

func (*FakeContainerManager) StartCount 

func (m *FakeContainerManager) StartCount(agentID uuid.UUID) int

StartCount reports how many times StartAgent was invoked for an agent — non-zero means dispatcher.EnsureRunning fired.

func (*FakeContainerManager) StartToolserver 

func (m *FakeContainerManager) StartToolserver(ctx context.Context, opts container.ToolserverOpts) (*container.Container, error)

Toolserver methods are no-ops — the harness never drives builds.

func (*FakeContainerManager) StopAgent 

func (m *FakeContainerManager) StopAgent(ctx context.Context, agentID uuid.UUID) error

func (*FakeContainerManager) StopToolserver 

func (m *FakeContainerManager) StopToolserver(ctx context.Context, name string) error

type Harness 

type Harness struct {
	T              *testing.T
	Server         *httptest.Server
	DB             *db.DB
	S3             *storage.S3Client
	Secrets        secrets.Store
	FakeContainers *FakeContainerManager
	Hub            *realtime.Hub
	PubSub         *realtime.PubSub
	Dispatcher     *trigger.Dispatcher
	BuildService   *builder.BuildService
	JWTSecret      string
}

Harness is the per-test toolbox. Build it via Setup; teardown is registered with t.Cleanup automatically.

The Server field is an httptest.NewServer wrapping the real api.NewRouter — middleware chain, CORS, auth, the lot. Tests drive HTTP requests against Server.URL.

FakeContainers lets tests register canned upstream handlers per agent before driving requests that would reach the dispatcher.

func Setup 

func Setup(t *testing.T) *Harness

Setup builds a fresh harness. It restores the DB to the post-migration snapshot first so per-test state is isolated.

The order is:

  1. SkipIfUnavailable — bail cleanly if no Docker/MinIO/PG.
  2. resetDB — DROP+CREATE WITH TEMPLATE.
  3. open pool, build S3, build all router deps.
  4. NewRouter, NewServer, register t.Cleanup.

func (*Harness) DecodeProto 

func (h *Harness) DecodeProto(resp *http.Response, dst proto.Message)

DecodeProto reads resp.Body fully and unmarshals into dst, failing the test on any error. Closes the body. Use after asserting status code on a successful response.

func (*Harness) Do 

func (h *Harness) Do(req *http.Request) *http.Response

Do executes req against the harness server, decoding any response body into out when out is non-nil and the response carries one. On non-2xx the body is returned via the (*ResponseError).Body field of the returned error — tests typically just assert status code from the returned response then read body bytes themselves.

func (*Harness) NewRequest 

func (h *Harness) NewRequest(method, path, token string, body any) *http.Request

NewRequest builds an authenticated HTTP request against the harness server. method/path are joined with Server.URL. Bearer auth is added when token is non-empty. body is encoded as JSON.

func (*Harness) NewSubdomainRequest 

func (h *Harness) NewSubdomainRequest(method, slug, path, token string, body any) *http.Request

NewSubdomainRequest is NewRequest plus a Host header of "{slug}{AgentSubdomainSuffix}". httptest.Server listens on 127.0.0.1, so SubdomainProxy reads the Host header to decide which branch to take — overriding it lets a test drive the subdomain path without DNS games.

func (*Harness) ReadBody 

func (h *Harness) ReadBody(resp *http.Response) []byte

ReadBody is DecodeProto for non-proto responses (error bodies, plain JSON). Returns the body bytes; closes the body.

type S3Params 

type S3Params struct {
	Endpoint  string
	AccessKey string
	SecretKey string
	Bucket    string
	Region    string
}

S3Params is everything an S3 client needs to point at MinIO.

type SSHTestServer 

type SSHTestServer struct {
	// contains filtered or unexported fields
}

SSHTestServer is a minimal in-process SSH server for exec-endpoint integration tests. It accepts public-key auth against a single authorized key (set via Authorize) and serves canned responses to exec channel requests defined per command.

Lifecycle: NewSSHTestServer(t) → registers t.Cleanup; tests use .Addr(), .Authorize(pubKey), and .HandleCommand(cmd, handler). No global state — every test gets a fresh server on a random port.

func NewSSHTestServer 

func NewSSHTestServer(t *testing.T) *SSHTestServer

NewSSHTestServer starts a fresh server on a random localhost port. Auto-generates a fresh ED25519 host key per server so every test exercises the TOFU code path with a unique fingerprint.

func (*SSHTestServer) Addr 

func (s *SSHTestServer) Addr() (host string, port int)

Addr returns host:port of the listening server, suitable for paste into ConfigureExecEndpointSSH.

func (*SSHTestServer) Authorize 

func (s *SSHTestServer) Authorize(openSSHPubKey string)

Authorize registers the OpenSSH-format public key authorized to log in. Tests pass the public key the operator-configure flow generated. Without this, all auth attempts fail.

func (*SSHTestServer) Close 

func (s *SSHTestServer) Close()

Close stops the listener; safe to call multiple times.

func (*SSHTestServer) Fingerprint 

func (s *SSHTestServer) Fingerprint() string

Fingerprint returns SHA256:base64 fingerprint, matching what the operator UI surfaces.

func (*SSHTestServer) HandleCommand 

func (s *SSHTestServer) HandleCommand(command string, fn CommandHandler)

HandleCommand registers a handler for an exact command line. The SSH client sends the full command as one string (the result of JoinCommand on the airlock side), so match on the joined form: `kick-build --branch main` rather than the {cmd, args} pair.

HandleCommandPrefix is the looser alternative when only the verb matters.

func (*SSHTestServer) HandleDefault 

func (s *SSHTestServer) HandleDefault(fn CommandHandler)

HandleDefault registers a fallback handler invoked for any command without an exact match. Useful for "always succeed" smoke handlers.

func (*SSHTestServer) HostKeyOpenSSH 

func (s *SSHTestServer) HostKeyOpenSSH() string

HostKeyOpenSSH returns the server's host key in OpenSSH wire format, suitable for direct comparison or pinning. Tests use this to assert the host-key TOFU code path captured the right key.

func (*SSHTestServer) String 

func (s *SSHTestServer) String() string

CommandsHandled returns the number of exec requests this server has served. Useful in tests that need to assert cache reuse vs cold dial (we expect one exec per call regardless of caching, but it's a secondary signal worth having). Reads are not synchronized with in-flight serveChannel goroutines; call after the test's exec calls have completed.

type Session 

type Session struct {
	Command string         // full command line as sent on the SSH exec request
	Stdin   io.Reader      // stdin bytes from the client
	Stdout  io.WriteCloser // canonical stdout pipe
	Stderr  io.WriteCloser // canonical stderr pipe
}

Session carries the per-exec I/O bundle. Stdin is provided by the SSH client (the agent under test); Stdout/Stderr feed back to the client. Close cleans up if the handler bails early — tests typically won't call it explicitly.

type Upstream 

type Upstream struct {
	// contains filtered or unexported fields
}

Upstream builds an http.Handler that imitates an agent container's POST /prompt endpoint by streaming NDJSON events.

Events match the wire format that `trigger.StreamNDJSONResponse` reads ([trigger/prompt.go:716+](airlock/trigger/prompt.go#L716)). Each event is a single JSON line: {"type":"<kind>","data":{...}}.

Method-chain API: 

apitest.NewUpstream().
    TextDelta("Hello ").TextDelta("world").
    ToolCall("id-1", "add", `{"a":1,"b":2}`).
    ToolResult("id-1", "add", `3`).
    Finish().
    Handler()

Per-event delay simulates streaming pace; defaults to 0.

func NewUpstream 

func NewUpstream() *Upstream

func (*Upstream) ConfirmationRequired 

func (u *Upstream) ConfirmationRequired(toolCallID, permission, code string, patterns ...string) *Upstream

ConfirmationRequired mirrors the operator-approval signal an agent emits before suspending for a tool call.

Note: confirmation_required alone does NOT end the run on airlock's side — pair it with Suspend() to mark the run paused.

func (*Upstream) Finish 

func (u *Upstream) Finish() *Upstream

Finish emits the run-completion event with zero token usage. Tests that care about usage accounting should use FinishWithUsage.

func (*Upstream) FinishWithUsage 

func (u *Upstream) FinishWithUsage(inputTokens, outputTokens int) *Upstream

func (*Upstream) Handler 

func (u *Upstream) Handler() http.Handler

Handler returns an http.Handler that streams the configured events on POST /prompt, or 405 on any other path/method. The response is NDJSON: one JSON object per line, flushed after each line so the dispatcher's bufio.Scanner sees streaming behaviour.

func (*Upstream) Raw 

func (u *Upstream) Raw(eventType string, data map[string]any) *Upstream

Raw lets a test push an event the convenience methods don't cover.

func (*Upstream) Suspend 

func (u *Upstream) Suspend(reason string) *Upstream

Suspend emits the terminal-suspended marker. Use after ConfirmationRequired so airlock keeps the WS replay buffer (a completion would clear it).

func (*Upstream) TextDelta 

func (u *Upstream) TextDelta(text string) *Upstream

func (*Upstream) ToolCall 

func (u *Upstream) ToolCall(id, name, inputJSON string) *Upstream

func (*Upstream) ToolError 

func (u *Upstream) ToolError(id, name, errorText string) *Upstream

func (*Upstream) ToolResult 

func (u *Upstream) ToolResult(id, name, outputJSON string) *Upstream

func (*Upstream) WithDelay 

func (u *Upstream) WithDelay(d time.Duration) *Upstream

WithDelay inserts a pause between events to simulate token-by-token streaming. Keep test delays under 1ms unless a test is specifically asserting timing behaviour.

type WSClient 

type WSClient struct {
	// contains filtered or unexported fields
}

WSClient is a test-friendly wrapper around coder/websocket that connects to a running httptest.Server hosting the airlock router, authenticates with a user JWT, and exposes a Next/Drain API for asserting envelope sequences.

The WS upgrade handler auto-subscribes the connection to every agent the user is a member of, so a test only needs to ensure membership rows exist before calling Connect.

func Connect 

func Connect(t *testing.T, srv *httptest.Server, jwt string, since uint64) *WSClient

Connect dials srv.URL/ws[?since=<seq>] with the HttpOnly-cookie equivalent, performs the WebSocket upgrade, and starts a background goroutine that reads envelopes into a buffered channel. The connection closes via t.Cleanup; tests do not call Close directly.

func (*WSClient) Drain 

func (c *WSClient) Drain(window time.Duration) []realtime.Envelope

Drain returns every envelope that arrives within window. Use when a test expects a known count of events but doesn't care about exact ordering versus other concurrent flows.

func (*WSClient) Next 

func (c *WSClient) Next(timeout time.Duration) realtime.Envelope

Next blocks until the next envelope arrives, or fails the test on timeout. Use for asserting "the next event is X" semantics.

func (*WSClient) WaitFor 

func (c *WSClient) WaitFor(timeout time.Duration, pred func(realtime.Envelope) bool) realtime.Envelope

WaitFor blocks until an envelope satisfying pred arrives, or fails the test on timeout. Earlier envelopes that don't match are discarded — use Drain if a test needs the full sequence.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.