mesh-catalog/modules/mesh-build-image/cmd/mesh-build-image/bus_integration_test.go

//go:build integration

package main

// ADR 0487 §1: the build step answers the image's digest, through a real bus (issue 1703e). A nats-server from the Go
// library, at the version this module's go.mod names, runs with JetStream on in the test's directory (the pattern of
// mesh-build-go's bus.go). The module's steps state is a key-value bucket on that bus, and its tools are served on bus
// subjects by a small bridge in the test, so a call, the step that runs behind it and the key it ends on all travel over
// the bus; the mesh's own tool runner is not started here. buildah and git are the fake runner's, so this runs wherever
// the check runs; the real buildah is TestARealBuild... in integration_test.go. What the keys hold is compared with the
// captured shapes (testdata/captured, capture_test.go) that the builder's tests copy.

import (
	"context"
	"encoding/json"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/delivery"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"
)

// bus starts the throwaway server and answers a connection to it.
func bus(t *testing.T) *nats.Conn {
	t.Helper()
	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		t.Fatal("the bus was not ready")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	nc, err := nats.Connect(s.ClientURL())
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(nc.Close)
	return nc
}

// onBus wires a module's steps state to a bucket and serves its tools on "test.tool.<name>": the request body is the
// arguments, the answer is {"result": ...} or {"error": "..."}.
func onBus(t *testing.T, nc *nats.Conn, m *Module) jetstream.KeyValue {
	t.Helper()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	ctx := context.Background()
	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-build-image-steps"})
	if err != nil {
		t.Fatal(err)
	}
	m.Steps = func(key string, value any) (uint64, error) {
		raw, err := json.Marshal(value)
		if err != nil {
			return 0, err
		}
		return kv.Put(ctx, key, raw)
	}
	for _, tl := range m.Tools(m.StepTools()) {
		tl := tl
		if _, err := nc.Subscribe("test.tool."+tl.Name, func(msg *nats.Msg) {
			var args map[string]any
			_ = json.Unmarshal(msg.Data, &args)
			res, err := tl.Run(args)
			out := map[string]any{"result": res}
			if err != nil {
				out = map[string]any{"error": err.Error()}
			}
			b, _ := json.Marshal(out)
			_ = msg.Respond(b)
		}); err != nil {
			t.Fatal(err)
		}
	}
	return kv
}

// call calls a tool over the bus and answers its result, which for a long step is the running step's key.
func call(t *testing.T, nc *nats.Conn, tool string, args map[string]any) map[string]any {
	t.Helper()
	body, _ := json.Marshal(args)
	msg, err := nc.Request("test.tool."+tool, body, 10*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	var out struct {
		Result map[string]any `json:"result"`
		Error  string         `json:"error"`
	}
	if err := json.Unmarshal(msg.Data, &out); err != nil || out.Error != "" {
		t.Fatalf("%s answered %s (%v)", tool, msg.Data, err)
	}
	return out.Result
}

// ended waits for the step's key in the bucket to leave running and answers the step as the bucket holds it.
func ended(t *testing.T, kv jetstream.KeyValue, key string) delivery.Step {
	t.Helper()
	deadline := time.Now().Add(30 * time.Second)
	for time.Now().Before(deadline) {
		if e, err := kv.Get(context.Background(), key); err == nil {
			st, err := delivery.DecodeStep(e.Value())
			if err != nil {
				t.Fatal(err)
			}
			if st.State != delivery.StepRunning {
				return st
			}
		}
		time.Sleep(20 * time.Millisecond)
	}
	t.Fatalf("the key %s never left running", key)
	return delivery.Step{}
}

func TestTheStepsAnswerThroughARealBusWithTheCapturedShapes(t *testing.T) {
	nc := bus(t)
	f := newFake(t)
	m := moduleWith(t, f, newStates())
	kv := onBus(t, nc, m)
	place(t, m, same)

	// The fetch: a running key at once, then the workspace on the key.
	started := call(t, nc, FetchTool, map[string]any{"repository": "novox/mesh-catalog", "commit": commitSHA})
	if started["state"] != delivery.StepRunning || started["key"] == "" {
		t.Fatalf("the fetch answered %v, want a running step and its key", started)
	}
	st := ended(t, kv, started["key"].(string))
	compareCaptured(t, "fetch.json", normalized(t, st, workspaceOf(t, st), commitSHA, ""))

	// The build of a pull request: the image's digest, nothing pushed. The workspace is the fixture's, "ws1".
	for name, extra := range map[string]map[string]any{"build-candidate.json": nil, "build-trunk.json": {"publish": true}} {
		args := map[string]any{"workspace": "ws1", "module": "modules/demo", "artifact": "server"}
		for k, v := range extra {
			args[k] = v
		}
		started := call(t, nc, BuildTool, args)
		st := ended(t, kv, started["key"].(string))
		if st.State != delivery.StepPassed {
			t.Fatalf("%s: step = %+v", name, st)
		}
		var res BuildResult
		if err := json.Unmarshal(st.Result, &res); err != nil || !digestOK.MatchString(res.Digest) || res.Pushed != (extra != nil) {
			t.Errorf("%s: result = %s (%v), want the image's digest and pushed %v", name, st.Result, err, extra != nil)
		}
		compareCaptured(t, name, normalized(t, st, "", "", f.address()))
	}
	if len(f.registry) != 1 || f.pushes != 1 {
		t.Errorf("the registry holds %v after %d pushes, want only the trunk build's one", f.registry, f.pushes)
	}
}