Let a process read and change its module's state by compare-and-set, and emit with a chosen message id (issue 1188)
This commit is contained in:
jochen
@@ -0,0 +1,69 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// Emit publishes an event as this process's module (novox/hq ADR 0436 clause 4), under its module's own
|
||||
// namespace (mesh.mod.<module>.event.<type>, the same subject the module's own principal would publish it
|
||||
// on), with the SDK's own envelope and headers (stdio.go:498-509). A chosen message id
|
||||
// (EmitOptions.MessageID) is sent as x-event-id, and as the bus's own dedup id (Nats-Msg-Id), so a second
|
||||
// publish inside the stream's duplicate window is dropped; empty is a random one, as stdio.Emit on main.
|
||||
func (c *Conn) Emit(eventType string, body any, opts ...stdio.EmitOptions) error {
|
||||
data, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
id := ""
|
||||
for _, o := range opts {
|
||||
if o.MessageID != "" {
|
||||
id = o.MessageID
|
||||
}
|
||||
}
|
||||
if id == "" {
|
||||
id = eventID()
|
||||
}
|
||||
msg := nats.NewMsg("mesh.mod." + c.account.Module + ".event." + eventType)
|
||||
msg.Header.Set("x-event-id", id)
|
||||
msg.Header.Set("x-source", c.account.Module)
|
||||
msg.Header.Set("x-node", c.account.Node)
|
||||
msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
|
||||
msg.Header.Set("content-type", "application/json")
|
||||
for _, o := range opts {
|
||||
if o.CausationID != "" {
|
||||
msg.Header.Set("x-causation-id", o.CausationID)
|
||||
}
|
||||
if o.Schema != "" {
|
||||
msg.Header.Set("x-schema", o.Schema)
|
||||
}
|
||||
}
|
||||
msg.Data = data
|
||||
js, err := jetstream.New(c.nc)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
||||
defer cancel()
|
||||
_, err = js.PublishMsg(ctx, msg, jetstream.WithMsgID(id))
|
||||
return err
|
||||
}
|
||||
|
||||
// eventID is a random UUID, version 4, as stdio's own eventID (go/stdio.go:515-522) — duplicated rather
|
||||
// than imported, since stdio's is unexported and this package touches none of stdio's bundle machinery.
|
||||
func eventID() string {
|
||||
var b [16]byte
|
||||
_, _ = rand.Read(b[:])
|
||||
b[6] = (b[6] & 0x0f) | 0x40
|
||||
b[8] = (b[8] & 0x3f) | 0x80
|
||||
h := hex.EncodeToString(b[:])
|
||||
return h[0:8] + "-" + h[8:12] + "-" + h[12:16] + "-" + h[16:20] + "-" + h[20:32]
|
||||
}
|
||||
@@ -0,0 +1,57 @@
|
||||
//go:build integration
|
||||
|
||||
package process
|
||||
|
||||
import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// 1188/AC3: given a stream accepting mail.send.sent, when the process emits it twice with the same
|
||||
// message id inside the duplicate window, then the stream holds one message; with the headers stdio's
|
||||
// Emit sets (stdio.go:498-509).
|
||||
func TestEmitTwiceWithTheSameIDIsKeptToOneWithTheSDKsHeaders(t *testing.T) {
|
||||
url, account := grantedSendProcess(t)
|
||||
account.Node = "novox"
|
||||
_, adminJetStream := adminJS(t, url)
|
||||
stream, err := adminJetStream.CreateStream(ctxFor(t), jetstream.StreamConfig{
|
||||
Name: "MAIL_EVENTS", Subjects: []string{"mesh.mod.mesh-mail.event.>"}, Duplicates: time.Minute,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("the admin could not make the events stream: %v", err)
|
||||
}
|
||||
|
||||
conn, err := Connect(account)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
for i := 0; i < 2; i++ {
|
||||
if err := conn.Emit("mail.send.sent", map[string]any{"send": "s1"}, stdio.EmitOptions{MessageID: "send-s1"}); err != nil {
|
||||
t.Fatalf("emit %d: %v", i, err)
|
||||
}
|
||||
}
|
||||
|
||||
info, err := stream.Info(ctxFor(t))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if info.State.Msgs != 1 {
|
||||
t.Fatalf("the stream holds %d messages, want exactly 1 (the second publish is a duplicate)", info.State.Msgs)
|
||||
}
|
||||
|
||||
msg, err := stream.GetMsg(ctxFor(t), 1)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if msg.Header.Get("x-event-id") != "send-s1" || msg.Header.Get("x-source") != "mesh-mail" ||
|
||||
msg.Header.Get("x-node") != "novox" || msg.Header.Get("content-type") != "application/json" ||
|
||||
msg.Header.Get("x-time") == "" {
|
||||
t.Errorf("the stored message's headers are not the SDK's: %v", msg.Header)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,100 @@
|
||||
//go:build integration
|
||||
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
)
|
||||
|
||||
// sendUser and sendPassword are mesh-mail's process "send" on novox, as mesh-controller would mint it
|
||||
// (mesh-controller internal/broker/process_grants_test.go mailSenderProcess, Principal.Username).
|
||||
const (
|
||||
sendUser = "novox.mesh-mail.send"
|
||||
sendPassword = "the-pw"
|
||||
adminUser = "admin"
|
||||
adminPass = "admin-pw"
|
||||
)
|
||||
|
||||
// sendGrants is the process principal mailSenderProcess's whole publish set, copied from
|
||||
// mesh-controller internal/broker/process_grants_test.go (TestAProcessIsGrantedExactlyItsBusDeclaration,
|
||||
// "dumped from a passing run... 16 subjects, correct today"), since this package may not import the
|
||||
// controller: the exact grant the controller composes for novox.mesh-mail.send.
|
||||
var sendGrants = []string{
|
||||
"$JS.ACK.MAIL_SENDS.novox_mesh-mail_send.>",
|
||||
"$JS.API.CONSUMER.CREATE.KV_mesh-mail_sends.>",
|
||||
"$JS.API.CONSUMER.DELETE.KV_mesh-mail_sends.>",
|
||||
"$JS.API.CONSUMER.INFO.MAIL_SENDS.novox_mesh-mail_send",
|
||||
"$JS.API.CONSUMER.MSG.NEXT.MAIL_SENDS.novox_mesh-mail_send",
|
||||
"$JS.API.DIRECT.GET.KV_mesh-mail_sends.>",
|
||||
"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.mesh-mail.>",
|
||||
"$JS.API.STREAM.INFO.KV_mesh-mail_sends",
|
||||
"$JS.FC.KV_mesh-mail_sends.>",
|
||||
"$KV.mesh-mail_sends.>",
|
||||
"mesh.mod.mesh-mail.event.mail.send.failed",
|
||||
"mesh.mod.mesh-mail.event.mail.send.sent",
|
||||
"mesh.seat.operator-channel.call.tell.*.novox~mesh-mail~send",
|
||||
"mesh.seat.operator-channel.call.tell.novox~mesh-mail~send",
|
||||
"mesh.seat.operator-channel.tool.tell",
|
||||
"mesh.seat.operator-channel.tool.tell.*",
|
||||
}
|
||||
|
||||
// grantedSendProcess is a test bus with JetStream, with two users: novox.mesh-mail.send, permissioned to
|
||||
// exactly sendGrants and its own inbox (nats-server server.Options Users with Permissions, as
|
||||
// mesh-controller's cmd/mesh-controller/process_live_test.go composes accounts for a real bus — copied as
|
||||
// literal grants here since this package may not import the controller); and an unrestricted admin for the
|
||||
// test's own setup (creating streams and buckets, standing in for the holder and the runtime).
|
||||
func grantedSendProcess(t *testing.T) (url string, account Account) {
|
||||
t.Helper()
|
||||
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
|
||||
NoLog: true, NoSigs: true,
|
||||
Users: []*server.User{
|
||||
{Username: adminUser, Password: adminPass},
|
||||
{Username: sendUser, Password: sendPassword, Permissions: &server.Permissions{
|
||||
Publish: &server.SubjectPermission{Allow: sendGrants},
|
||||
Subscribe: &server.SubjectPermission{Allow: []string{"_INBOX." + sendUser + ".>"}},
|
||||
}},
|
||||
},
|
||||
}
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatalf("a granted test bus could not be made: %v", err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(10 * time.Second) {
|
||||
s.Shutdown()
|
||||
t.Fatal("a granted test bus did not come up within 10s")
|
||||
}
|
||||
t.Cleanup(func() {
|
||||
s.Shutdown()
|
||||
s.WaitForShutdown()
|
||||
})
|
||||
url = s.ClientURL()
|
||||
return url, Account{URL: url, Node: "novox", Module: "mesh-mail", User: sendUser, Password: sendPassword}
|
||||
}
|
||||
|
||||
// adminJS is the test's own unrestricted connection to url, for setting up fixtures.
|
||||
func adminJS(t *testing.T, url string) (*nats.Conn, jetstream.JetStream) {
|
||||
t.Helper()
|
||||
nc, err := nats.Connect(url, nats.UserInfo(adminUser, adminPass), nats.CustomInboxPrefix("_INBOX."+adminUser))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
t.Cleanup(nc.Close)
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return nc, js
|
||||
}
|
||||
|
||||
func ctxFor(t *testing.T) context.Context {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
||||
t.Cleanup(cancel)
|
||||
return ctx
|
||||
}
|
||||
@@ -0,0 +1,117 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// StateTimeout bounds a state call (Get, Create, Update) and an Emit (emit.go): a var, not a const, so a
|
||||
// test whose account is not granted a subject — refused only as an async protocol error with no reply,
|
||||
// never a reply naming the refusal — need not wait out the full bound (mesh-tools' own ProofTimeout,
|
||||
// node-tools/internal/bus/seat.go:65, is a var for the same reason).
|
||||
var StateTimeout = 30 * time.Second
|
||||
|
||||
// State is one of this process's module's states, read and written on this connection (novox/hq ADR 0436
|
||||
// clause 4), in the bucket <module>_<name> (mesh-controller internal/broker/state.go:49) — only as the
|
||||
// account's grants allow. It satisfies go/asker's Store.
|
||||
type State struct {
|
||||
Module string
|
||||
Name string
|
||||
// kv binds this state's underlying key-value store: a live connection's own (Conn.State), or, in a
|
||||
// unit test, a fake that never reaches a bus.
|
||||
kv func(ctx context.Context) (jetstream.KeyValue, error)
|
||||
}
|
||||
|
||||
// bucket is this state's name on the bus.
|
||||
func (s State) bucket() string { return s.Module + "_" + s.Name }
|
||||
|
||||
// State is the named state this connection's module keeps, bound lazily to its bucket on each call so a
|
||||
// bucket the account is not granted fails at the call that needed it, naming the bucket.
|
||||
func (c *Conn) State(name string) State {
|
||||
return State{Module: c.account.Module, Name: name, kv: func(ctx context.Context) (jetstream.KeyValue, error) {
|
||||
js, err := jetstream.New(c.nc)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return js.KeyValue(ctx, c.account.Module+"_"+name)
|
||||
}}
|
||||
}
|
||||
|
||||
func (s State) store(ctx context.Context) (jetstream.KeyValue, error) {
|
||||
kv, err := s.kv(ctx)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s's state %s (bucket %s): %w", s.Module, s.Name, s.bucket(), err)
|
||||
}
|
||||
return kv, nil
|
||||
}
|
||||
|
||||
// Get is a key's current value, or nil when it has none.
|
||||
func (s State) Get(key string) (*stdio.StateEntry, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
entry, err := kv.Get(ctx, key)
|
||||
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
||||
return nil, nil
|
||||
}
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
||||
}
|
||||
return &stdio.StateEntry{Key: key, Value: json.RawMessage(entry.Value()), Revision: entry.Revision()}, nil
|
||||
}
|
||||
|
||||
// Create writes a key only when it has none, and returns the revision; a key that has one is refused as
|
||||
// stdio.ErrChanged.
|
||||
func (s State) Create(key string, value any) (uint64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
data, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Create(ctx, key, data)
|
||||
if errors.Is(err, jetstream.ErrKeyExists) {
|
||||
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
||||
}
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
||||
}
|
||||
return rev, nil
|
||||
}
|
||||
|
||||
// Update writes a key only when it is still at revision, and returns the new one; a key changed meanwhile
|
||||
// is refused as stdio.ErrChanged.
|
||||
func (s State) Update(key string, value any, revision uint64) (uint64, error) {
|
||||
ctx, cancel := context.WithTimeout(context.Background(), StateTimeout)
|
||||
defer cancel()
|
||||
kv, err := s.store(ctx)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
data, err := json.Marshal(value)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
rev, err := kv.Update(ctx, key, data, revision)
|
||||
if errors.Is(err, jetstream.ErrKeyRevisionMismatch) {
|
||||
return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
||||
}
|
||||
if err != nil {
|
||||
return 0, fmt.Errorf("%s's state %s, key %s: %w", s.Module, s.Name, key, err)
|
||||
}
|
||||
return rev, nil
|
||||
}
|
||||
@@ -0,0 +1,88 @@
|
||||
//go:build integration
|
||||
|
||||
package process
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
stdio "git.novox.be/novox/mesh-sdk/go"
|
||||
)
|
||||
|
||||
// 1188/AC1: given a process connection to a test bus granted the bucket mesh-mail_sends, when it creates a
|
||||
// key, reads it, and updates it with the revision it read, then each succeeds and returns the new
|
||||
// revision; an update with a stale revision fails as stdio.ErrChanged; a create of an existing key fails
|
||||
// as stdio.ErrChanged.
|
||||
func TestStateCreateGetUpdateOnAGrantedBucket(t *testing.T) {
|
||||
url, account := grantedSendProcess(t)
|
||||
_, adminJetStream := adminJS(t, url)
|
||||
if _, err := adminJetStream.CreateKeyValue(ctxFor(t), jetstream.KeyValueConfig{Bucket: "mesh-mail_sends"}); err != nil {
|
||||
t.Fatalf("the admin could not make the bucket: %v", err)
|
||||
}
|
||||
|
||||
conn, err := Connect(account)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
s := conn.State("sends")
|
||||
|
||||
rev, err := s.Create("key", map[string]any{"v": 1})
|
||||
if err != nil {
|
||||
t.Fatalf("Create on a granted bucket: %v", err)
|
||||
}
|
||||
if rev == 0 {
|
||||
t.Fatal("Create returned no revision")
|
||||
}
|
||||
|
||||
e, err := s.Get("key")
|
||||
if err != nil || e == nil {
|
||||
t.Fatalf("Get after Create: %+v, %v", e, err)
|
||||
}
|
||||
if e.Revision != rev || string(e.Value) != `{"v":1}` {
|
||||
t.Fatalf("Get read back %+v, want revision %d and {\"v\":1}", e, rev)
|
||||
}
|
||||
|
||||
rev2, err := s.Update("key", map[string]any{"v": 2}, rev)
|
||||
if err != nil {
|
||||
t.Fatalf("Update at the revision read: %v", err)
|
||||
}
|
||||
if rev2 == rev {
|
||||
t.Fatal("Update did not return a new revision")
|
||||
}
|
||||
|
||||
if _, err := s.Update("key", map[string]any{"v": 3}, rev); !errors.Is(err, stdio.ErrChanged) {
|
||||
t.Fatalf("Update at a stale revision: %v, want stdio.ErrChanged", err)
|
||||
}
|
||||
|
||||
if _, err := s.Create("key", map[string]any{"v": 4}); !errors.Is(err, stdio.ErrChanged) {
|
||||
t.Fatalf("Create of an existing key: %v, want stdio.ErrChanged", err)
|
||||
}
|
||||
}
|
||||
|
||||
// 1188/AC4: given an account not granted a bucket, when it writes there, then the bus's refusal comes back
|
||||
// as an error naming the bucket. A denied publish is refused only as an async protocol error with no
|
||||
// reply (never a reply naming the refusal), so StateTimeout is shortened for this test alone.
|
||||
func TestStateRefusesAnUngrantedBucketNamingIt(t *testing.T) {
|
||||
old := StateTimeout
|
||||
StateTimeout = 2 * time.Second
|
||||
t.Cleanup(func() { StateTimeout = old })
|
||||
|
||||
_, account := grantedSendProcess(t)
|
||||
conn, err := Connect(account)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer conn.Close()
|
||||
|
||||
s := conn.State("unrelated")
|
||||
if _, err := s.Create("key", map[string]any{"v": 1}); err == nil {
|
||||
t.Fatal("writing to an ungranted bucket was not refused")
|
||||
} else if !strings.Contains(err.Error(), "mesh-mail_unrelated") {
|
||||
t.Errorf("the refusal %q does not name the bucket mesh-mail_unrelated", err.Error())
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,105 @@
|
||||
package process
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"sync"
|
||||
"testing"
|
||||
|
||||
"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
|
||||
}
|
||||
|
||||
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 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")
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user