Files
mesh-catalog/modules/nats/cmd/nats-tools/keep_test.go
T

148 lines
7.3 KiB
Go

package main
import (
"encoding/json"
"errors"
"fmt"
"strings"
"testing"
"time"
"git.novox.be/novox/mesh-contracts/go/seats/meshbroker"
"git.novox.be/novox/mesh-sdk/go/busretry"
"git.novox.be/novox/mesh-sdk/go/seatlist"
catalog "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog"
"nats-tools/internal/keeper"
)
// The catalog half as the keeper plans from it (design 47 G §532, §533, §547, §591): every module some watched branch's
// newest revision lists, each local name from the first of its watched branches in its registration's order whose entry
// asks it, another branch asking it with other settings a drift naming that branch; a branch the registration does not
// name after; a verdict not accepted asks nothing; the claims of every branch; a foundation part and its old sections'
// caps from the first branch.
func TestTheCatalogHalfIsReadPerWatchedBranch(t *testing.T) {
entryOf := func(verdict string, manifest map[string]any) catalog.CatalogEntry {
manifest["module"] = "m"
return catalog.CatalogEntry{Module: "m", Verdict: verdict, ManifestDigest: "sha256:" + verdict, Manifest: manifest}
}
state := func(opts string) map[string]any {
e := map[string]any{"name": "drafts", "storage": "key-value", "template": "state"}
if opts != "" {
e["options"] = map[string]any{"history": opts}
}
return e
}
log := map[string]any{"name": "log", "storage": "stream", "template": "log"}
main := entryOf("accepted", map[string]any{"bus-state": []any{state("")}, "claims": []any{map[string]any{"name": "a"}}})
feature := entryOf("accepted", map[string]any{"bus-state": []any{state("5"), log}, "claims": []any{map[string]any{"name": "b"}}})
for _, row := range []struct {
name string
branches []Branch
order []string
want string
}{
{"one branch", []Branch{{Name: "main", Revision: 3, Entries: map[string]catalog.CatalogEntry{"m": main}}}, []string{"main"},
"m 3 [drafts/key-value/state] [a] false 0;"},
{"a feature branch asking a local name with other settings drifts, its new one planned",
[]Branch{{Name: "feature", Revision: 9, Entries: map[string]catalog.CatalogEntry{"m": feature}}, {Name: "main", Revision: 3, Entries: map[string]catalog.CatalogEntry{"m": main}}},
[]string{"main", "feature"}, "m 3 [drafts/key-value/state log/stream/log] [a b] false 0; drifted m drafts feature from main"},
{"the registration's order decides", []Branch{{Name: "main", Revision: 3, Entries: map[string]catalog.CatalogEntry{"m": main}}, {Name: "feature", Revision: 9, Entries: map[string]catalog.CatalogEntry{"m": feature}}},
[]string{"feature"}, "m 9 [drafts/key-value/state log/stream/log] [b a] false 0; drifted m drafts main from feature"},
{"a verdict not accepted asks nothing", []Branch{{Name: "main", Revision: 3, Entries: map[string]catalog.CatalogEntry{"m": entryOf("refused", map[string]any{"bus-state": []any{state("")}})}}}, []string{"main"},
"m 3 [] [] false 0;"},
{"a foundation part, its old sections counted", []Branch{{Name: "main", Revision: 3, Entries: map[string]catalog.CatalogEntry{"m": entryOf("accepted", map[string]any{
"objects": []any{map[string]any{"name": "sent-objects", "max-mib": 256}}, "logs": []any{map[string]any{"name": "l", "max-mib": 4}}, "streams": []any{map[string]any{"name": "s"}}})}}}, []string{"main"},
"m 3 [] [] true 1284;"},
} {
t.Run(row.name, func(t *testing.T) {
v, err := ViewOf(row.branches, func(string) []string { return row.order })
if err != nil {
t.Fatal(err)
}
var got string
for _, m := range v.Modules {
es := []string{}
for _, e := range m.BusState {
es = append(es, e.Name+"/"+e.Storage+"/"+e.Template)
}
got += fmt.Sprintf("%s %d [%s] %v %v %d;", m.Module, m.Revision, strings.Join(es, " "), m.Claims, m.Foundation, m.OldBytes/keeper.MiB)
}
for _, d := range v.Drifts {
got += fmt.Sprintf(" drifted %s %s %s from %s", d.Module, d.Local, d.Branch, d.From)
}
if got != row.want {
t.Errorf("read %q, want %q", got, row.want)
}
})
}
}
// The keeper's health tool (CAT-N6): healthy naming its pass; unhealthy, naming why, when its pass is old or verdicts
// wait on its consumer `<node>_nats_keeper` past the bound.
func TestTheKeepersHealthToolReadsItsStatusAndItsConsumer(t *testing.T) {
at := time.Date(2026, 11, 4, 12, 0, 0, 0, time.UTC)
jsz := `{"account_details":[{"stream_detail":[{"name":"EVENTS","consumer_detail":[{"name":"anchor_nats_keeper","num_pending":3},{"name":"anchor_nats","num_pending":9}]}]}]}`
status := keeper.Status{Started: at.Add(-time.Hour), Pass: at.Add(-time.Minute), Heard: at.Add(-time.Minute),
FirstView: keeper.FirstViewStatus{Owner: keeper.FirstViewOwner, Verb: keeper.FirstViewVerb, Answered: at.Add(-time.Hour)}}
none := func() (map[string]keeper.Asked, error) { return nil, nil }
for _, row := range []struct {
name string
now time.Time
state string
why string
}{
{"its pass is fresh", at, "healthy", "last pass 2026-11-04T11:59:00Z"},
{"its pass and its verdicts are old", at.Add(5 * time.Minute), "unhealthy", "3 accepted verdict(s) wait on its consumer"},
} {
t.Run(row.name, func(t *testing.T) {
a := keeperHealth(func() (keeper.Status, error) { return status, nil }, none,
func() ([]byte, error) { return []byte(jsz), nil }, func() time.Time { return row.now })
if a.State != row.state || !strings.Contains(a.Why, row.why) {
t.Errorf("answered %+v: %s, naming %q", a, row.state, row.why)
}
})
}
}
// listingCatalog is the seat mesh-catalog as the reader asks it: an empty listing, so the bootstrapped definitions stand, or,
// silent, no answer at all.
type listingCatalog struct{ silent bool }
func (c listingCatalog) Ask(string, any) (json.RawMessage, error) {
if c.silent {
return nil, errors.New("no responders")
}
return json.Marshal(seatlist.Listing{Revision: 1})
}
func (listingCatalog) Subscribe(string, func(string, json.RawMessage) error) error { return nil }
// A-179, A-186: the definition the seat mesh-broker's snapshot verbs check their callers against is the seat list's, none
// before the reader started or while the seat mesh-catalog has not answered, and none for a seat the list does not hold.
func TestTheAnsweredSeatsDefinitionIsTheSeatListsOnceRead(t *testing.T) {
t.Cleanup(busretry.Waits(20*time.Millisecond, time.Millisecond, 5*time.Millisecond))
follow := func(silent bool) *catalogFollow {
r, err := seatlist.Start(seatlist.Config{Bus: listingCatalog{silent: silent}, Bootstrap: seatlist.Bootstrapped()})
if err != nil {
t.Fatal(err)
}
<-r.Ready()
return &catalogFollow{seatList: r}
}
if _, ok := seatlist.Definition(func() *seatlist.Reader { return (&catalogFollow{}).seatList }, meshbroker.Seat)(); ok {
t.Error("a definition is answered with no seat list")
}
if _, ok := seatlist.Definition(func() *seatlist.Reader { return follow(true).seatList }, meshbroker.Seat)(); ok {
t.Error("a definition is answered while the seat mesh-catalog has not answered the list")
}
read := follow(false)
if d, ok := seatlist.Definition(func() *seatlist.Reader { return read.seatList }, meshbroker.Seat)(); !ok || d.Seat != meshbroker.Seat {
t.Errorf("the read list's %s is answered as %q, %v", meshbroker.Seat, d.Seat, ok)
}
if _, ok := seatlist.Definition(func() *seatlist.Reader { return read.seatList }, "no-such-seat")(); ok {
t.Error("a seat the list does not hold is answered a definition")
}
}