98 lines
3.3 KiB
Go
98 lines
3.3 KiB
Go
//go:build integration
|
|
|
|
package link
|
|
|
|
import (
|
|
"context"
|
|
"sync/atomic"
|
|
"testing"
|
|
"time"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/delivery"
|
|
"github.com/nats-io/nats.go"
|
|
)
|
|
|
|
// 1132/AC8: against a real bus with the bucket node-engine_have, a node-engine that applies a declaration of two
|
|
// modules writes <node>.<a>, <node>.<b> and <node>.node, and a second apply with nothing changed adds no revision.
|
|
//
|
|
// MESH_TEST_NATS=nats://127.0.0.1:4222 go test -tags integration ./internal/link/ -run TestNatsHavesAreWrittenOnceAndOnlyOnChange
|
|
//
|
|
// The bucket is made here by the test: in the mesh the controller makes it, and the node-engine only binds to it.
|
|
func TestNatsHavesAreWrittenOnceAndOnlyOnChange(t *testing.T) {
|
|
conn, js := aBus(t)
|
|
const node = "haveok"
|
|
kv, err := js.KeyValue(delivery.HaveBucket)
|
|
if err != nil {
|
|
if kv, err = js.CreateKeyValue(&nats.KeyValueConfig{Bucket: delivery.HaveBucket, History: delivery.HaveHistory}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
for _, name := range []string{"a", "b", "node"} {
|
|
_ = kv.Purge(node + "." + name)
|
|
}
|
|
|
|
m, _ := aMember(t)
|
|
m.Node = node
|
|
var failB atomic.Bool
|
|
failB.Store(true)
|
|
facts := func() Report {
|
|
return Report{Applied: []string{"a.x", "b.x"},
|
|
Modules: []ModuleFact{{Module: "a", Build: "b-1"},
|
|
{Module: "b", Build: "b-2", Failed: failB.Load(), Detail: "b.x failed: no space"}},
|
|
Health: healthOf(map[string]string{"a": StateHealthy, "b": StateHealthy}),
|
|
Windows: []Window{{Step: "backup", Holds: []string{"db"}}}}
|
|
}
|
|
q := &Session{Membership: m, Timeout: 2 * time.Second}
|
|
q.Attach(context.Background(), OverNATS{Conn: conn, JS: js})
|
|
ctx := context.Background()
|
|
|
|
revisions := func() map[string]uint64 {
|
|
out := map[string]uint64{}
|
|
for _, name := range []string{"a", "b", "node"} {
|
|
if e, err := kv.Get(node + "." + name); err == nil {
|
|
out[name] = e.Revision()
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
arrives := func(what string) {
|
|
if !q.SayFacts(ctx, facts()) {
|
|
t.Fatal(what + ": the facts were not taken")
|
|
}
|
|
time.Sleep(300 * time.Millisecond)
|
|
}
|
|
arrives("the first facts")
|
|
eventuallyHeld(t, "the three keys were never written", func() bool { return len(revisions()) == 3 })
|
|
first := revisions()
|
|
|
|
a, _ := kv.Get(node + ".a")
|
|
haveA, err := delivery.DecodeHave(a.Value())
|
|
if err != nil || haveA.Phase != delivery.PhaseRunning || haveA.Build != "b-1" || haveA.Health != delivery.HealthOK {
|
|
t.Fatalf("a holds %+v (%v)", haveA, err)
|
|
}
|
|
b, _ := kv.Get(node + ".b")
|
|
haveB, err := delivery.DecodeHave(b.Value())
|
|
if err != nil || haveB.Phase != delivery.PhaseFailed || haveB.Detail == "" {
|
|
t.Fatalf("b holds %+v (%v)", haveB, err)
|
|
}
|
|
n, _ := kv.Get(node + ".node")
|
|
if hn, err := delivery.DecodeHaveNode(n.Value()); err != nil || len(hn.Windows) == 0 {
|
|
t.Fatalf("the node key holds %+v (%v)", hn, err)
|
|
}
|
|
|
|
arrives("the second facts")
|
|
time.Sleep(200 * time.Millisecond)
|
|
for name, rev := range revisions() {
|
|
if rev != first[name] {
|
|
t.Errorf("%s moved from revision %d to %d with nothing changed", name, first[name], rev)
|
|
}
|
|
}
|
|
|
|
failB.Store(false)
|
|
arrives("the third facts")
|
|
eventuallyHeld(t, "b was never updated", func() bool { return revisions()["b"] > first["b"] })
|
|
if now := revisions(); now["a"] != first["a"] || now["node"] != first["node"] {
|
|
t.Errorf("a key that did not change moved: %v from %v", now, first)
|
|
}
|
|
}
|