Local tests

github.com/graphene-ci/pipeline/pkg/pipelinetest attaches a Graphene contract simulator to go.temporal.io/sdk/testsuite.TestWorkflowEnvironment. Run the real pipeline function with typed inputs, fixtures for external state, and assertions on both its result and its resources. No Graphene server, Docker daemon, Kubernetes cluster, cloud account or host shell is needed.

Install

Add the SDK and the adapters your pipeline uses to its Go module:

go get github.com/graphene-ci/pipeline@v0.2.0
go get github.com/graphene-ci/library/docker@v0.2.0
go get github.com/graphene-ci/library/k8s@v0.2.0

The adapters are packages inside their library modules; they have no separate module versions. Tests work with published dependencies without a local workspace.

A complete test

package report

import (
    "testing"
    "time"

    "github.com/graphene-ci/pipeline/pkg/artifact"
    "github.com/graphene-ci/pipeline/pkg/pipeline"
    "github.com/graphene-ci/pipeline/pkg/pipelinetest"
    "github.com/stretchr/testify/require"
    "go.temporal.io/sdk/testsuite"
)

func run(ctx pipeline.Context, text string) (pipeline.ArtifactState, error) {
    report := pipeline.NewArtifact(ctx, "report", artifact.FromBytes([]byte(text)))
    state := report.Ready(ctx)
    pipeline.ToStand(ctx, report, pipeline.KeepFor(time.Hour))
    return state, nil
}

func TestReport(t *testing.T) {
    var suite testsuite.WorkflowTestSuite
    env := suite.NewTestWorkflowEnvironment()
    world := pipelinetest.Install(t, env)
    wf := pipelinetest.Workflow(world, "report", run)

    env.ExecuteWorkflow(wf, "benchmark passed")
    require.NoError(t, env.GetWorkflowError())
    var result pipeline.ArtifactState
    require.NoError(t, env.GetWorkflowResult(&result))
    content, found := world.Blob(result.Blob)
    require.True(t, found)
    require.Equal(t, "benchmark passed", string(content))
    world.AssertOwner(t, "artifact/report", "stand/report")
    world.AssertNoLeaks(t)

    world.Advance(time.Hour)
    record, found := world.Resource("artifact/report")
    require.True(t, found)
    require.Equal(t, "deleted", record.Phase)
}

The wrapper and cleanup interceptor are shared with pipeline.Main. Parameter validation, Ready failures, cancellation and cleanup therefore use the same pipeline code. Create a new environment and world for every test; independent worlds support t.Parallel().

Activities and agents

Configure fixtures, install library adapters, then call pipelinetest.Workflow. Preparation discovers activity signatures through the normal recording pass. After preparation, set mocks and execute the workflow.

world.ConnectAfter("bare-1", time.Second)
world.File("bare-1", "/var/log/perf/report.tgz", []byte("fixture content"))
world.SeedArtifact("baseline-report", []byte("baseline"))

wf := pipelinetest.Workflow(world, "perf-nightly", run)
world.OnAgentActivity("bare-1", "run-work", mock.Anything, params.Work).
    Return("benchmark passed", nil).
    Once()

env.ExecuteWorkflow(wf, params)
env.AssertExpectations(t)

OnAgentActivity uses Temporal's mock argument convention, including the activity context as the first argument. It distinguishes the same activity on different agents. Ordinary env.OnActivity also works for user activities. An activity without a replacement fails with UnmockedActivity; its discovered production body is never called. Explicitly registering a replacement on env is an opt-in to executing that replacement.

ConnectAfter only connects an identity. The pipeline must still declare its resource. SeedAgent instead supplies an existing foreign record for attach and selection. Disconnect, PublishCapability and FailResource can change fixtures from env.RegisterDelayedCallback callbacks. Labels and capability needs are evaluated against the world's current state; creating agents are not selected.

User activity retries remain Temporal retries. AtMostOnce keeps one attempt and classifies a timeout as pipeline.ErrUnknown, which the pipeline can handle. env.CancelWorkflow, signals and delayed callbacks keep their normal testsuite behavior. Use world.SetDataConverter when configuring a custom payload converter so both the environment and simulated results use it.

Docker and Kubernetes adapters

Install adapters before preparing the pipeline:

dockertest.Install(world, "28.5.2")
objects := k8stest.Install(world)

Imports are github.com/graphene-ci/library/docker/dockertest and github.com/graphene-ci/library/k8s/k8stest.

A Kubernetes object gets its observed fields from objects.Set(ref, live) or, timed, objects.After(delay, ref, live). A test that does not know which objects a spec will declare answers them all at once with objects.SetDefault(func(kind, name string, manifest map[string]any) any): the function sees the declared manifest and returns the observed fields (nil — nothing observed yet); a Set for a specific object still wins.

The Docker adapter models containers, volumes and networks, preserves their native specs and flows, and returns deterministic fixture identifiers. Its docker.install handler publishes a Docker capability. Override an installation for one agent with OnAgentActivity to test a failure; a mocked failure does not publish the capability.

