diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go new file mode 100644 index 0000000..739b315 --- /dev/null +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -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) + } + } +}