mesh-controller/internal/catalogue/process_bus_test.go

package catalogue

import (
	"strings"
	"testing"
)

// A process resource may declare `bus` (novox/hq ADR 0436 §1): what it emits, the module's own work-queue
// streams it consumes, the module's own state it reads and writes, whether it reads the router's records of
// this module's own asks, and the tools and seat verbs it calls. The fixture is mesh-mail's future manifest
// shape (ADR 0436 §7, §8), copied from mesh-catalog/modules/mesh-mail/module.json and extended as settled by
// the lead: process id "send" is the manifest's own `id` (internal/catalogue/runtime.go:72-74 is the model for
// a process id); the account file uses the controller's real JSON tags (modules.go:1041-1056 issueWith).
func mailSenderProcessManifest(t *testing.T, bus string) string {
	t.Helper()
	process := `{"id":"send","type":"process","run":["./mesh-mail","send"],"user":"mesh-mail-send"`
	if bus != "" {
		process += `,"bus":` + bus
	}
	process += `}`
	// The stream itself is plain here, not retention work-queue: 1179 checks only that a process names a
	// stream the module owns, never its retention — that rule is 1178's (WorkQueueStreamProblems), and is
	// tested against its own fixture there (work_queue_consumer_test.go) so this one stays accepted whether
	// or not a process currently names "mail-sends" in bus.queues.
	return `{"module":"mesh-mail","version":"1","emits":["mail.send.sent","mail.send.failed"],` +
		`"state":["sends"],"streams":[{"name":"mail-sends","events":["mail.send.approved"],"max-age-days":1,` +
		`"max-mib":64}],"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` + process + `]}`
}

const mailSenderBus = `{"emits":["mail.send.sent","mail.send.failed"],"queues":["mail-sends"],` +
	`"state":["sends"],"asks":true,"invokes":["seat:operator-channel.tell"]}`

// 1179/AC1: a well-formed bus on a process whose account the module makes is accepted, and the catalogue
// gives it whole to the grant composition (ProcessBuses, which 1180 reads).
func TestAProcessDeclaresItsBusAndIsAccepted(t *testing.T) {
	m, err := ParseManifest([]byte(mailSenderProcessManifest(t, mailSenderBus)))
	if err != nil {
		t.Fatalf("a well-formed process bus was refused: %v", err)
	}
	buses := m.ProcessBuses()
	b, ok := buses["send"]
	if !ok {
		t.Fatalf("the process's bus was not given to the grant composition: %+v", buses)
	}
	if !strSliceEqual(b.Emits, []string{"mail.send.sent", "mail.send.failed"}) ||
		!strSliceEqual(b.Queues, []string{"mail-sends"}) || !strSliceEqual(b.State, []string{"sends"}) ||
		!b.Asks || !strSliceEqual(b.Invokes, []string{"seat:operator-channel.tell"}) {
		t.Fatalf("the bus read as %+v", b)
	}
}

func strSliceEqual(a, b []string) bool {
	if len(a) != len(b) {
		return false
	}
	for i := range a {
		if a[i] != b[i] {
			return false
		}
	}
	return true
}

// 1179/AC2: a bus on a process with no user, user root, or the operator account is refused naming the
// process and the account rule.
func TestAProcessBusAccountMustBeTheModulesOwn(t *testing.T) {
	for name, c := range map[string]struct{ user, says string }{
		"no user":              {"", "send"},
		"user root":            {"root", "root"},
		"the operator account": {"${machine:account}", "operator"},
	} {
		raw := `{"module":"mesh-mail","version":"1","emits":["mail.send.sent"],"state":["sends"],` +
			`"streams":[{"name":"mail-sends","events":["mail.send.approved"],"retention":"work-queue",` +
			`"max-deliveries":5}],"resources":[{"id":"account","type":"user","name":"mesh-mail-send"},` +
			`{"id":"send","type":"process","run":["./mesh-mail","send"]`
		if c.user != "" {
			raw += `,"user":"` + c.user + `"`
		}
		raw += `,"bus":{"emits":["mail.send.sent"]}}]}`
		_, err := ParseManifest([]byte(raw))
		if err == nil || !strings.Contains(err.Error(), "send") || !strings.Contains(err.Error(), c.says) {
			t.Errorf("%s: %v; want a refusal naming the process and %q", name, err, c.says)
		}
	}
}

