From f8212dc20321d18b15fdae1fc8caf55f001b5fc2 Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 13:01:15 +0100 Subject: [PATCH] CAT-D7: distribution's whole-module integration test against a real bus The program had no test that runs it whole. This one starts run() on a nats-server of its own, with the registry container faked on the loopback and node_haves and manifest-reads answered on their call subjects: an away node holds both windows open, its return closes them, each closing said once on distribution.events, a direct write is then refused, reads stay open, the health check answers through the program's port, and a waiter on published wakes on none of the windows' events. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig --- .../distribution/whole_integration_test.go | 220 ++++++++++++++++++ modules/distribution/go.mod | 6 + modules/distribution/go.sum | 1 + 3 files changed, 227 insertions(+) create mode 100644 modules/distribution/cmd/distribution/whole_integration_test.go diff --git a/modules/distribution/cmd/distribution/whole_integration_test.go b/modules/distribution/cmd/distribution/whole_integration_test.go new file mode 100644 index 00000000..3bfc3be8 --- /dev/null +++ b/modules/distribution/cmd/distribution/whole_integration_test.go @@ -0,0 +1,220 @@ +//go:build integration + +package main + +// distribution's program whole against a real bus (CAT-D7; design 51 *Acceptance tests*, the holder's cases): its +// run() as the node starts it, its broker account on a nats-server of this test alone, the registry container a fake +// OCI store on the loopback, the status view's node_haves and the node-engine seat's manifest-reads answered on their +// call subjects. An away node holds the old name open; its return (manifest-reads-changed) closes it, said once on +// distribution.events, after which a direct write is refused naming publish; reads stay open; and a waiter on a seat's +// `published` hears none of the windows' events. + +import ( + "context" + "encoding/json" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + neturl "net/url" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/nats-io/nats.go" + "github.com/nats-io/nats.go/jetstream" + + "git.novox.be/novox/mesh-sdk/go/testbus" + + "distribution/internal/holder" +) + +func TestDistributionWhole(t *testing.T) { + url := testbus.URL(t) + nc, err := nats.Connect(url) + if err != nil { + t.Fatal(err) + } + defer nc.Close() + js, _ := jetstream.New(nc) + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + for _, b := range []string{"distribution_windows", "distribution_findings"} { + if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: b}); err != nil { + t.Fatal(err) + } + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "DISTRIBUTION_EVENTS", Subjects: []string{"mesh.stream.distribution.event.>"}, Duplicates: time.Hour}); err != nil { + t.Fatal(err) + } + if _, err := js.CreateOrUpdateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{Durable: "novox_distribution", AckPolicy: jetstream.AckExplicitPolicy, + FilterSubjects: []string{"mesh.mod.mesh-delivery-view.event.delivered", "mesh.seat.node-engine.event.manifest-reads-changed.*"}}); err != nil { + t.Fatal(err) + } + + // The status view and the node-engines: g14 away (its node-engine answers nothing) and running the old build. + var mu sync.Mutex + g14 := false + answer := func(m *nats.Msg, result any) { + b, _ := json.Marshal(map[string]any{"result": result, "node": "novox"}) + _ = m.Respond(b) + } + if _, err := nc.Subscribe("mesh.mod.mesh-delivery-view.>", func(m *nats.Msg) { + mu.Lock() + defer mu.Unlock() + fp := "fp-old" + if g14 { + fp = "fp-new" + } + answer(m, map[string]any{"keys": map[string]any{ + "ace.mesh-build-go": map[string]any{"fp": "fp-new"}, "g14.mesh-build-go": map[string]any{"fp": fp}, + "ace.gitea": map[string]any{"fp": "x"}, "g14.gitea": map[string]any{"fp": "x"}, + }}) + }); err != nil { + t.Fatal(err) + } + engine := func(node string) { + for _, subj := range []string{"mesh.seat.node-engine.call.manifest-reads." + node + ".>", "mesh.seat.node-engine.tool.manifest-reads." + node} { + if _, err := nc.Subscribe(subj, func(m *nats.Msg) { + answer(m, map[string]any{"v": 1, "manifest-reads": []string{holder.NewNamesCapability}}) + }); err != nil { + t.Fatal(err) + } + } + _ = nc.Flush() + } + engine("ace") // g14 is away: nothing answers its node-engine, so it states nothing + // A waiter on a seat's published: it must hear none of the windows' events. + var woke []string + if _, err := nc.Subscribe("mesh.seat.container-image-registry.event.published", func(m *nats.Msg) { woke = append(woke, m.Subject) }); err != nil { + t.Fatal(err) + } + pub, _ := json.Marshal(holder.Publishing{Name: holder.PublishingKey, Modules: map[string][]string{"mesh-build-go": {"fp-new"}}}) + windows, _ := js.KeyValue(ctx, "distribution_windows") + if _, err := windows.Put(ctx, holder.PublishingKey, pub); err != nil { + t.Fatal(err) + } + + // The registry container on the loopback, and the program's port. + store := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if strings.HasSuffix(r.URL.Path, "/tags/list") { + _, _ = io.WriteString(w, `{"tags":[]}`) + return + } + if r.Method == http.MethodHead { + w.WriteHeader(http.StatusNotFound) // no tag published through the seat + return + } + w.WriteHeader(http.StatusCreated) + })) + defer store.Close() + l, _ := net.Listen("tcp", "127.0.0.1:0") + listen := l.Addr().String() + l.Close() + dir := t.TempDir() + acct, _ := json.Marshal(map[string]any{"url": url, "node": "novox", "module": "distribution", "user": "novox.distribution", "password": "unused"}) + _ = os.WriteFile(filepath.Join(dir, "broker"), acct, 0o600) + env := map[string]string{"MESH_BROKER_FILE": filepath.Join(dir, "broker"), "MESH_REGISTRY_LOOPBACK": store.URL, "MESH_REGISTRY_LISTEN": listen} + runCtx, stop := context.WithCancel(ctx) + done := make(chan error, 1) + go func() { done <- run(runCtx, func(k string) string { return env[k] }) }() + defer func() { stop(); <-done }() + + window := func(name string) holder.Window { + var w holder.Window + if e, err := windows.Get(ctx, name); err == nil { + _ = json.Unmarshal(e.Value(), &w) + } + return w + } + eventually := func(what string, ok func() bool) { + t.Helper() + for end := time.Now().Add(20 * time.Second); time.Now().Before(end); time.Sleep(50 * time.Millisecond) { + if ok() { + return + } + select { + case err := <-done: + t.Fatalf("the program ended: %v", err) + default: + } + } + t.Fatalf("not within 20s: %s (old-name %+v, direct-writes %+v)", what, window(holder.OldNameWindow), window(holder.DirectWritesWindow)) + } + write := func() int { + req, _ := http.NewRequest(http.MethodPut, "http://"+listen+"/v2/m/i/manifests/0.1.9", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + return 0 + } + resp.Body.Close() + return resp.StatusCode + } + eventually("the start's wake: g14 holds both windows", func() bool { + o, d := window(holder.OldNameWindow), window(holder.DirectWritesWindow) + return o.Open && fmt.Sprint(o.Holding) == "[g14]" && d.Open && fmt.Sprint(d.Holding) == "[g14]" + }) + if got := write(); got != http.StatusCreated { + t.Errorf("a direct write while the window is open = %d, want it through", got) + } + + // g14 returns with the new node-engine and the new build: its manifest-reads-changed closes both windows. + mu.Lock() + g14 = true + mu.Unlock() + engine("g14") + if _, err := js.Publish(ctx, "mesh.seat.node-engine.event.manifest-reads-changed.g14", []byte(`{"v":1,"node":"g14","why":"returned"}`)); err != nil { + t.Fatal(err) + } + eventually("g14's return closes both windows", func() bool { + return !window(holder.OldNameWindow).Closed.IsZero() && !window(holder.DirectWritesWindow).Closed.IsZero() + }) + // A delivered after the close says nothing again. + if _, err := js.Publish(ctx, "mesh.mod.mesh-delivery-view.event.delivered", []byte(`{}`)); err != nil { + t.Fatal(err) + } + eventually("a direct write refused naming publish once closed", func() bool { return write() == http.StatusForbidden }) + resp, err := http.Get("http://" + listen + "/v2/m/i/tags/list") + if err != nil || resp.StatusCode != http.StatusOK { + t.Errorf("a read once the window closed: %v %v", resp, err) + } + if _, err := holder.Health(&holder.OCI{Base: mustURL("http://" + listen)}, nil); err != nil { + t.Errorf("the health check through the program's port: %v", err) + } + s, _ := js.Stream(ctx, "DISTRIBUTION_EVENTS") + eventually("the program heard every waking event, the delivered after the close included", func() bool { + c, err := js.Consumer(ctx, "EVENTS", "novox_distribution") + if err != nil { + return false + } + i, err := c.Info(ctx) + return err == nil && i.NumPending == 0 && i.NumAckPending == 0 && i.Delivered.Consumer >= 2 + }) + info, _ := s.Info(ctx) + if info.State.Msgs != 2 { + t.Errorf("distribution.events holds %d events, want old-name-closed and direct-writes-closed once each", info.State.Msgs) + } + for _, ev := range []string{holder.OldNameClosedEvent, holder.DirectWritesClosedEvent} { + if _, err := s.GetLastMsgForSubject(ctx, "mesh.stream.distribution.event."+ev); err != nil { + t.Errorf("%s not said: %v", ev, err) + } + } + if len(woke) != 0 { + t.Errorf("a waiter on published woke on %v", woke) + } +} + +func mustURL(s string) *neturl.URL { + u, err := neturl.Parse(s) + if err != nil { + panic(err) + } + return u +} diff --git a/modules/distribution/go.mod b/modules/distribution/go.mod index b7e3cf30..e8d93c52 100644 --- a/modules/distribution/go.mod +++ b/modules/distribution/go.mod @@ -8,9 +8,15 @@ require ( ) require ( + github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op // indirect + github.com/google/go-tpm v0.9.8 // indirect github.com/klauspost/compress v1.20.0 // indirect + github.com/minio/highwayhash v1.0.4 // indirect + github.com/nats-io/jwt/v2 v2.8.1 // indirect + github.com/nats-io/nats-server/v2 v2.11.17 // indirect github.com/nats-io/nkeys v0.4.16 // indirect github.com/nats-io/nuid v1.0.1 // indirect golang.org/x/crypto v0.57.0 // indirect golang.org/x/sys v0.48.0 // indirect + golang.org/x/time v0.15.0 // indirect ) diff --git a/modules/distribution/go.sum b/modules/distribution/go.sum index d72bd4ff..7679a5dc 100644 --- a/modules/distribution/go.sum +++ b/modules/distribution/go.sum @@ -20,6 +20,7 @@ github.com/nats-io/nuid v1.0.1 h1:5iA8DT8V7q8WK2EScv2padNa/rTESc1KdnPw4TC2paw= github.com/nats-io/nuid v1.0.1/go.mod h1:19wcPz3Ph3q0Jbyiqsd0kePYG7A95tJPxeL+1OSON2c= golang.org/x/crypto v0.57.0 h1:3ZVCjf8Ggz7zneR/EHRVx68Ctf+2pmIMP2UFhh9cC6M= golang.org/x/crypto v0.57.0/go.mod h1:Fdz0i5U6CoizGwLda9DttjSk6qlZo25zYNtR+ycvuZA= +golang.org/x/sys v0.21.0/go.mod h1:/VUhepiaJMQUp4+oa/7Zr1D23ma6VTLIYjOOTFZPUcA= golang.org/x/sys v0.48.0 h1:bbX/i/6MgT9BVLM9RT1thmxL04yeTAhbEz4SyadbXoo= golang.org/x/sys v0.48.0/go.mod h1:hNLxWAXmnKAxqDtdwIYC4bM9oQPEecfsnNMuSxOs3og= golang.org/x/time v0.15.0 h1:bbrp8t3bGUeFOx08pvsMYRTCVSMk89u4tKbNOZbp88U=