Testing Your Plugin

Unit Testing with net.Pipe

The SDK is designed for testability. Use net.Pipe() to create a connected pair of connections, then use sdk.NewWithConn() on one end and simulate the engine on the other.

import (
    "net"
    "testing"

    "github.com/ze-software/ze/pkg/plugin/rpc"
    "github.com/ze-software/ze/pkg/plugin/sdk"
)

func newTestPair(t *testing.T) (*sdk.Plugin, *rpc.MuxConn) {
    t.Helper()

    pluginEnd, engineEnd := net.Pipe()
    t.Cleanup(func() {
        pluginEnd.Close()
        engineEnd.Close()
    })

    p := sdk.NewWithConn("test-plugin", pluginEnd)

    engineConn := rpc.NewConn(engineEnd, engineEnd)
    engineMux := rpc.NewMuxConn(engineConn)
    t.Cleanup(func() { engineMux.Close() })

    return p, engineMux
}

The engine side uses rpc.MuxConn for bidirectional RPCs: it reads plugin requests via Requests() and sends engine callbacks via CallRPC.

Testing Event Handlers

Register callbacks with OnEvent, OnConfigure, or OnExecuteCommand, then run the plugin in a goroutine and simulate engine messages from the other end of the pipe.

func TestEventHandler(t *testing.T) {
    p, engineMux := newTestPair(t)

    eventReceived := make(chan string, 1)
    p.OnEvent(func(event string) error {
        eventReceived <- event
        return nil
    })

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    errCh := make(chan error, 1)
    go func() {
        errCh <- p.Run(ctx, sdk.Registration{})
    }()

    // Complete the 5-stage startup from the engine side
    completeStartup(t, ctx, engineMux)

    // Deliver an event
    eventInput := struct {
        Event string `json:"event"`
    }{Event: `{"type":"bgp","bgp":{"peer":{"address":"10.0.0.1"}}}`}

    _, err := engineMux.CallRPC(ctx, "ze-plugin-callback:deliver-event", eventInput)
    require.NoError(t, err)

    select {
    case got := <-eventReceived:
        assert.Contains(t, got, "10.0.0.1")
    case <-time.After(time.Second):
        t.Fatal("event callback not called")
    }

    // Shutdown
    byeInput := struct {
        Reason string `json:"reason"`
    }{Reason: "test-done"}
    _, _ = engineMux.CallRPC(ctx, "ze-plugin-callback:bye", byeInput)

    require.NoError(t, <-errCh)
}

Simulating the 5-Stage Startup

The Run() method performs the 5-stage startup protocol before entering the event loop. Tests must simulate all five stages from the engine side.

func completeStartup(t *testing.T, ctx context.Context, engineMux *rpc.MuxConn) {
    t.Helper()

    // Stage 1: read declare-registration, respond OK
    req := <-engineMux.Requests()
    assert.Equal(t, "ze-plugin-engine:declare-registration", req.Method)
    require.NoError(t, engineMux.SendOK(ctx, req.ID))

    // Stage 2: send configure
    configInput := struct {
        Sections []sdk.ConfigSection `json:"sections"`
    }{}
    _, err := engineMux.CallRPC(ctx, "ze-plugin-callback:configure", configInput)
    require.NoError(t, err)

    // Stage 3: read declare-capabilities, respond OK
    req = <-engineMux.Requests()
    assert.Equal(t, "ze-plugin-engine:declare-capabilities", req.Method)
    require.NoError(t, engineMux.SendOK(ctx, req.ID))

    // Stage 4: send share-registry
    registryInput := struct {
        Commands []sdk.RegistryCommand `json:"commands"`
    }{}
    _, err = engineMux.CallRPC(ctx, "ze-plugin-callback:share-registry", registryInput)
    require.NoError(t, err)

    // Stage 5: read ready, respond OK
    req = <-engineMux.Requests()
    assert.Equal(t, "ze-plugin-engine:ready", req.Method)
    require.NoError(t, engineMux.SendOK(ctx, req.ID))
}

The five stages are:

Testing Command Execution

Plugins register command handlers with OnExecuteCommand. The handler receives a serial, command name, args, and peer address, and returns status, data, and error.

func TestCommandHandler(t *testing.T) {
    p, engineMux := newTestPair(t)

    p.OnExecuteCommand(func(serial, command string, args []string, peer string) (string, any, error) {
        if command == "show-status" {
            return "done", `{"status":"healthy"}`, nil
        }
        return "error", "unknown command", nil
    })

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    errCh := make(chan error, 1)
    go func() {
        errCh <- p.Run(ctx, sdk.Registration{
            Commands: []sdk.CommandDecl{
                {Name: "show-status", Description: "Show health status"},
            },
        })
    }()

    completeStartup(t, ctx, engineMux)

    // Send execute-command
    cmdInput := struct {
        Serial  string   `json:"serial"`
        Command string   `json:"command"`
        Args    []string `json:"args,omitempty"`
        Peer    string   `json:"peer,omitempty"`
    }{Serial: "1", Command: "show-status"}

    out := executeCommand(t, ctx, engineMux, cmdInput)
    assert.Equal(t, "done", out.Status)
    assert.Contains(t, string(out.Data), "healthy")
}