// 1179/AC3: a bus naming a stream, state or event the module does not declare as its own is refused naming
// it.
func TestAProcessBusNamesOnlyTheModulesOwn(t *testing.T) {
	for name, c := range map[string]struct{ bus, says string }{
		"another module's event": {`{"emits":["billing.paid"]}`, "billing.paid"},
		"another module's queue": {`{"queues":["other-queue"]}`, "other-queue"},
		"another module's state": {`{"state":["other-state"]}`, "other-state"},
	} {
		_, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus)))
		if err == nil || !strings.Contains(err.Error(), c.says) {
			t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says)
		}
	}
}

// 1179/AC4: a bus declaring serves, holds, claims or a seat is refused naming the key; a bus with invokes of
// a tool or a seat verb is accepted.
func TestAProcessBusNeverServesOrHoldsASeat(t *testing.T) {
	for name, c := range map[string]struct{ bus, says string }{
		"serves": {`{"emits":["mail.send.sent"],"serves":["send"]}`, "serves"},
		"holds":  {`{"emits":["mail.send.sent"],"holds":["mail-sender"]}`, "holds"},
		"claims": {`{"emits":["mail.send.sent"],"claims":["mail-sender"]}`, "claims"},
		"seat":   {`{"emits":["mail.send.sent"],"seat":"mail-sender"}`, "seat"},
	} {
		_, err := ParseManifest([]byte(mailSenderProcessManifest(t, c.bus)))
		if err == nil || !strings.Contains(err.Error(), c.says) {
			t.Errorf("%s: %v; want a refusal naming %q", name, err, c.says)
		}
	}
	accepted := `{"emits":["mail.send.sent","mail.send.failed"],"invokes":["seat:operator-channel.tell","mail.status"]}`
	if _, err := ParseManifest([]byte(mailSenderProcessManifest(t, accepted))); err != nil {
		t.Fatalf("a bus invoking a tool and a seat verb was refused: %v", err)
	}
}

// 1179/AC5: a process declaring bus is composed with no bus key on the wire, and is otherwise byte for byte
// what main composes for the same process without bus.
func TestAProcessesBusNeverReachesTheWire(t *testing.T) {
	withBus, err := ParseManifest([]byte(mailSenderProcessManifest(t, mailSenderBus)))
	if err != nil {
		t.Fatal(err)
	}
	withoutBus, err := ParseManifest([]byte(mailSenderProcessManifest(t, "")))
	if err != nil {
		t.Fatal(err)
	}
	with := Rendering{Needed: map[string]map[string]string{}}
	outWith, err := Resolution{Node: "novox", Modules: []Manifest{withBus}}.Declaration(with)
	if err != nil {
		t.Fatalf("composing with bus: %v", err)
	}
	outWithout, err := Resolution{Node: "novox", Modules: []Manifest{withoutBus}}.Declaration(with)
	if err != nil {
		t.Fatalf("composing without bus: %v", err)
	}
	var withProcess, withoutProcess map[string]any
	for _, r := range outWith {
		if r["id"] == "mesh-mail.send" {
			withProcess = r
		}
	}
	for _, r := range outWithout {
		if r["id"] == "mesh-mail.send" {
			withoutProcess = r
		}
	}
	if withProcess == nil || withoutProcess == nil {
		t.Fatalf("the process was not composed at all: with=%v without=%v", withProcess, withoutProcess)
	}
	if _, has := withProcess["bus"]; has {
		t.Fatalf("the process resource carried a bus key onto the wire: %v", withProcess)
	}
	delete(withProcess, "bus")
	if !mapsEqual(withProcess, withoutProcess) {
		t.Fatalf("with bus declared, the process composed differently: %+v\nwant: %+v", withProcess, withoutProcess)
	}
}

func mapsEqual(a, b map[string]any) bool {
	if len(a) != len(b) {
		return false
	}
	for k, v := range a {
		bv, ok := b[k]
		if !ok {
			return false
		}
		if av, ok := v.([]any); ok {
			bbv, ok := bv.([]any)
			if !ok || len(av) != len(bbv) {
				return false
			}
			for i := range av {
				if av[i] != bbv[i] {
					return false
				}
			}
			continue
		}
		if v != bv {
			return false
		}
	}
	return true
}