mesh-catalog/modules/distribution/cmd/distribution/whole_integration_test.go

//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; a waiter on a seat's
// `published` hears none of the windows' events; and the build modules counted are publish's callers as the tools
// bundle answers the config value, not design 51's defaults.

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"},
			// A build module the operator took off publish's list: it holds no window (the caller list as configured).
			"g14.mesh-build-node": map[string]any{"fp": "fp-node-old"},
		}})
	}); err != nil {
		t.Fatal(err)
	}
	// distribution's own tools bundle answers the config value callers as mesh-config resolved it: publish by
	// mesh-build-go alone. The program reads it there, never from an environment variable nothing renders.
	if _, err := nc.Subscribe("mesh.mod.distribution.>", func(m *nats.Msg) {
		if !strings.Contains(m.Subject, holder.CallersTool) {
			return
		}
		answer(m, map[string]any{"callers": map[string]any{"publish": []string{"mesh-build-go"}}})
	}); 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
}