mesh-catalog/modules/pacman/cmd/pacman-tools/keeper_integration_test.go

//go:build integration

package main

// The keeper whole over a real bus (novox/hq ADR 0507 §2, §3, §6; acceptance 1840/AC10; the shape of brother's
// device_integration_test.go): a JetStream server of the test's own, the module's state buckets and its event
// stream, and a fake machine. At its start and after a released lock the node's key and package.changed are on
// the bus, and pacman_packages on another instance reads them from there. Compiled by the merge gate's
// `go vet -tags integration`; run by the merge check, never by hand.

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

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

func newTestBus(t *testing.T) string {
	t.Helper()
	opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
		NoLog: true, NoSigs: true}
	s, err := server.NewServer(opts)
	if err != nil {
		t.Fatalf("a bus for this test could not be made: %v", err)
	}
	go s.Start()
	if !s.ReadyForConnections(30 * time.Second) {
		s.Shutdown()
		t.Fatal("a bus for this test did not come up within 30s")
	}
	t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
	return s.ClientURL()
}

// kvReader reads a bucket as a tools bundle's state does (stdio.State): what another instance's verbs see.
type kvReader struct{ kv jetstream.KeyValue }

func (r kvReader) Get(key string) (*stdio.StateEntry, error) {
	e, err := r.kv.Get(context.Background(), key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return nil, nil
	}
	if err != nil {
		return nil, err
	}
	return &stdio.StateEntry{Key: key, Value: json.RawMessage(e.Value()), Revision: e.Revision()}, nil
}

func (r kvReader) Keys() ([]string, error) {
	keys, err := r.kv.Keys(context.Background())
	if errors.Is(err, jetstream.ErrNoKeysFound) {
		return []string{}, nil
	}
	return keys, err
}

func (r kvReader) History(key string) ([]stdio.StateRevision, error) {
	h, err := r.kv.History(context.Background(), key)
	if errors.Is(err, jetstream.ErrKeyNotFound) {
		return []stdio.StateRevision{}, nil
	}
	if err != nil {
		return nil, err
	}
	out := []stdio.StateRevision{}
	for _, e := range h {
		op := "put"
		if e.Operation() != jetstream.KeyValuePut {
			op = "delete"
		}
		out = append(out, stdio.StateRevision{Key: key, Value: e.Value(), Revision: e.Revision(), Op: op, At: e.Created()})
	}
	return out, nil
}

