jschoubben
b5fdcb88c8
stdio.Emit put them on the module's subject, which no consumer of a seat's event hears. The manifests no longer list a seat's events as the module's own, which the SDK now refuses. mesh-hand-acts takes its act layout and names from mesh-sdk's handacts, and counts a repeated cause by its Repair rule.
256 lines
9.8 KiB
Go
256 lines
9.8 KiB
Go
package main
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"slices"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/health"
|
|
)
|
|
|
|
// memStore is the bus's buckets in memory, counting the writes.
|
|
type memStore struct {
|
|
mu sync.Mutex
|
|
kv map[string]map[string][]byte
|
|
puts map[string]int
|
|
keysErr error
|
|
}
|
|
|
|
func newStore() *memStore {
|
|
return &memStore{kv: map[string]map[string][]byte{}, puts: map[string]int{}}
|
|
}
|
|
|
|
func (s *memStore) Get(state, key string) ([]byte, bool, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
v, ok := s.kv[state][key]
|
|
return v, ok, nil
|
|
}
|
|
|
|
func (s *memStore) Put(state, key string, value any) error {
|
|
raw, err := json.Marshal(value)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.kv[state] == nil {
|
|
s.kv[state] = map[string][]byte{}
|
|
}
|
|
s.kv[state][key] = raw
|
|
s.puts[state+"/"+key]++
|
|
return nil
|
|
}
|
|
|
|
func (s *memStore) Delete(state, key string) error {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
delete(s.kv[state], key)
|
|
return nil
|
|
}
|
|
|
|
func (s *memStore) Keys(state string) ([]string, error) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.keysErr != nil && state == FindingsState {
|
|
return nil, s.keysErr
|
|
}
|
|
var out []string
|
|
for k := range s.kv[state] {
|
|
out = append(out, k)
|
|
}
|
|
sort.Strings(out)
|
|
return out, nil
|
|
}
|
|
|
|
func (s *memStore) findingKeys() []string {
|
|
keys, _ := s.Keys(FindingsState)
|
|
return keys
|
|
}
|
|
|
|
// checker is a self-check over memory, with a registry the test drives.
|
|
func checker(store *memStore, registry ...Probe) (*Checker, *[]string) {
|
|
var said []string
|
|
now := time.Date(2026, 10, 29, 21, 0, 0, 0, time.UTC)
|
|
c := &Checker{Store: store, Registry: registry, Host: "novox", Logf: func(string, ...any) {},
|
|
Now: func() time.Time { now = now.Add(time.Second); return now },
|
|
Publish: func(subject string, body any, id string) error {
|
|
raw, _ := json.Marshal(body)
|
|
said = append(said, subject+" "+id+" "+string(raw))
|
|
return nil
|
|
}}
|
|
return c, &said
|
|
}
|
|
|
|
// A run writes what a probe finds, writes nothing it already wrote, and deletes what it no longer sees; a probe that
|
|
// could not run clears nothing and is written as its own finding at the second run in a row; a finding one look can
|
|
// be wrong about waits for the second run; a deferred probe is never a pass (ADR 0564 §2, §6; issue 1952 SC1).
|
|
func TestARunWritesFindingsAsTheyChange(t *testing.T) {
|
|
var found []Observation
|
|
var fail error
|
|
steady := Probe{ID: "P1", Kind: "stream-wrong", Run: func(context.Context, *Checker) ([]Observation, error) { return found, fail }}
|
|
deferred := Probe{ID: "P2", Kind: "resolver-wrong", Deferred: "the overlay's addresses"}
|
|
missing := Observation{Token: "stream-missing", Severity: Urgent, Subject: Subject{Scope: ScopeBus, ID: "EVENTS"},
|
|
Summary: "the stream EVENTS is not on the bus"}
|
|
behind := Observation{Kind: "consumer-behind", Token: "behind", Severity: Warning, Confirm: true,
|
|
Subject: Subject{Scope: ScopeBus, ID: "EVENTS.controller"}, Summary: "far behind"}
|
|
store := newStore()
|
|
c, said := checker(store, steady, deferred)
|
|
|
|
steps := []struct {
|
|
name string
|
|
found []Observation
|
|
fail error
|
|
want []string // finding keys written after the run
|
|
verdict string
|
|
missingPs int // puts of the missing finding so far
|
|
}{
|
|
{"a finding is written", []Observation{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"the same again is no write", []Observation{missing}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"one look can be wrong: held", []Observation{missing, behind}, nil, []string{"bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"seen twice: written", []Observation{missing, behind}, nil,
|
|
[]string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing"}, VerdictFail, 1},
|
|
{"blind once: clears nothing, held", nil, errors.New("timed out"),
|
|
[]string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing"}, VerdictFailedToRun, 1},
|
|
{"blind twice: its own finding", nil, errors.New("timed out"),
|
|
[]string{"bus.EVENTS.controller.behind", "bus.EVENTS.stream-missing", "probe.P1.failed"}, VerdictFailedToRun, 1},
|
|
{"no longer seen: cleared", nil, nil, nil, VerdictPass, 1},
|
|
}
|
|
for _, s := range steps {
|
|
found, fail = s.found, s.fail
|
|
run := c.RunOnce(context.Background(), "the test")
|
|
if got := store.findingKeys(); !slices.Equal(got, s.want) {
|
|
t.Errorf("%s: findings %v, want %v", s.name, got, s.want)
|
|
}
|
|
if run.Probes[0].Verdict != s.verdict {
|
|
t.Errorf("%s: P1 %s, want %s", s.name, run.Probes[0].Verdict, s.verdict)
|
|
}
|
|
if run.Probes[1].Verdict != VerdictDeferred || !strings.Contains(run.Probes[1].Error, "overlay") {
|
|
t.Errorf("%s: a deferred probe answered %+v", s.name, run.Probes[1])
|
|
}
|
|
if n := store.puts[FindingsState+"/bus.EVENTS.stream-missing"]; n != s.missingPs {
|
|
t.Errorf("%s: the unchanged finding was written %d times", s.name, n)
|
|
}
|
|
}
|
|
raw, _, _ := store.Get(FindingsState, "probe.P1.failed")
|
|
if raw != nil {
|
|
t.Errorf("the blind finding outlived the probe running again: %s", raw)
|
|
}
|
|
// Every run is kept, the last as `last`, and said as checked with the controller's heartbeat body.
|
|
if keys, _ := store.Keys(VerdictsState); len(keys) != len(steps) {
|
|
t.Errorf("verdicts kept %d runs, want %d", len(keys), len(steps))
|
|
}
|
|
// Said on the seat's subject, where its consumers hear it, de-duplicated by the run's id.
|
|
if len(*said) != len(steps) || !strings.HasPrefix((*said)[0], "mesh.seat.mesh-self-check.event.checked doctor-") {
|
|
t.Fatalf("said %v", *said)
|
|
}
|
|
_, rest, _ := strings.Cut((*said)[0], " ")
|
|
_, rest, _ = strings.Cut(rest, " ")
|
|
var body map[string]any
|
|
_ = json.Unmarshal([]byte(rest), &body)
|
|
for _, field := range []string{"run", "at", "started", "took", "interval-seconds", "counts", "probes", "controller", "why"} {
|
|
if _, ok := body[field]; !ok {
|
|
t.Errorf("checked's body lacks %s, which doctor-heartbeat carries: %v", field, body)
|
|
}
|
|
}
|
|
last, _, _ := store.Get(VerdictState, LastKey)
|
|
var r Run
|
|
_ = json.Unmarshal(last, &r)
|
|
if r.Counts.Passed != 1 || r.Counts.Deferred != 1 {
|
|
t.Errorf("the last verdict counts %+v", r.Counts)
|
|
}
|
|
// A finding carries the operator's words of its kind, the probe that found it and who observed it.
|
|
found, fail = []Observation{missing}, nil
|
|
c.RunOnce(context.Background(), "the test")
|
|
raw, _, _ = store.Get(FindingsState, "bus.EVENTS.stream-missing")
|
|
f, err := health.DecodeFinding(raw)
|
|
if err != nil || f.Kind != "stream-wrong" || f.Source != "P1" || f.Observer != Module || f.Headline == "" ||
|
|
f.Explanation == "" || f.Token != "stream-missing" || len(f.Evidence) != 1 || f.Run == "" {
|
|
t.Errorf("the written finding is %+v", f)
|
|
}
|
|
// Blind to its own bucket, a run writes and clears nothing.
|
|
store.keysErr = errors.New("the bus is away")
|
|
found = nil
|
|
c.RunOnce(context.Background(), "the test")
|
|
store.keysErr = nil
|
|
if got := store.findingKeys(); !slices.Equal(got, []string{"bus.EVENTS.stream-missing"}) {
|
|
t.Errorf("a run blind to its findings changed them: %v", got)
|
|
}
|
|
}
|
|
|
|
// A probe that panics or overruns its bound could not run: never a pass.
|
|
func TestAProbeThatPanicsOrOverrunsCouldNotRun(t *testing.T) {
|
|
within := Within
|
|
Within = 50 * time.Millisecond
|
|
t.Cleanup(func() { Within = within })
|
|
for name, run := range map[string]func(context.Context, *Checker) ([]Observation, error){
|
|
"panic": func(context.Context, *Checker) ([]Observation, error) { panic("boom") },
|
|
"overrun": func(ctx context.Context, _ *Checker) ([]Observation, error) { time.Sleep(time.Second); return nil, nil },
|
|
} {
|
|
c, _ := checker(newStore(), Probe{ID: "P", Kind: "k", Run: run})
|
|
c.Now = time.Now
|
|
r := c.RunOnce(context.Background(), "the test")
|
|
if r.Probes[0].Verdict != VerdictFailedToRun || r.Counts.Passed != 0 {
|
|
t.Errorf("%s: %+v", name, r.Probes[0])
|
|
}
|
|
}
|
|
}
|
|
|
|
// The verb answers the last run, one probe whole, a run now, and before any run says when one comes; the reads answer
|
|
// every state the module keeps (ADR 0564 §12).
|
|
func TestTheVerbsAndReads(t *testing.T) {
|
|
store := newStore()
|
|
many := make([]Observation, 25)
|
|
for i := range many {
|
|
many[i] = Observation{Severity: Warning, Subject: Subject{Scope: ScopeBus, ID: fmt.Sprintf("S%02d", i)}, Summary: "x"}
|
|
}
|
|
c, _ := checker(store, Probe{ID: "D7", Kind: "stream-wrong",
|
|
Run: func(context.Context, *Checker) ([]Observation, error) { return many, nil }})
|
|
if _, err := c.Doctor(false, "", "x"); err == nil || !strings.Contains(err.Error(), "first run") {
|
|
t.Fatalf("before a run: %v", err)
|
|
}
|
|
if _, err := c.Doctor(true, "", "asked by person.jochen"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
over, _ := c.Doctor(false, "", "")
|
|
if v := over["run"].(Run).Probes[0]; len(v.Found) != overviewFindings || v.More == "" {
|
|
t.Errorf("the overview carries %d findings, more %q", len(v.Found), v.More)
|
|
}
|
|
whole, _ := c.Doctor(false, "d7", "")
|
|
if v := whole["run"].(Run).Probes[0]; len(v.Found) != 25 {
|
|
t.Errorf("one probe whole carries %d findings", len(v.Found))
|
|
}
|
|
if _, err := c.Doctor(false, "D99", ""); err == nil {
|
|
t.Error("an unknown probe answered")
|
|
}
|
|
// A fresh checker over the same bucket answers the last verdict from it.
|
|
fresh, _ := checker(store)
|
|
if _, err := fresh.Doctor(false, "", ""); err != nil {
|
|
t.Errorf("the last verdict is not read from its bucket: %v", err)
|
|
}
|
|
fs, err := c.Findings("")
|
|
if err != nil || fs.(map[string]any)["count"] != 25 {
|
|
t.Errorf("findings answered %v, %v", fs, err)
|
|
}
|
|
if one, err := c.Findings("bus.S03.stream-wrong"); err != nil || !strings.Contains(string(one.(json.RawMessage)), "S03") {
|
|
t.Errorf("one finding answered %v, %v", one, err)
|
|
}
|
|
vs, err := c.Verdicts("", 0)
|
|
if err != nil || len(vs.(map[string]any)["verdicts"].([]map[string]any)) != 1 {
|
|
t.Errorf("verdicts answered %v, %v", vs, err)
|
|
}
|
|
if _, err := c.Verdicts("doctor-1-1", 0); err == nil {
|
|
t.Error("an unknown run answered")
|
|
}
|
|
if p := c.Probes()["probes"].([]map[string]any); len(p) != 1 || p[0]["id"] != "D7" {
|
|
t.Errorf("probes answered %v", p)
|
|
}
|
|
}
|