The Kubernetes adapter accepts native Go objects or maps. The key is the exact Graphene reference (k8s.<group>.<version>.<Kind>/<id>; namespaced ids are <namespace>.<name>). Supply observed fields with objects.Set(ref, live) or objects.After(delay, ref, live) and check the returned error. Omitted fields are taken from the declared manifest. The adapter executes the kind's real WithReady predicate with its polling interval and timeout. A pending or missing fixture does not become ready automatically. Crossplane reference resolution and provider-generated ids must be represented by fixtures.

Assertions and lifetime

Surface	What it checks or returns
Resource(ref)	Detached record including owner, phase, spec, state, agent, flows and From — the run that handed it to its current owner
Resources()	Every record the world knows, deleted ones included, ordered by ref
OnDeclare(fn)	A hook called with each record as it is declared — connect the agent, set the object's observed state, count what a spec produced — without knowing the names in advance
Calls()	Every dispatched activity with its outcome: name, queue, serialized arguments, virtual times, result or error, and a causal sequence number
Events()	Declaration, readiness, transfer and deletion order
Outcome("run/test-<pipeline>")	Cleanup outcome: success, failure or canceled
AssertOwner(t, ref, owner)	The resource is live under the expected owner
AssertNoLeaks(t)	No orphan, cycle or survivor under a completed run
Advance(duration)	Advance modeled stand TTL after the root workflow completes; during the run the TTL runs on its own in virtual time

ToStand moves the root of the declared dependency tree, matching the server's stand transfer rule. A point transfer through Children only reparents that child. Cleanup deletes resources still owned by the run. Explicitly transferred survivors remain even if later pipeline code fails. Foreign fixtures are never adopted or deleted by cleanup. Deleted records remain inspectable. Blob content is removed when no live artifact in the world references it.

A stand's TTL runs in virtual time: a holding given away with KeepFor expires while the run still goes, the way it does on a real stand, so a run that sleeps past its own KeepFor sees the resource deleted. Advance is cumulative and only available after workflow completion; it covers what the run did not wait out and advances the model, not a real stand workflow.

A run that fails keeps what it collected: the workflow error is a Temporal ApplicationError of type pipeline.FailureType whose details carry the partial result and whose cause is the pipeline's own error — errors.Is and errors.As see through it, app.Details(&partial) reads the result.

Reading activity outcomes from Calls

The simulator answers most activities itself — Handle, the library adapters, server.* — and it stands last in the interceptor chain, so no other interceptor or activity listener sees those activities end. Calls is where the outcome is read, for the simulator's own handlers and for mocks alike:

Field	Meaning
Name, TaskQueue, Args, Time	the dispatch: what, where, with which serialized arguments, at which virtual time
Seq	order of the dispatch
Done	virtual time the future settled; zero while the call is pending — a run that ended first leaves it so
DoneSeq	order of the completion, in the same counter as Seq
Result	the JSON payload on success; empty for an activity that returns only an error
Err	the failure message, "" on success
Canceled	the future ended by cancellation (Err is set too)

Virtual time stands still while a workflow task runs, so many events carry one Time. Seq and DoneSeq share a counter precisely for that: a.DoneSeq < b.Seq means a ended before b was sent, b.Seq < a.DoneSeq means they were in flight together.

for _, call := range world.Calls() {
	if call.Name == "docker.job" && call.Err != "" {
		t.Logf("%s failed after %s: %s", call.Name, call.Done.Sub(call.Time), call.Err)
	}
}

A dispatch is one Call whatever its retries: attempts happen under the future, and the recorded outcome is the final one. Use Temporal mock expectations to assert individual attempts.

Full example and development checkouts

examples/full/main_test.go exercises the unchanged run function with two agents, typed Crossplane status, Docker, artifacts, flows and stand transfers. It covers success, work failure, cancellation while waiting for the VM, VM creation failure, failed Docker installation, and a missing foreign artifact. The last case deliberately retains resources already handed to the stand before the failure.

To test the released example with its pinned dependencies:

git clone --branch full/v0.1.0 https://github.com/graphene-ci/examples.git
cd examples/full
GOWORK=off go test -race ./...
GOWORK=off go run . plan

For development across unpublished SDK and library changes, clone pipeline, library and examples as sibling directories, then run:

cd examples
make configure
make test
make lint
cd full
go run . plan
go test -race ./...

configure creates an ignored local go.work pointing to those checkouts and installs pinned tools into the repository's bin/. This setup leaves the versions in go.mod unchanged. The library repository has equivalent configure, test and lint targets covering all its modules.

Verification boundary

The simulator runs pipeline code and models service contracts. It does not run the Graphene control plane, agent processes, Kubernetes controllers, Docker containers, stand workflows or a Temporal service. Simulated service activities use workflow coroutines and virtual timers; their transport retries and heartbeats are not reproduced. User activities use normal Temporal testsuite execution and mocks.

Network flows are declarations, not connectivity checks. Telemetry delivery, secret resolution, cron/webhook admission, cross-pipeline managed runs, ContinueAsNew, visibility consistency and worker crash recovery need separate tests against the relevant components. There is no automatic fallback to live infrastructure for unsupported activities. Adapter authors can use Handle, Handle1, Declare, Ready and OnPrepare to add explicit contract simulations.

The library tests also execute the actual Docker entity init/finalize (including interrupted creation) and Kubernetes apply/heal/finalize workflows with fake external operations. These check lifecycle implementation separately from the pipeline model; they are not a full installation test.