118 lines
3.6 KiB
Go
118 lines
3.6 KiB
Go
//go:build integration
|
|
|
|
package conditions
|
|
|
|
import (
|
|
"context"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-controller/internal/broker"
|
|
|
|
"github.com/novox/mesh-controller/internal/testbus"
|
|
)
|
|
|
|
// The condition store against a real server: compare-and-set, an unreadable store refused, and the
|
|
// history read back by time are claims about what the bus does.
|
|
|
|
func busStoreForTest(t *testing.T) (*broker.JetStream, Backend, History) {
|
|
t.Helper()
|
|
url := testbus.URL(t)
|
|
js, err := broker.Dial(url)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(js.Close)
|
|
api, err := jetstream.New(js.Conn())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = api.DeleteKeyValue(t.Context(), broker.ConditionsBucket)
|
|
_ = api.DeleteKeyValue(t.Context(), broker.ConditionHistoryBucket)
|
|
if err := js.EnsureControllerBuckets(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
store, history, err := OnTheBus(t.Context(), js.Conn())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return js, store, history
|
|
}
|
|
|
|
// **A condition outlives the controller that raised it**, and two writers on the bus cannot lose each
|
|
// other's word: a stale revision is refused as moved.
|
|
func TestNatsTheStoreKeepsConditionsByCompareAndSet(t *testing.T) {
|
|
_, store, history := busStoreForTest(t)
|
|
ctx := t.Context()
|
|
told, looks := &Told{}, NewMemoryLooks()
|
|
k := NewKeeper(ctx, Options{Store: store, History: history, Teller: told, Looks: looks})
|
|
if _, err := k.Observe(ctx, silent("ace")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := k.Observe(ctx, silent("ace")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
k.Close(context.Background())
|
|
|
|
again := NewKeeper(ctx, Options{Store: store, History: history, Looks: looks})
|
|
defer again.Close(context.Background())
|
|
open, err := again.Open(ctx)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(open) != 1 || open[0].Observations != 2 {
|
|
t.Fatalf("a new keeper read %+v", open)
|
|
}
|
|
e, _, err := store.Get(ctx, "machine.ace.silent")
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := store.Update(ctx, "machine.ace.silent", []byte(`{}`), e.Revision+7); err != ErrMoved {
|
|
t.Fatalf("a write at a stale revision answered %v", err)
|
|
}
|
|
if err := store.Create(ctx, "machine.ace.silent", []byte(`{}`)); err != ErrMoved {
|
|
t.Fatalf("creating an open condition answered %v", err)
|
|
}
|
|
if err := store.Delete(ctx, "machine.ace.silent", e.Revision+7); err != ErrMoved {
|
|
t.Fatalf("a delete at a stale revision answered %v", err)
|
|
}
|
|
if cleared, err := again.Clear(ctx, "machine.ace.silent", "heard"); err != nil || !cleared {
|
|
t.Fatalf("cleared %v: %v", cleared, err)
|
|
}
|
|
// Raised again at once: the store takes a key whose last word was a delete.
|
|
if _, err := again.Observe(ctx, silent("ace")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got, _, _ := again.Get(ctx, "machine.ace.silent")
|
|
if got.Count != 2 {
|
|
t.Fatalf("raised again after its clearing as %+v", got)
|
|
}
|
|
}
|
|
|
|
// **The history is read back from a moment, oldest first**, through the stream under its bucket.
|
|
func TestNatsTheHistoryIsReadByTime(t *testing.T) {
|
|
_, store, history := busStoreForTest(t)
|
|
ctx := t.Context()
|
|
k := NewKeeper(ctx, Options{Store: store, History: history})
|
|
if _, err := k.Observe(ctx, silent("ace")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := k.Clear(ctx, "machine.ace.silent", "heard"); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
k.Close(context.Background())
|
|
all, err := history.Since(ctx, time.Now().Add(-time.Hour))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(all) != 2 || all[0].Change != ChangeRaised || all[1].Change != ChangeCleared {
|
|
t.Fatalf("history %+v", all)
|
|
}
|
|
none, err := history.Since(ctx, time.Now().Add(time.Hour))
|
|
if err != nil || len(none) != 0 {
|
|
t.Fatalf("history from the future: %+v %v", none, err)
|
|
}
|
|
}
|