func TestTheKeeperPutsTheNodesPackagesOnARealBusAndAnotherInstanceReadsThem(t *testing.T) {
	url := newTestBus(t)
	ctx := context.Background()
	nc, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	account := process.Account{URL: url, Node: "node-a", Module: "pacman", User: "node-a.pacman.keeper", Password: "pw"}
	buckets := map[string]jetstream.KeyValue{}
	for _, name := range States {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: account.Module + "_" + name, History: 8})
		if err != nil {
			t.Fatal(err)
		}
		buckets[name] = kv
	}
	declared, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: "mesh-controller_declared-packages"})
	if err != nil {
		t.Fatal(err)
	}
	declared.Put(ctx, "node-a", []byte(`{"node":"node-a","packages":[{"package":"zsh","modules":["shell"],"absent":false}]}`))
	stream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "pacman-events", Subjects: []string{"mesh.mod.pacman.event.>"}})
	if err != nil {
		t.Fatal(err)
	}

	db := &pacmanDB{sl: syncDBs, checkup: Ran{Stdout: "zsh 5.9-1 -> 5.9-2\n"}}
	db.set("glibc 2.42-1\nzsh 5.9-1\n", "zsh\n")
	signals := make(chan Signal)
	k := &Keeper{Config: KeeperConfig{}, Machine: machine(db.runner(), 1000), Clock: realClock{},
		Watcher: fakeWatcher{signals}, Logf: t.Logf, MakeHooks: keeperHooks}
	runCtx, cancel := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- runKeeper(runCtx, account, k) }()

	waitKey := func(kv jetstream.KeyValue, key, holds string) string {
		t.Helper()
		for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
			if e, err := kv.Get(ctx, key); err == nil && strings.Contains(string(e.Value()), holds) {
				return string(e.Value())
			}
		}
		t.Fatalf("%s never held %q", key, holds)
		return ""
	}
	waitKey(buckets[StateInstalled], "node-a", `"zsh"`)
	waitKey(buckets[StateUpdates], "node-a", `"reachable":true`)

	// a transaction: tmux added, zsh upgraded; the lock released.
	db.set("glibc 2.42-1\ntmux 3.5-1\nzsh 5.9-2\n", "tmux\nzsh\n")
	db.setUpdates(Ran{Status: 2})
	signals <- LockReleased
	waitKey(buckets[StateInstalled], "node-a", `"tmux"`)
	waitKey(buckets[StateUpdates], "node-a", `"packages":[]`)

	var changes []PackageChanged
	for deadline := time.Now().Add(15 * time.Second); len(changes) < 2 && time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		changes = nil
		info, err := stream.Info(ctx)
		if err != nil {
			t.Fatal(err)
		}
		for seq := uint64(1); seq <= info.State.LastSeq; seq++ {
			msg, err := stream.GetMsg(ctx, seq)
			if err != nil || msg.Subject != "mesh.mod.pacman.event."+EventPackageChanged {
				continue
			}
			var pc PackageChanged
			if err := json.Unmarshal(msg.Data, &pc); err != nil {
				t.Fatal(err)
			}
			changes = append(changes, pc)
		}
	}
	if len(changes) != 2 {
		t.Fatalf("package.changed at the start and after the transaction: %+v", changes)
	}
	last := changes[1]
	if last.Node != "node-a" || len(last.Added) != 1 || last.Added[0].Name != "tmux" || len(last.Changed) != 1 ||
		last.Changed[0] != (VersionChange{"zsh", "5.9-1", "5.9-2"}) || len(last.Removed) != 0 {
		t.Errorf("the transaction's package.changed: %+v", last)
	}

	// another instance's verbs read what the keeper wrote, asking no machine.
	other := machine(func(_ context.Context, name string, args ...string) Ran {
		t.Errorf("another instance asked a machine: %s %v", name, args)
		return Ran{Status: 99}
	}, 1000)
	other.Node = "node-b"
	other.States = func(name string) StateReader {
		if name == DeclaredPackagesState {
			return kvReader{declared}
		}
		return kvReader{buckets[name]}
	}
	r, err := other.Packages("zsh", 500)
	if err != nil {
		t.Fatal(err)
	}
	on := r["nodes"].([]PackageOnNode)
	if len(on) != 1 || on[0].Node != "node-a" || on[0].Version != "5.9-2" || strings.Join(on[0].DeclaredBy, ",") != "shell" {
		t.Errorf("pacman_packages zsh from another instance: %+v", on)
	}
	r, err = other.Packages("", 500)
	if err != nil {
		t.Fatal(err)
	}
	if v := r["nodes"].([]NodeView); len(v) != 1 || v[0].Packages != 3 || len(v[0].Undeclared) != 1 || v[0].Undeclared[0].Name != "tmux" {
		t.Errorf("the view from another instance: %+v", v)
	}
	u, err := other.UpdatesFor("node-a")
	if err != nil {
		t.Fatal(err)
	}
	if v := u["nodes"].([]UpdatesView); len(v) != 1 || !v[0].Reachable || v[0].Count != 0 || v[0].Changed == nil {
		t.Errorf("pacman_updates from another instance: %+v", v)
	}

	cancel()
	select {
	case err := <-done:
		if !errors.Is(err, context.Canceled) {
			t.Errorf("runKeeper ended with %v, want its context's end", err)
		}
	case <-time.After(10 * time.Second):
		t.Fatal("runKeeper did not stop once its context was cancelled")
	}
}