Start the assigner whole on an embedded bus in an integration test: it says its three states and refuses to start without a bucket it reads (issue 1297, subtask B2b)

2026-10-23 15:19:49 +02:00
parent 034a6de9a7
commit c2b7b8867e
1 changed files with 144 additions and 0 deletions
@@ -0,0 +1,144 @@
+//go:build integration
+
+package main
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/nats-io/nats-server/v2/server"
+	"github.com/nats-io/nats.go"
+	"github.com/nats-io/nats.go/jetstream"
+
+	"git.novox.be/novox/mesh-sdk/go/announce"
+	"git.novox.be/novox/mesh-sdk/go/process"
+)
+
+// bed is an embedded bus with JetStream, every bucket the assigner reads and keeps, and the stream the mesh's events
+// are kept in, as the controller's broker raises them.
+type bed struct {
+	srv     *server.Server
+	js      jetstream.JetStream
+	ctx     context.Context
+	dir     string
+	buckets map[string]jetstream.KeyValue
+}
+
+func startBed(t *testing.T, leaveOut string) *bed {
+	t.Helper()
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "assigner")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: filepath.Join(dir, "js"), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	t.Cleanup(cancel)
+	b := &bed{srv: srv, js: js, ctx: ctx, dir: dir, buckets: map[string]jetstream.KeyValue{}}
+	for _, name := range append(append([]string{}, readBuckets...), ownBuckets...) {
+		if name == leaveOut {
+			continue
+		}
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
+		if err != nil {
+			t.Fatal(err)
+		}
+		b.buckets[name] = kv
+	}
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	return b
+}
+
+// start runs the program as the node does, through run with its environment and a broker file; it answers the
+// channel run's error arrives on.
+func (b *bed) start(t *testing.T) <-chan error {
+	t.Helper()
+	broker := filepath.Join(b.dir, "broker")
+	cred, _ := json.Marshal(process.Account{URL: b.srv.ClientURL(), Node: "n1", Module: "mesh-delivery-assigner", User: "n1.mesh-delivery-assigner", Password: "pw"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker}
+	ctx, stop := context.WithCancel(b.ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(ctx, func(k string) string { return env[k] }, io.Discard) }()
+	t.Cleanup(func() {
+		stop()
+		select {
+		case <-done:
+		case <-time.After(10 * time.Second):
+		}
+	})
+	return done
+}
+
+// 1297/B2b: started whole on its credential, the assigner says on the bus the three states it keeps, each readable,
+// having opened every bucket it names.
+func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) {
+	b := startBed(t, "")
+	done := b.start(t)
+	events, err := b.js.Stream(b.ctx, "EVENTS")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var said announce.States
+	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
+		select {
+		case err := <-done:
+			t.Fatalf("the assigner stopped: %v", err)
+		default:
+		}
+		m, err := events.GetLastMsgForSubject(b.ctx, announce.Subject("mesh-delivery-assigner"))
+		if err != nil {
+			continue
+		}
+		if err := json.Unmarshal(m.Data, &said); err != nil {
+			t.Fatal(err)
+		}
+		break
+	}
+	if said.Module != "mesh-delivery-assigner" || said.Node != "n1" || len(said.States) != 3 ||
+		!said.Readable("wants") || !said.Readable("differences") || !said.Readable("compositions") {
+		t.Fatalf("the assigner said %+v, want its three states, each readable", said)
+	}
+}
+
+// 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket.
+func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
+	for _, missing := range readBuckets {
+		b := startBed(t, missing)
+		select {
+		case err := <-b.start(t):
+			if err == nil || !strings.Contains(err.Error(), missing) {
+				t.Errorf("without %s the assigner stopped with %v, want the bucket named", missing, err)
+			}
+		case <-time.After(15 * time.Second):
+			t.Errorf("without %s the assigner started", missing)
+		}
+	}
+}