CallRPC cannot read this answer. A plugin answers execute-command with a head, its records and a terminator, so a test that reads one JSON result takes the head line's tail for its payload. Read the answer with CallAnswer and collapse it, which is the engine's own three steps:

func executeCommand(t *testing.T, ctx context.Context, mux *rpc.MuxConn, input any) *rpc.ExecuteCommandOutput {
    t.Helper()

    answer, err := mux.CallAnswer(ctx, "ze-plugin-callback:execute-command", input)
    require.NoError(t, err)

    document, collapseErr := rpc.CollapseAnswer(answer)
    require.NoError(t, answer.Err(), "the answer must reach its terminator")
    require.NoError(t, collapseErr)
    return &rpc.ExecuteCommandOutput{Status: answer.Status, Data: document}
}

answer.Err() is read AFTER the collapse, because the walk is what fills it.

Testing Configuration Handling

The OnConfigure callback receives config sections during Stage 2. The OnConfigVerify and OnConfigApply callbacks handle config reload.

func TestConfigHandler(t *testing.T) {
    p, engineMux := newTestPair(t)

    var receivedConfig []sdk.ConfigSection
    p.OnConfigure(func(sections []sdk.ConfigSection) error {
        receivedConfig = sections
        return nil
    })

    ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
    defer cancel()

    errCh := make(chan error, 1)
    go func() {
        errCh <- p.Run(ctx, sdk.Registration{
            WantsConfig: []string{"bgp"},
        })
    }()

    // Stage 1
    req := <-engineMux.Requests()
    require.NoError(t, engineMux.SendOK(ctx, req.ID))

    // Stage 2: send config
    configInput := struct {
        Sections []sdk.ConfigSection `json:"sections"`
    }{
        Sections: []sdk.ConfigSection{
            {Root: "bgp", Data: `{"router-id":"1.2.3.4"}`},
        },
    }
    _, err := engineMux.CallRPC(ctx, "ze-plugin-callback:configure", configInput)
    require.NoError(t, err)

    assert.Equal(t, 1, len(receivedConfig))
    assert.Equal(t, "bgp", receivedConfig[0].Root)
}

Table-Driven Tests

Test multiple scenarios efficiently using Go table-driven test patterns:

func TestCommandDispatch(t *testing.T) {
    tests := []struct {
        name       string
        command    string
        wantStatus string
        wantData   string
    }{
        {"known command", "show-routes", "done", `{"count":42}`},
        {"unknown command", "invalid", "error", "unknown command"},
        {"empty args", "show-routes", "done", `{"count":42}`},
    }

    for _, tt := range tests {
        t.Run(tt.name, func(t *testing.T) {
            p, engineMux := newTestPair(t)

            p.OnExecuteCommand(func(serial, cmd string, args []string, peer string) (string, any, error) {
                if cmd == "show-routes" {
                    return "done", `{"count":42}`, nil
                }
                return "error", "unknown command", nil
            })

            ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
            defer cancel()

            go func() { _ = p.Run(ctx, sdk.Registration{}) }()
            completeStartup(t, ctx, engineMux)

            cmdInput := struct {
                Serial  string `json:"serial"`
                Command string `json:"command"`
            }{Serial: "1", Command: tt.command}

            out := executeCommand(t, ctx, engineMux, cmdInput)
            assert.Equal(t, tt.wantStatus, out.Status)
            assert.JSONEq(t, tt.wantData, string(out.Data))
        })
    }
}

CI Integration

# .github/workflows/test.yml (or Forgejo/Woodpecker equivalent)
jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3

      - name: Set up Go
        uses: actions/setup-go@v4
        with:
          go-version: '1.21'

      - name: Test
        run: go test -race -v ./...

      - name: Build plugin
        run: go build -o my-plugin

Coverage

Run with coverage:

go test -race -cover -coverprofile=coverage.out ./...
go tool cover -html=coverage.out

Target at least 80% coverage on handlers.

Debugging

Enable debug logging with ze's hierarchical log system:

# Set log level for your plugin
export ZE_LOG_LEVEL=debug
./my-plugin

Or add structured logging in your plugin:

import "log/slog"

p.OnEvent(func(event string) error {
    slog.Debug("event received", "event", event)
    // ...
    return nil
})