package process import ( "context" "errors" "sort" "strings" "sync" "testing" "time" "github.com/nats-io/nats.go/jetstream" stdio "git.novox.be/novox/mesh-sdk/go" "git.novox.be/novox/mesh-sdk/go/asker" ) // fakeEntry is a fakeKV's KeyValueEntry: only Key, Value and Revision are read by State. type fakeEntry struct { jetstream.KeyValueEntry key string value []byte rev uint64 } func (e fakeEntry) Key() string { return e.key } func (e fakeEntry) Value() []byte { return e.value } func (e fakeEntry) Revision() uint64 { return e.rev } // fakeKV is an in-memory jetstream.KeyValue: embedding the nil interface satisfies every method this test // never calls, so only Get, Create and Update need a body (1188/AC2's "a fake KV"). type fakeKV struct { jetstream.KeyValue mu sync.Mutex values map[string][]byte revs map[string]uint64 next uint64 // initial is what a Watch hands first, before its nil marker; watcher is the last one made. initial []jetstream.KeyValueEntry watcher *fakeWatcher } func newFakeKV() *fakeKV { return &fakeKV{values: map[string][]byte{}, revs: map[string]uint64{}} } func (f *fakeKV) Get(_ context.Context, key string) (jetstream.KeyValueEntry, error) { f.mu.Lock() defer f.mu.Unlock() v, ok := f.values[key] if !ok { return nil, jetstream.ErrKeyNotFound } return fakeEntry{key: key, value: v, rev: f.revs[key]}, nil } func (f *fakeKV) Create(_ context.Context, key string, value []byte, _ ...jetstream.KVCreateOpt) (uint64, error) { f.mu.Lock() defer f.mu.Unlock() if _, ok := f.values[key]; ok { return 0, jetstream.ErrKeyExists } f.next++ f.values[key], f.revs[key] = value, f.next return f.next, nil } func (f *fakeKV) Update(_ context.Context, key string, value []byte, revision uint64) (uint64, error) { f.mu.Lock() defer f.mu.Unlock() if f.revs[key] != revision { return 0, jetstream.ErrKeyRevisionMismatch } f.next++ f.values[key], f.revs[key] = value, f.next return f.next, nil } func (f *fakeKV) Put(_ context.Context, key string, value []byte) (uint64, error) { f.mu.Lock() defer f.mu.Unlock() f.next++ f.values[key], f.revs[key] = value, f.next return f.next, nil } func (f *fakeKV) Delete(_ context.Context, key string, _ ...jetstream.KVDeleteOpt) error { f.mu.Lock() defer f.mu.Unlock() delete(f.values, key) delete(f.revs, key) return nil } func (f *fakeKV) Keys(_ context.Context, _ ...jetstream.WatchOpt) ([]string, error) { f.mu.Lock() defer f.mu.Unlock() if len(f.values) == 0 { return nil, jetstream.ErrNoKeysFound } keys := make([]string, 0, len(f.values)) for k := range f.values { keys = append(keys, k) } sort.Sort(sort.Reverse(sort.StringSlice(keys))) // as a bucket lists them: in no order the caller may rely on return keys, nil } // fakeWatcher is a jetstream.KeyWatcher fed by a test. type fakeWatcher struct { jetstream.KeyWatcher updates chan jetstream.KeyValueEntry pattern string } func (w *fakeWatcher) Updates() <-chan jetstream.KeyValueEntry { return w.updates } func (w *fakeWatcher) Stop() error { return nil } func (f *fakeKV) Watch(_ context.Context, pattern string, _ ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { f.mu.Lock() defer f.mu.Unlock() w := &fakeWatcher{updates: make(chan jetstream.KeyValueEntry, 16), pattern: pattern} f.watcher = w for _, e := range f.initial { w.updates <- e } w.updates <- nil return w, nil } func stateOn(kv jetstream.KeyValue) State { return State{Module: "mesh-mail", Name: "sends", kv: func(context.Context) (jetstream.KeyValue, error) { return kv, nil }} } // 1188/AC2: given the process connection's state type, when used as go/asker's Store, then it compiles and // behaves as the interface says, against a fake KV. func TestStateSatisfiesAskerStore(t *testing.T) { var _ asker.Store = State{} s := stateOn(newFakeKV()) if e, err := s.Get("ask.1"); err != nil || e != nil { t.Fatalf("Get of an absent key: %v, %v", e, err) } rev, err := s.Create("ask.1", map[string]any{"a": 1}) if err != nil { t.Fatalf("Create: %v", err) } e, err := s.Get("ask.1") if err != nil || e == nil || e.Revision != rev { t.Fatalf("Get after Create: %+v, %v", e, err) } if _, err := s.Create("ask.1", map[string]any{"a": 2}); !errors.Is(err, stdio.ErrChanged) { t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err) } rev2, err := s.Update("ask.1", map[string]any{"a": 3}, rev) if err != nil { t.Fatalf("Update at the revision read: %v", err) } if _, err := s.Update("ask.1", map[string]any{"a": 4}, rev); !errors.Is(err, stdio.ErrChanged) { t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err) } if rev2 == rev { t.Fatal("Update did not advance the revision") } } // 1840/S1: given a state, when a key is put twice, then each Put returns a new revision and the key reads // back the last value, whether or not it had one. func TestStatePutWritesAKeyWhole(t *testing.T) { s := stateOn(newFakeKV()) rev, err := s.Put("k", map[string]any{"a": 1}) if err != nil || rev == 0 { t.Fatalf("Put of a new key: %d, %v", rev, err) } rev2, err := s.Put("k", map[string]any{"a": 2}) if err != nil || rev2 <= rev { t.Fatalf("Put over a key: %d, %v, want a revision after %d", rev2, err, rev) } e, err := s.Get("k") if err != nil || e == nil || e.Revision != rev2 || string(e.Value) != `{"a":2}` { t.Fatalf("Get after Put: %+v, %v", e, err) } } // 1840/S1: given a state, when its keys are listed, then they come sorted, and none is an empty list, not an error. func TestStateKeysAreSortedAndEmptyWhenNone(t *testing.T) { s := stateOn(newFakeKV()) if keys, err := s.Keys(); err != nil || len(keys) != 0 { t.Fatalf("Keys of an empty state: %v, %v", keys, err) } for _, k := range []string{"b", "c", "a"} { if _, err := s.Put(k, 1); err != nil { t.Fatal(err) } } keys, err := s.Keys() if err != nil || strings.Join(keys, ",") != "a,b,c" { t.Fatalf("Keys: %v, %v, want a,b,c", keys, err) } } type watchEntry struct { fakeEntry op jetstream.KeyValueOp } func (e watchEntry) Operation() jetstream.KeyValueOp { return e.op } // 1840/S2: given a state with current values (one of them deleted), when it is watched, then the handler gets // each current value marked Current and not the deleted one, Watch returns, and a later put and delete arrive // in order unmarked. func TestStateWatchHandsCurrentValuesThenChanges(t *testing.T) { kv := newFakeKV() kv.initial = []jetstream.KeyValueEntry{ watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}, watchEntry{fakeEntry{key: "gone", rev: 2}, jetstream.KeyValueDelete}, watchEntry{fakeEntry{key: "b", value: []byte(`2`), rev: 3}, jetstream.KeyValuePut}, } got := make(chan stdio.StateChange, 8) s := stateOn(kv) if err := s.Watch("", func(c stdio.StateChange) error { got <- c; return nil }); err != nil { t.Fatal(err) } for _, want := range []stdio.StateChange{ {State: "sends", Key: "a", Op: "put", Value: []byte(`1`), Revision: 1, Current: true}, {State: "sends", Key: "b", Op: "put", Value: []byte(`2`), Revision: 3, Current: true}, } { select { case c := <-got: if c.Key != want.Key || c.Op != want.Op || string(c.Value) != string(want.Value) || c.Revision != want.Revision || c.Current != want.Current || c.State != want.State { t.Fatalf("got %+v, want %+v", c, want) } default: t.Fatalf("Watch returned before handling %+v", want) } } if len(got) != 0 { t.Fatalf("a deleted key was handed: %+v", <-got) } kv.watcher.updates <- watchEntry{fakeEntry{key: "a", value: []byte(`3`), rev: 4}, jetstream.KeyValuePut} kv.watcher.updates <- watchEntry{fakeEntry{key: "b", rev: 5}, jetstream.KeyValueDelete} for _, want := range []stdio.StateChange{ {Key: "a", Op: "put", Revision: 4}, {Key: "b", Op: "delete", Revision: 5}, } { select { case c := <-got: if c.Key != want.Key || c.Op != want.Op || c.Revision != want.Revision || c.Current { t.Fatalf("got %+v, want %+v", c, want) } case <-time.After(5 * time.Second): t.Fatalf("no change %+v", want) } } } // 1840/S2: a key narrowing reaches the bus as a subject pattern: `*` one name, `**` the rest, empty every key. func TestStateWatchNarrowing(t *testing.T) { for key, want := range map[string]string{"": ">", "a.b": "a.b", "a.*": "a.*", "a.**": "a.>", "**": ">"} { if got := watchPattern(key); got != want { t.Errorf("watchPattern(%q) = %q, want %q", key, got, want) } } } // 1840/S2: given a handler that refuses a current value, then Watch returns its error. func TestStateWatchReturnsAHandlersErrorOnCurrentValues(t *testing.T) { kv := newFakeKV() kv.initial = []jetstream.KeyValueEntry{watchEntry{fakeEntry{key: "a", value: []byte(`1`), rev: 1}, jetstream.KeyValuePut}} boom := errors.New("boom") if err := stateOn(kv).Watch("", func(stdio.StateChange) error { return boom }); !errors.Is(err, boom) { t.Fatalf("Watch: %v, want the handler's error", err) } } // 1840/S2: given a connection that closes before the current values arrive, then Watch returns an error. func TestStateWatchEndsWithAnErrorWhenTheConnectionCloses(t *testing.T) { s := State{Module: "mesh-mail", Name: "sends", closed: func() bool { return true }, kv: func(context.Context) (jetstream.KeyValue, error) { return silentKV{}, nil }} if err := s.Watch("", func(stdio.StateChange) error { return nil }); err == nil { t.Fatal("Watch on a closed connection returned no error") } } // silentKV watches but never delivers. type silentKV struct{ jetstream.KeyValue } func (silentKV) Watch(context.Context, string, ...jetstream.WatchOpt) (jetstream.KeyWatcher, error) { return &fakeWatcher{updates: make(chan jetstream.KeyValueEntry)}, nil } // 1840/S1: given a key, when it is deleted, then it reads as absent; deleting a key that was not there is no error. func TestStateDeleteRemovesAKeyAndAnAbsentOneIsNoError(t *testing.T) { s := stateOn(newFakeKV()) if _, err := s.Put("k", 1); err != nil { t.Fatal(err) } if err := s.Delete("k"); err != nil { t.Fatalf("Delete of a key: %v", err) } if e, err := s.Get("k"); err != nil || e != nil { t.Fatalf("Get after Delete: %+v, %v, want absent", e, err) } if err := s.Delete("never"); err != nil { t.Fatalf("Delete of an absent key: %v", err) } }