Skip to content
Documentation
Reference
Go SDK
dagnatstest
API

dagnatstest

import "github.com/danmestas/dagnats/dagnatstest"

dagnatstest/cli_fixture.go CLIFixture drives the dagnats CLI in-process against the fixture’s embedded NATS server. Stdout/stderr are captured by redirecting the process-global file descriptors during the call. The CLI calls os.Exit on errors; the fixture intercepts that via the ExitInterceptor wired by the caller (typically the cli package itself, which owns the os.Exit indirection).

The fixture takes the CLI entry point as a function value rather than importing the cli package directly. This avoids a package import cycle (cli tests already import dagnatstest, so dagnatstest must not import cli). The cli package wires the bridge by passing cli.Run and cli.SwapExitFunc into NewCLIFixture.

dagnatstest/cluster.go In-process N-node NATS cluster helper for tests. Spins up real nats-server instances configured as a cluster, waits for routes to mesh and a JetStream meta-leader to be elected, then confirms the cluster is API-healthy via the production WaitForClusterQuorum helper. Cleanup registered with t.Cleanup so tests stay leak-free.

Package dagnatstest provides test helpers for DagNats workflows. It starts an embedded NATS server with all required streams and KV buckets in a single call, ready for workflow testing.

Usage:

func TestMyWorkflow(t *testing.T) {
    nc := dagnatstest.Server(t)
    // nc is ready — register workflows, start workers, etc.
}

dagnatstest/dlq_fixture.go DLQFixture clusters DLQ-related test helpers off a shared Harness. Construct via NewDLQFixture(h). Helpers stay under 70 lines (TigerStyle).

Concern boundary: the Harness owns embedded NATS lifecycle; the fixture owns DLQ-stream observation and DLQ-publish drivers. The fixture never mutates the Harness beyond reading its NATS connection.

dagnatstest/fixtures.go Pre-built workflow definitions for common DAG topologies. Eliminates boilerplate in tests that rebuild linear, fan-out, fan-in, and diamond patterns from scratch every time.

dagnatstest/harness.go Test harness that bundles NATS server, orchestrator, API service, and worker into a single struct. Eliminates ~15 lines of boilerplate that every integration test otherwise duplicates.

dagnatstest/helpers.go Convenience helpers for workflow integration tests. RunAndWait and WaitForStatus eliminate the poll-loop boilerplate that every test otherwise duplicates.

dagnatstest/log_capture.go LogCapture installs a process-wide slog handler that records every log record emitted while the test runs. The capture is restored on t.Cleanup, so tests cannot leak captured state into siblings.

