Seat list reader: an integration case over a real bus, from the bootstrap alone to the seat's answer and its events (ADR 0636 §8; issue 1966)
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -0,0 +1,107 @@
|
||||
//go:build integration
|
||||
|
||||
package seatlist_test
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatdef"
|
||||
"git.novox.be/novox/mesh-sdk/go/seatlist"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
// natsBus is a reader's bus over a real NATS connection: a seat verb is a request on the seat's tool subject, a seat's
|
||||
// event a subscription on its event subject.
|
||||
type natsBus struct{ c *nats.Conn }
|
||||
|
||||
func (b natsBus) Ask(key string, body any) (json.RawMessage, error) {
|
||||
seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
|
||||
raw, _ := json.Marshal(body)
|
||||
m, err := b.c.Request("mesh.seat."+seat+".tool."+verb, raw, time.Second)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return m.Data, nil
|
||||
}
|
||||
|
||||
func (b natsBus) Subscribe(pattern string, handle func(string, json.RawMessage) error) error {
|
||||
seat, event, _ := strings.Cut(pattern, ".")
|
||||
_, err := b.c.Subscribe(seatdef.EventSubject(seat, event), func(m *nats.Msg) {
|
||||
_ = handle(seat+"."+event, m.Data)
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// ADR 0636 §8: a reader started with only the bootstrapped definitions, and the seat not yet serving, acts on the bootstrapped
|
||||
// modules alone and holds nothing new; when the seat answers it reads the whole list, and it follows the seat's changed and
|
||||
// removed events, over a real bus.
|
||||
func TestAReaderStartsFromTheBootstrapAndReadsTheSeatOnceItAnswers(t *testing.T) {
|
||||
c, err := nats.Connect(testbus.URL(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer c.Close()
|
||||
bus := natsBus{c}
|
||||
changed := make(chan string, 8)
|
||||
r, err := seatlist.Start(seatlist.Config{Bus: bus, Bootstrap: seatlist.Bootstrapped(),
|
||||
OnChange: func([]catalogue.Seat) { changed <- "changed" }})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st := r.Status(); st.Heard || st.LastError == "" || !r.MayAct("node-engine") || r.MayAct("node-files") {
|
||||
t.Fatalf("with the seat silent: %+v, may act on node-files: %v", st, r.MayAct("node-files"))
|
||||
}
|
||||
|
||||
table := map[string]string{"node-files": `{"format":1,"seat":"node-files","scope":"node","grain":"seat-list","decision":"ADR 1",` +
|
||||
`"verbs":[{"name":"read","input":{"type":"object"}}]}`}
|
||||
serve := func(m *nats.Msg) {
|
||||
var q struct {
|
||||
Seat string `json:"seat"`
|
||||
}
|
||||
_ = json.Unmarshal(m.Data, &q)
|
||||
if q.Seat == "" {
|
||||
out, _ := json.Marshal(seatlist.Listing{Revision: 3, Seats: []seatlist.Entry{{Seat: "node-files", Source: seatlist.SourceTrunk, Digest: "d1"}}})
|
||||
_ = m.Respond(out)
|
||||
return
|
||||
}
|
||||
out, _ := json.Marshal(seatlist.Answer{Seat: q.Seat, Found: true, Revision: 3,
|
||||
Entry: &seatlist.Entry{Seat: q.Seat, Source: seatlist.SourceTrunk, Digest: "d1", Definition: json.RawMessage(table[q.Seat])}})
|
||||
_ = m.Respond(out)
|
||||
}
|
||||
if _, err := c.Subscribe("mesh.seat."+seatlist.Seat+".tool."+seatlist.Verb, serve); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// The seat says a definition changed: the silent reader reads the list at the event, not on a timer.
|
||||
if err := c.Publish(seatdef.EventSubject(seatlist.Seat, seatlist.ChangedEvent), []byte(`{"seat":"node-files","source":"trunk","digest":"d1"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
until(t, changed, func() bool { return r.Has("node-files") })
|
||||
if err := r.Reload(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if st := r.Status(); !st.Heard || st.LastError != "" || !r.MayAct("node-files") || st.Revision != 3 {
|
||||
t.Fatalf("after the seat answered: %+v", st)
|
||||
}
|
||||
if err := c.Publish(seatdef.EventSubject(seatlist.Seat, seatlist.RemovedEvent), []byte(`{"seat":"node-files"}`)); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
until(t, changed, func() bool { return !r.Has("node-files") })
|
||||
}
|
||||
|
||||
// until waits for the reader's own notice of a change (OnChange), never a sleep, until ok holds.
|
||||
func until(t *testing.T, changed <-chan string, ok func() bool) {
|
||||
t.Helper()
|
||||
for !ok() {
|
||||
select {
|
||||
case <-changed:
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("the reader did not follow the seat's event")
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user