mesh-catalog/modules/mesh-self-check/cmd/mesh-self-check/probes_test.go

package main

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

	"git.novox.be/novox/mesh-sdk/go/buscompose"
)

// bus is the bus module's answers as a probe asks them: every mesh stream and consumer as defined, unless changed.
type bus struct {
	drop     map[string]bool // stream or stream.consumer missing
	change   func(rows map[string]*streamRow)
	closed   string
	signals  error
	asksFail error
}

func (b bus) rows() map[string]*streamRow {
	rows := map[string]*streamRow{}
	for _, s := range buscompose.MeshStreams() {
		if b.drop[s.Name] {
			continue
		}
		retention := "limits"
		per := int64(s.MaxMsgsPerSubject)
		switch s.Retention {
		case buscompose.RetentionWorkQueue:
			retention = "workqueue"
		case buscompose.RetentionLastPerSubject:
			per = 1
		}
		rows[s.Name] = &streamRow{Name: s.Name, Subjects: s.Subjects, Retention: retention, MaxPerSubject: per}
	}
	for _, c := range buscompose.MeshConsumers() {
		r := rows[c.Stream]
		if r == nil || b.drop[c.Stream+"."+c.Name] {
			continue
		}
		kind := "pull"
		if c.Push || c.Queue != "" {
			kind = "push"
		}
		r.Consumers = append(r.Consumers, consumerRow{Name: c.Name, Kind: kind, Filters: c.Filters, MaxDeliver: int64(c.MaxDeliver)})
	}
	if b.change != nil {
		b.change(rows)
	}
	return rows
}

func (b bus) ask(key string, body any) (json.RawMessage, error) {
	if b.asksFail != nil {
		return nil, b.asksFail
	}
	switch key {
	case StreamsTool:
		stream, _ := body.(map[string]any)["stream"].(string)
		var detail []streamRow
		for name, r := range b.rows() {
			if stream == "" {
				row := *r
				row.Consumers = nil
				detail = append(detail, row)
			} else if name == stream {
				detail = append(detail, *r)
			}
		}
		if stream != "" && len(detail) == 0 {
			return nil, errors.New("no stream called " + stream)
		}
		return json.Marshal(map[string]any{"detail": detail})
	case ClosedTool:
		if b.closed == "" {
			return json.RawMessage(`{"hours":1,"reaches":true,"users":[]}`), nil
		}
		return json.RawMessage(b.closed), nil
	case SignalsVerb:
		if b.signals != nil {
			return nil, b.signals
		}
		return json.RawMessage(`{"signals":[]}`), nil
	}
	return nil, errors.New("nats: no responders available for request")
}

func keysOf(found []Observation) []string {
	var out []string
	for _, f := range found {
		out = append(out, f.Key())
	}
	return out
}

// Each probe that moved judges the bus module's answers with its assertion unchanged (ADR 0564 §6; issue 1952 SC2).
func TestTheBusProbes(t *testing.T) {
	cases := []struct {
		name  string
		probe func(context.Context, *Checker) ([]Observation, error)
		bus   bus
		want  []string
		err   string
	}{
		{"D6 all as defined", probeConsumers, bus{}, nil, ""},
		{"D6 a consumer gone", probeConsumers, bus{drop: map[string]bool{"DEAD_LETTER_NOTICES.controller": true}},
			[]string{"bus.DEAD_LETTER_NOTICES.controller.missing"}, ""},
		{"D6 a stream gone takes its consumers", probeConsumers, bus{drop: map[string]bool{"CONTROL": true}},
			[]string{"bus.CONTROL.controller.missing"}, ""},
		{"D6 redefined and far behind", probeConsumers, bus{change: func(rows map[string]*streamRow) {
			for i, c := range rows["EVENTS"].Consumers {
				if c.Name == "controller" {
					rows["EVENTS"].Consumers[i].MaxDeliver, rows["EVENTS"].Consumers[i].Pending = 9, 5000
				}
			}
		}}, []string{"bus.EVENTS.controller.behind", "bus.EVENTS.controller.redefined"}, ""},
		{"D6 the bus module silent", probeConsumers, bus{asksFail: errors.New("no responders")}, nil, "did not answer"},
		{"D7 all as defined", probeStreams, bus{}, nil, ""},
		{"D7 missing and redefined", probeStreams, bus{drop: map[string]bool{"NODES": true}, change: func(rows map[string]*streamRow) {
			rows["EVENTS"].Retention = "workqueue"
		}}, []string{"bus.EVENTS.stream-redefined", "bus.NODES.stream-missing"}, ""},
		{"D15 quiet", probeReconnects, bus{}, nil, ""},
		{"D15 a loop", probeReconnects, bus{closed: `{"hours":1,"reaches":true,"users":[
			{"user":"g14.node-tools","closed":30,"dropped":20,"names":[{"name":"node-tools","dropped":20,"reasons":{"Stale Connection":20}}]},
			{"user":"ace.records","closed":5,"dropped":3},{"user":"(unauthenticated)","closed":90,"dropped":90}]}`},
			[]string{"bus.g14.node-tools.bus-reconnects"}, ""},
		{"DW answers", probeWatchdogs, bus{}, nil, ""},
		{"DW silent", probeWatchdogs, bus{signals: errors.New("no responders")}, []string{"core.watchdogs.silent"}, ""},
		{"H-bus healthy", probeBusHealth, bus{}, nil, ""},
		{"H-bus a stream and a consumer missing", probeBusHealth,
			bus{drop: map[string]bool{"NODES": true, "EVENTS.controller-merges": true}}, []string{"bus.mesh.unhealthy"}, ""},
		{"H-bus no round trip", probeBusHealth, bus{asksFail: errors.New("timeout")}, nil, "no request crossed"},
	}
	for _, tc := range cases {
		c := &Checker{Ask: tc.bus.ask}
		found, err := tc.probe(context.Background(), c)
		if tc.err != "" {
			if err == nil || !strings.Contains(err.Error(), tc.err) {
				t.Errorf("%s: error %v, want %q", tc.name, err, tc.err)
			}
			continue
		}
		if err != nil {
			t.Errorf("%s: %v", tc.name, err)
			continue
		}
		if got := keysOf(found); !slices.Equal(got, tc.want) {
			t.Errorf("%s: found %v, want %v", tc.name, got, tc.want)
		}
	}
}

// The registry holds exactly the probes ADR 0564's table moves here, each with a kind, and a deferred one names what
// it lacks; the probes that end with the controller or moved to the watchman are not in it (issue 1952 SC4).
func TestTheRegistryIsADR0564sTable(t *testing.T) {
	moved := []string{"D2", "D3", "D4", "D6", "D7", "D8", "D11", "D12", "D13", "D15", "DA", "D-root", "D-fill", "D-heard",
		"DW", "H-bus"}
	var ids []string
	for _, p := range Registry() {
		ids = append(ids, p.ID)
		if p.Kind == "" || p.Asserts == "" {
			t.Errorf("%s has no kind or assertion", p.ID)
		}
		if (p.Run == nil) == (p.Deferred == "") {
			t.Errorf("%s is neither run nor deferred naming what it lacks", p.ID)
		}
	}
	slices.Sort(ids)
	slices.Sort(moved)
	if !slices.Equal(ids, moved) {
		t.Errorf("the registry holds %v, ADR 0564 moves %v", ids, moved)
	}
}