89 lines
2.7 KiB
Go
89 lines
2.7 KiB
Go
//go:build integration
|
|
|
|
package process
|
|
|
|
import (
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// ADR 0593 §6: a state its owner keeps one bucket per node (composed-<node>) is read as one state: a key is read from
|
|
// the bucket its node part names, the keys of every node's bucket are listed together, and a watch hears every bucket,
|
|
// one created after it began included, while a bucket outside the family is never read.
|
|
func TestAPerNodeStateIsReadAcrossEveryNodesBucket(t *testing.T) {
|
|
url, _ := grantedSendProcess(t)
|
|
_, admin := adminJS(t, url)
|
|
put := func(bucket, key, value string) {
|
|
t.Helper()
|
|
kv, err := admin.CreateOrUpdateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: bucket})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := kv.Put(ctxFor(t), key, []byte(value)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
put("composed-ace", "ace.netcheck.aa", `{"n":1}`)
|
|
put("composed-g14", "g14.netcheck.bb", `{"n":2}`)
|
|
put("composedx-ace", "ace.other.cc", `{"n":3}`)
|
|
|
|
conn, err := Connect(Account{URL: url, Node: "novox", Module: "mesh-provisioner", User: adminUser, Password: adminPass})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer conn.Close()
|
|
family := conn.ReadPerNode("mesh-host.composed", "composed-")
|
|
|
|
if nodes, err := family.Nodes(); err != nil || !slices.Equal(nodes, []string{"ace", "g14"}) {
|
|
t.Fatalf("nodes = %v, %v", nodes, err)
|
|
}
|
|
if e, err := family.Get("g14.netcheck.bb"); err != nil || e == nil || string(e.Value) != `{"n":2}` {
|
|
t.Fatalf("get from g14's bucket: %+v, %v", e, err)
|
|
}
|
|
if e, err := family.Get("laptop.netcheck.bb"); err != nil || e != nil {
|
|
t.Fatalf("a node with no bucket: %+v, %v", e, err)
|
|
}
|
|
if keys, err := family.Keys(); err != nil || !slices.Equal(keys, []string{"ace.netcheck.aa", "g14.netcheck.bb"}) {
|
|
t.Fatalf("keys = %v, %v", keys, err)
|
|
}
|
|
|
|
var mu sync.Mutex
|
|
heard := map[string]bool{}
|
|
err = family.Watch("", func(c stdio.StateChange) error {
|
|
mu.Lock()
|
|
defer mu.Unlock()
|
|
heard[c.Key] = true
|
|
return nil
|
|
})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
put("composed-shanks", "shanks.netcheck.dd", `{"n":4}`)
|
|
put("composed-ace", "ace.netcheck.ee", `{"n":5}`)
|
|
deadline := time.Now().Add(5 * time.Second)
|
|
want := []string{"ace.netcheck.aa", "ace.netcheck.ee", "g14.netcheck.bb", "shanks.netcheck.dd"}
|
|
for {
|
|
mu.Lock()
|
|
var got []string
|
|
for k := range heard {
|
|
got = append(got, k)
|
|
}
|
|
mu.Unlock()
|
|
slices.Sort(got)
|
|
if slices.Equal(got, want) {
|
|
break
|
|
}
|
|
if time.Now().After(deadline) || slices.ContainsFunc(got, func(k string) bool { return strings.HasPrefix(k, "ace.other") }) {
|
|
t.Fatalf("heard %v, want %v", got, want)
|
|
}
|
|
time.Sleep(50 * time.Millisecond)
|
|
}
|
|
}
|