Motivation: engine/worker code paths emit warnings via slog package functions (slog.WarnContext, slog.InfoContext) which always route through slog.Default(). To assert “log fires exactly once” we swap the default handler with a capturing one for the duration of the test. See PR 5 of the AFK plan (issue #195) for the consuming test.

dagnatstest/run_fixture.go RunFixture clusters run-population helpers around a shared Harness. Keeping these off Harness lets the harness stay a thin lifecycle owner; per-concern helpers cluster by concern, not by accretion (the inventory pattern documented in the AFK plan).

dagnatstest/setup.go Higher-level setup helpers that eliminate boilerplate when tests need a registered workflow or a running worker. These build on Server(), RunAndWait(), and WaitForStatus() from helpers.go.

Index

func DiamondDef 

func DiamondDef(t *testing.T) dag.WorkflowDef

DiamondDef builds the classic diamond: A -> {B, C} -> D.

func FailHandler 

func FailHandler(msg string) worker.HandlerFunc

FailHandler returns a HandlerFunc that always fails permanently with the given message. Panics if msg is empty.

func FanInDef 

func FanInDef(t *testing.T, n int) dag.WorkflowDef

FanInDef builds a workflow with root -> n branches -> join: root -> {branch-0, …, branch-(n-1)} -> join.

func FanOutDef 

func FanOutDef(t *testing.T, n int) dag.WorkflowDef

FanOutDef builds a workflow with 1 root and n parallel branches: root -> {branch-0, branch-1, …, branch-(n-1)}.

func HandleTypedOn 

func HandleTypedOn[I, O any](h *Harness, t *testing.T, taskType string, fn worker.TypedHandlerFunc[I, O])

HandleTypedOn registers a typed handler on the harness worker. Must be a package-level function because Go generics cannot appear on methods.

func LinearDef 

func LinearDef(t *testing.T, n int) dag.WorkflowDef

LinearDef builds a workflow with n steps in sequence: task-0 -> task-1 -> … -> task-(n-1). Each step uses task type “task-{i}”.

func PassHandler 

func PassHandler() worker.HandlerFunc

PassHandler returns a HandlerFunc that completes immediately, passing the input through as output.

func RunAndWait 

func RunAndWait(t *testing.T, svc *api.Service, workflow string, input []byte, timeout time.Duration) dag.WorkflowRun

RunAndWait starts a workflow run and blocks until it reaches any terminal status (Completed, Failed, Cancelled, Compensated, CompensateFailed). Returns the final WorkflowRun snapshot. Fatals the test if the run does not finish within timeout.

func Server 

func Server(t *testing.T) *nats.Conn

Server starts an embedded NATS server with JetStream and all required streams/KV buckets provisioned. Returns the connected client. Server and connection are cleaned up automatically when the test ends.

func StartTestCluster 

func StartTestCluster(t *testing.T, n int) *nats.Conn

StartTestCluster starts n in-process NATS servers configured as a cluster, waits for routes to mesh and quorum to form, and returns a connection to peer 0. All cleanup is registered with t.Cleanup. Panics if n < 3 or n > 5.

func WaitForStatus 

func WaitForStatus(t *testing.T, svc *api.Service, runID string, timeout time.Duration, statuses ...dag.RunStatus) dag.WorkflowRun

WaitForStatus polls svc.GetRun every 25ms until the run reaches one of the given target statuses. Returns the matching snapshot. Fatals the test with a descriptive message on timeout.

func Worker 

func Worker[I, O any](t *testing.T, nc *nats.Conn, taskType string, fn worker.TypedHandlerFunc[I, O]) *worker.Worker

Worker starts a worker with a single typed handler and stops it automatically when the test ends. Uses generics so callers get compile-time type safety on handler input and output. Must be a package-level function (not a method) because Go does not allow generic methods.

func Workflow 

func Workflow(t *testing.T, svc *api.Service, builder *dag.WorkflowBuilder) dag.WorkflowDef

Workflow builds and registers a workflow definition, failing the test immediately on any error. Returns the built definition so callers can inspect the name or steps if needed.

type CLIFixture

CLIFixture drives the dagnats CLI in-process. Constructed off a Harness; the harness owns NATS lifecycle, the fixture owns the shape of “execute the CLI and observe its output”.

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

func NewCLIFixture 

func NewCLIFixture(h *Harness, runCLI CLIRunner, swapExit ExitSwapper) *CLIFixture

NewCLIFixture constructs a CLIFixture. runCLI is typically cli.Run and swapExit is typically cli.SwapExitFunc — the caller injects them so this package does not import cli (which already imports this one via tests).

func (*CLIFixture) Run 

func (f *CLIFixture) Run(t *testing.T, args ...string) string

Run executes the dagnats CLI with the given args. On non-zero exit it fatals the test. Returns stdout (stderr discarded — use RunSplit when stderr matters).

func (*CLIFixture) RunErr 

func (f *CLIFixture) RunErr(t *testing.T, args ...string) (string, error)

RunErr executes the CLI and returns stdout plus any non-nil error produced by a non-zero exit. Stdout may be partially populated even when err is non-nil.

func (*CLIFixture) RunSplit 

func (f *CLIFixture) RunSplit(t *testing.T, args ...string) (string, string)

RunSplit executes the CLI and returns stdout + stderr separately. Fatals on non-zero exit; for negative-path tests use RunErr.

type CLIRunner

CLIRunner is the in-process entry point for the dagnats CLI. The concrete implementation is cli.Run, but the fixture takes it as a value to break the cli ↔ dagnatstest import cycle.

type CLIRunner func(args []string)

type DLQFixture

DLQFixture holds DLQ-focused test helpers bound to a Harness. Use NewDLQFixture(h) to construct.

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

func NewDLQFixture 

func NewDLQFixture(h *Harness) *DLQFixture

NewDLQFixture binds a DLQFixture to an existing Harness. Panics if h is nil.

func (*DLQFixture) AwaitReplay 

func (f *DLQFixture) AwaitReplay(t *testing.T, d time.Duration) <-chan []byte

AwaitReplay subscribes to the task subject space, blocks up to d for the next delivery, and returns the message body via a 1-buffer channel. nil means timeout. Install *before* Replay so the subscription catches the republished message.

func (*DLQFixture) Count 

func (f *DLQFixture) Count(ctx context.Context) (int, error)

Count returns the current number of entries on the DEAD_LETTERS stream. Reads stream info directly — no consumer needed.

func (*DLQFixture) IsBodyMissingError 

func (f *DLQFixture) IsBodyMissingError(err error) bool

IsBodyMissingError reports whether err is the typed body-missing error returned by Replay against a legacy DLQ entry.

func (*DLQFixture) PublishAndExhaustToDLQ 

func (f *DLQFixture) PublishAndExhaustToDLQ(t *testing.T, body []byte) uint64

PublishAndExhaustToDLQ drives a workflow whose step is failed non-retriably so the engine publishes a DLQ entry, and returns the DLQ stream sequence of that entry. The given body is used as the workflow input bytes; after the fix it must also appear on the DLQ entry’s Body field (the TaskPayload bytes containing this input). Drives via direct step.failed publish on history — mirrors orchestrator_test.go’s shape for deterministic exhaustion.

Side effect: drains the original task message from TASK_QUEUES so a subsequent AwaitReplay sees only the republished delivery.

func (*DLQFixture) PublishDLQNoMsgID 

func (f *DLQFixture) PublishDLQNoMsgID(ctx context.Context, subject string, body []byte) error

PublishDLQNoMsgID publishes a synthetic DLQ entry on `subject` with no Nats-Msg-Id header — mirrors the pre-fix production code path in engine.RecoveryManager.PublishDeadLetter. Used to demonstrate that without dedup, repeated publishes for the same logical event all land.

func (*DLQFixture) PublishDLQWithMsgID 

func (f *DLQFixture) PublishDLQWithMsgID(ctx context.Context, subject string, msgID string, body []byte) error

PublishDLQWithMsgID publishes a synthetic DLQ entry on subject `dead.<task>` with the given Nats-Msg-Id header for dedup testing. Returns the publish error so callers can assert success/failure.

Used by tests that exercise the dedup contract directly without driving the full workflow-failure path.

func (*DLQFixture) PublishLegacyEntry 

func (f *DLQFixture) PublishLegacyEntry(t *testing.T) uint64

PublishLegacyEntry writes a synthetic DLQ entry in the pre-fix shape: payload is just {run_id, step_id, task} with no Body field. Returns the DLQ stream sequence so the caller can replay it.

func (*DLQFixture) Replay 

func (f *DLQFixture) Replay(ctx context.Context, seq uint64) error

Replay invokes the API-layer replay for the given DLQ sequence. Returns the error verbatim so callers can match against ErrDLQBodyMissing or assert nil.

func (*DLQFixture) Seed 

func (f *DLQFixture) Seed(t *testing.T, n int)

Seed publishes n synthetic DLQ entries with deterministic body bytes (“seed-<i>”) and a populated metadata header set so #203’s CLI tests can assert on truncation, --all behavior, and the visibility of the delivery_count + consumer fields in --json output.

Bounded: n must be in (0, seedMax]. Helpers stay under 70 lines so the publish loop delegates to seedOne.

func (*DLQFixture) WaitForCount 

func (f *DLQFixture) WaitForCount(t *testing.T, target int, timeout time.Duration) int

WaitForCount polls Count() until it returns target or timeout. Bounded wait per CLAUDE.md testing rules — fails the test on timeout. Returns the final observed count.

type ExitSwapper

ExitSwapper installs an exit hook and returns the previous one. It mirrors cli.SwapExitFunc and is passed in for the same cycle reason.

type ExitSwapper func(next func(int)) func(int)

type FailRetryAfterCall

FailRetryAfterCall records a FailRetryAfter call.

type FailRetryAfterCall struct {
    Err   error
    After time.Duration
}

type Harness

Harness holds all components needed for a workflow integration test. Fields are public so tests can reach into them for assertions (e.g., h.Svc.GetRun()).

type Harness struct {
    NC     *nats.Conn
    Engine *engine.Orchestrator
    Svc    *api.Service
    Worker *worker.Worker
}

func NewHarness 

func NewHarness(t *testing.T) *Harness

NewHarness starts an embedded NATS server, orchestrator, and API service. The worker is created but NOT started — register handlers first, then call h.Start(t).

func (*Harness) Handle 

func (h *Harness) Handle(t *testing.T, taskType string, fn worker.HandlerFunc)

Handle registers a raw handler on the harness worker. Call before h.Start(t).

func (*Harness) RegisterAndRun 

func (h *Harness) RegisterAndRun(t *testing.T, def dag.WorkflowDef, input []byte, timeout time.Duration) dag.WorkflowRun

RegisterAndRun registers a workflow definition, starts a run, and blocks until it reaches a terminal status. Returns the final WorkflowRun snapshot.

func (*Harness) Start 

func (h *Harness) Start(t *testing.T)

Start starts the worker and registers cleanup. Call after all handlers are registered.

type LogCapture

LogCapture is a thread-safe, in-memory slog handler that records every record emitted via slog package functions while it is the default. Hits(substr) counts records whose message contains substr.

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

func NewLogCapture 

func NewLogCapture(t *testing.T) *LogCapture

NewLogCapture installs a fresh LogCapture as the slog default and arranges for the prior default to be restored on t.Cleanup. Panics on nil t — programmer error.

func (*LogCapture) Enabled 

func (c *LogCapture) Enabled(context.Context, slog.Level) bool

Enabled accepts every record so callers’ WarnContext / DebugContext hits land in the capture regardless of the prior log level.

func (*LogCapture) Handle 

func (c *LogCapture) Handle(_ context.Context, r slog.Record) error

Handle records the slog.Record under the mutex. The Record’s Clone() is unnecessary here: slog.Record is a struct value, and we only read its Message field.

func (*LogCapture) Hits 

func (c *LogCapture) Hits(substr string) int

Hits returns the number of captured records whose message contains substr. Empty substr panics — empty match would count every record and almost certainly indicates a caller bug.

func (*LogCapture) WithAttrs 

func (c *LogCapture) WithAttrs([]slog.Attr) slog.Handler

WithAttrs returns the receiver unchanged. Captured records keep only their Message, so per-handler attrs are discarded by design.

func (*LogCapture) WithGroup 

func (c *LogCapture) WithGroup(string) slog.Handler

WithGroup returns the receiver unchanged for the same reason as WithAttrs.

type MockTaskContext

MockTaskContext is a test double for worker.TaskContext that records all method calls. Safe for concurrent use. Satisfies TaskContext and all role interfaces (SimpleTask, CheckpointTask, LoopTask, StreamTask, SignalTask).

type MockTaskContext struct {

    // Configuration -- set before use
    InputData      []byte
    RunIDValue     string
    StepIDValue    string
    RetryCountVal  int
    CtxValue       context.Context
    CheckpointData []byte // returned by LoadCheckpoint
    SignalData     []byte // returned by WaitForSignal
    FailErr        error  // if set, Complete/Fail/Continue return this

    // Recorded calls
    Completed           bool
    CompletedOutput     []byte
    Failed              bool
    FailedErr           error
    FailedPermanent     bool
    Continued           bool
    ContinuedOutput     []byte
    Checkpoints         [][]byte
    Streams             [][]byte
    HeartbeatCount      int
    SignalsSent         []SentSignal
    FailRetryAfterCalls []FailRetryAfterCall
    // contains filtered or unexported fields
}

func (*MockTaskContext) Checkpoint 

func (m *MockTaskContext) Checkpoint(state []byte) error

func (*MockTaskContext) Complete 

func (m *MockTaskContext) Complete(output []byte) error

func (*MockTaskContext) Context 

func (m *MockTaskContext) Context() context.Context

func (*MockTaskContext) Continue 

func (m *MockTaskContext) Continue(output []byte) error

func (*MockTaskContext) ControlPlane 

func (m *MockTaskContext) ControlPlane() worker.ControlPlane

ControlPlane returns nil: the mock is ungated by default, matching the deny-by-default contract. Tests that need a granted handle set it explicitly via their own double.

func (*MockTaskContext) Fail 

func (m *MockTaskContext) Fail(err error) error

func (*MockTaskContext) FailPermanent 

func (m *MockTaskContext) FailPermanent(err error) error

func (*MockTaskContext) FailRetryAfter 

func (m *MockTaskContext) FailRetryAfter(err error, after time.Duration) error

func (*MockTaskContext) Heartbeat 

func (m *MockTaskContext) Heartbeat() error

func (*MockTaskContext) Input 

func (m *MockTaskContext) Input() []byte

func (*MockTaskContext) LoadCheckpoint 

func (m *MockTaskContext) LoadCheckpoint() ([]byte, error)

func (*MockTaskContext) Metadata 

func (m *MockTaskContext) Metadata() map[string]string

func (*MockTaskContext) Pause 

func (m *MockTaskContext) Pause(name string, duration time.Duration) error

func (*MockTaskContext) PutStream 

func (m *MockTaskContext) PutStream(data []byte) error

func (*MockTaskContext) RetryCount 

func (m *MockTaskContext) RetryCount() int

func (*MockTaskContext) RunID 

func (m *MockTaskContext) RunID() string

func (*MockTaskContext) SendSignal 

func (m *MockTaskContext) SendSignal(runID, name string, data []byte) error

func (*MockTaskContext) StepID 

func (m *MockTaskContext) StepID() string

func (*MockTaskContext) WaitForSignal 

func (m *MockTaskContext) WaitForSignal(name string, timeout time.Duration) ([]byte, error)

type RunFixture

RunFixture wraps a Harness with helpers that submit and advance workflow runs into terminal states. The fixture is concern-scoped (run shape) rather than infrastructure-scoped (NATS lifecycle).

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

func NewRunFixture 

func NewRunFixture(h *Harness) *RunFixture

NewRunFixture constructs a RunFixture bound to h. Panics on nil harness — programmer error.

func (*RunFixture) PublishStaleStepStarted 

func (f *RunFixture) PublishStaleStepStarted(ctx context.Context, t *testing.T, runID, stepID string, attempt int)

PublishStaleStepStarted synthesizes a step.started event for runID/stepID at the given attempt number and publishes it to the WORKFLOW_HISTORY stream so the engine’s history consumer receives it. The event carries the same shape a real worker would publish.

We deliberately do NOT set Nats-Msg-Id to the deterministic NATSMsgID(). The production failure mode this regression test guards against is delivery of an event the engine has already terminalised — that delivery can arrive via either consumer-side redelivery (same stream message, replayed past ack) or a fresh publish from a recovered worker that the stream’s dedup window has already aged out. Both produce identical engine input. To exercise the engine path without coupling the test to either JetStream’s consumer redelivery internals or the 5s dedup window on WORKFLOW_HISTORY, we publish with a unique tag in the MsgId so the stream accepts the message and the engine sees the duplicate payload. The engine’s terminal-state guard in handleStepStarted is the contract under test, not the dedup window.

attempt must be > 0; attempt == 0 is reserved for “not yet scheduled” elsewhere in the engine.

func (*RunFixture) RunSingleStepToCompletion 

func (f *RunFixture) RunSingleStepToCompletion(t *testing.T) (string, string)

RunSingleStepToCompletion drives one fresh single-step workflow through the worker harness and waits for it to reach Completed. Returns the run ID and step ID of the now-terminal step. Callers rely on this to seed the “step is terminal” precondition before synthesizing a stale step.started in idempotency tests.

func (*RunFixture) Snapshot 

func (f *RunFixture) Snapshot(t *testing.T, runID string) dag.WorkflowRun

Snapshot returns the current KV snapshot for runID. Fatals the test on load failure — callers use this to assert state didn’t change across an event publish, so a load error is unrecoverable.

func (*RunFixture) SubmitAndAdvanceTo 

func (f *RunFixture) SubmitAndAdvanceTo(t *testing.T, state string, n int)

SubmitAndAdvanceTo registers and runs n single-step workflows whose handlers drive each run to the requested terminal state (“completed” or “failed”). Returns once every run has reached state or the per-run timeout fires.

type SentSignal

SentSignal records a SendSignal call.

type SentSignal struct {
    RunID string
    Name  string
    Data  []byte
}

Generated by gomarkdoc