113 lines
4.3 KiB
Go
113 lines
4.3 KiB
Go
package bus_test
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"strings"
|
|
"testing"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/buscalls"
|
|
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
|
"git.novox.be/novox/mesh-sdk/go/membership"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"github.com/novox/mesh-tools/node-tools/internal/bus"
|
|
mt "github.com/novox/mesh-tools/node-tools/internal/meshtest"
|
|
)
|
|
|
|
// 1966/AC12: given a bundle whose membership binds objects `files`, when it puts and gets `a`, then it reads back what it
|
|
// put; a local name it was not bound is refused with the reason, and so is another module's binding and a binding of
|
|
// another storage type. A stream and a work queue bound alike answer their read-back, fill and size.
|
|
func TestABundleReachesItsOwnBoundObjectsAndStreamsOnly(t *testing.T) {
|
|
mesh := mt.New(t)
|
|
nc, err := nats.Connect(mt.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
js, _ := jetstream.New(nc)
|
|
ctx := context.Background()
|
|
if _, err := js.CreateObjectStore(ctx, jetstream.ObjectStoreConfig{Bucket: "m_files"}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "m_events", Subjects: []string{"mesh.state.m.events.>"}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, s := range []string{"mesh.state.m.events.one", "mesh.state.m.events.two"} {
|
|
if _, err := js.Publish(ctx, s, []byte(`{"n":1}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
m := mt.MembershipOf("m", "anchor", false, nil)
|
|
m.Bindings = []membership.BusBinding{
|
|
{Local: "files", Storage: catalogue.StorageObjectStore, Template: catalogue.TemplateObjects, ObjectStore: "m_files"},
|
|
{Local: "events", Storage: catalogue.StorageStream, Template: catalogue.TemplateEvents, Stream: "m_events",
|
|
Subject: "mesh.state.m.events"},
|
|
}
|
|
other := mt.MembershipOf("other", "anchor", false, nil)
|
|
mesh.Issue(t, m)
|
|
mesh.Issue(t, other)
|
|
conn, err := bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "anchor"})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
conn.Logf = func(string, ...any) {}
|
|
t.Cleanup(conn.Close)
|
|
conn.Follow("m")
|
|
conn.Follow("other")
|
|
mt.Until(t, func() error {
|
|
if conn.Membership("m") == nil || conn.Membership("other") == nil {
|
|
return errors.New("not issued yet")
|
|
}
|
|
return nil
|
|
})
|
|
|
|
if _, err := conn.ObjectsPut("m", buscalls.ObjectsPutRequest{Objects: "files", Name: "a", Data: []byte("what was put")}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
got, err := conn.ObjectsGet("m", buscalls.ObjectsNameRequest{Objects: "files", Name: "a"})
|
|
if err != nil || got == nil || string(got.Data) != "what was put" || got.Info.Name != "a" {
|
|
t.Fatalf("read back %+v, %v", got, err)
|
|
}
|
|
if listed, err := conn.ObjectsList("m", buscalls.ObjectsRequest{Objects: "files"}); err != nil || len(listed.Objects) != 1 {
|
|
t.Fatalf("listed %+v, %v", listed, err)
|
|
}
|
|
page, err := conn.StreamRead("m", buscalls.StreamReadRequest{Stream: "events", Limit: 1})
|
|
if err != nil || len(page.Messages) != 1 || page.Next != 2 || page.Last != 2 {
|
|
t.Fatalf("read %+v, %v", page, err)
|
|
}
|
|
if fill, err := conn.StreamFill("m", buscalls.StreamRequest{Stream: "events"}); err != nil || fill.Count != 2 {
|
|
t.Fatalf("fill %+v, %v", fill, err)
|
|
}
|
|
|
|
for _, c := range []struct {
|
|
name, module, refusal string
|
|
call func(module string) error
|
|
}{
|
|
{"a local name not bound", "m", `bound no object-store called "photos"`, func(module string) error {
|
|
_, err := conn.ObjectsGet(module, buscalls.ObjectsNameRequest{Objects: "photos", Name: "a"})
|
|
return err
|
|
}},
|
|
{"another module's binding", "other", `other was bound no object-store called "files"`, func(module string) error {
|
|
_, err := conn.ObjectsGet(module, buscalls.ObjectsNameRequest{Objects: "files", Name: "a"})
|
|
return err
|
|
}},
|
|
{"a binding of another storage type", "m", `"events" is bound with the storage type stream, not object-store`, func(module string) error {
|
|
_, err := conn.ObjectsList(module, buscalls.ObjectsRequest{Objects: "events"})
|
|
return err
|
|
}},
|
|
{"a work queue not bound", "m", `bound no stream called "jobs"`, func(module string) error {
|
|
_, err := conn.WorkQueueSize(module, buscalls.WorkQueueRequest{Queue: "jobs"})
|
|
return err
|
|
}},
|
|
} {
|
|
t.Run(c.name, func(t *testing.T) {
|
|
err := c.call(c.module)
|
|
if err == nil || !strings.Contains(err.Error(), c.refusal) {
|
|
t.Fatalf("refused with %v, want the reason %q", err, c.refusal)
|
|
}
|
|
})
|
|
}
|
|
}
|