Review round 1: mesh-mail's real shape has no broker, rollout mint issues its processes, the acceptance test connects with the minted account (issue 1181)
mayIssue no longer requires a module's own broker secret when one of its processes declares bus: mesh-mail's real manifest has none, its tools served by the tool runner. rollout mint gained a KindProcess case, so a process's account rotates the same way every other bus account does. The release's acceptance test now composes its principals the real way (inventory.ProcessesOf -> broker.Users -> broker.ComposeAccounts), mints through the store as module issue does, and sets each connection's own inbox prefix so replies are not refused; it moved from internal/broker (which cannot import inventory) to cmd/mesh-controller. Added an integration test proving the mint, accept and vault hand-off, and that issuing again rotates the password.
This commit is contained in:
jochen
@@ -1,203 +0,0 @@
|
||||
//go:build integration
|
||||
|
||||
package broker
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats-server/v2/server"
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
"golang.org/x/crypto/bcrypt"
|
||||
)
|
||||
|
||||
// 1181/AC3, the release's acceptance test, on a real bus: the account module issue delivers for mesh-mail's
|
||||
// process "send" does exactly what its bus declares — pull and acknowledge its one work-queue consumer,
|
||||
// publish its event, read and write its state, and call operator-channel's tell in its own name — and
|
||||
// nothing it does not: no tool call or seat verb answered, no warrant published. Modelled on
|
||||
// callers_live_test.go's TestAServerComposedFromTheGrantsRefusesACallNamingAnother.
|
||||
//
|
||||
// go test -tags integration ./internal/broker/ -run TestAProcessAccountOnARealBus
|
||||
func TestAProcessAccountOnARealBus(t *testing.T) {
|
||||
hash, err := bcrypt.GenerateFromPassword([]byte("pw"), bcrypt.MinCost)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
controller := Principal{Kind: KindController, PasswordHash: string(hash)}
|
||||
messenger := Principal{Kind: KindModule, Node: "novox", Module: "messenger", PasswordHash: string(hash),
|
||||
Holds: []Seat{operatorChannel()}}
|
||||
process := Principal{Kind: KindProcess, Node: "novox", Module: "mesh-mail", Process: "send",
|
||||
PasswordHash: string(hash), Emits: []string{"mail.send.sent", "mail.send.failed"},
|
||||
Queues: []QueueOwner{{Process: "send", Stream: "MAIL_SENDS", MaxDeliver: 5}},
|
||||
State: []string{"sends"}, Asks: true, Uses: []Seat{operatorChannel()},
|
||||
Invokes: []string{"seat:operator-channel.tell"},
|
||||
}
|
||||
accounts, err := ComposeAccounts([]Principal{controller, messenger, process})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
conf := filepath.Join(t.TempDir(), "bus.conf")
|
||||
if err := os.WriteFile(conf, []byte("listen: 127.0.0.1:-1\njetstream { store_dir: "+
|
||||
`"`+t.TempDir()+`"`+" }\n"+accounts), 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
opts, err := server.ProcessConfigFile(conf)
|
||||
if err != nil {
|
||||
t.Fatalf("the composed accounts do not parse: %v", err)
|
||||
}
|
||||
opts.NoLog, opts.NoSigs = true, true
|
||||
s, err := server.NewServer(opts)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
go s.Start()
|
||||
if !s.ReadyForConnections(10 * time.Second) {
|
||||
t.Fatal("the bus did not come up")
|
||||
}
|
||||
defer s.Shutdown()
|
||||
|
||||
// The controller sets up the stream, the one consumer, and the process's state bucket — exactly as
|
||||
// `module issue` and the controller's raise would, before anything connects as the process.
|
||||
admin, err := Dial(s.ClientURL(), nats.UserInfo("controller", "pw"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer admin.Close()
|
||||
mailSends := ModuleStream{Module: "mesh-mail", Name: "MAIL_SENDS", Events: []string{"mail.send.approved"},
|
||||
MaxAgeDays: 1, MaxMiB: 8, WorkQueue: true, Duplicates: 1}
|
||||
if err := admin.EnsureStream(mailSends.Stream()); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := admin.EnsureConsumer(QueueConsumer(process, process.Queues[0])); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := admin.EnsureBucket(Bucket{Module: "mesh-mail", Name: "sends"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The router's own records bucket operator-channel names (novox/hq ADR 0259 §3): present so the
|
||||
// process's direct-get of its own module's asks is a real subject, not merely one it is permitted.
|
||||
js, err := jetstream.New(admin.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js.CreateKeyValue(t.Context(), jetstream.KeyValueConfig{Bucket: "messenger_asks"}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// messenger, the router, answers operator-channel's tell in the caller's own name.
|
||||
holder, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.messenger", "pw"))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer holder.Close()
|
||||
heard := make(chan string, 1)
|
||||
if _, err := holder.Subscribe("mesh.seat.operator-channel.tool.>", func(m *nats.Msg) {
|
||||
heard <- m.Subject
|
||||
_ = m.Respond([]byte(`{"result":{}}`))
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = holder.Flush()
|
||||
|
||||
refusals := make(chan error, 8)
|
||||
sender, err := nats.Connect(s.ClientURL(), nats.UserInfo("novox.mesh-mail.send", "pw"),
|
||||
nats.ErrorHandler(func(_ *nats.Conn, _ *nats.Subscription, err error) { refusals <- err }))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer sender.Close()
|
||||
|
||||
// 1. Its event.
|
||||
if err := sender.Publish("mesh.mod.mesh-mail.event.mail.send.sent", []byte("{}")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// 2. Its state: writes and reads its own bucket.
|
||||
kv, err := jetstream.New(sender)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
bucket, err := kv.KeyValue(t.Context(), "mesh-mail_sends")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := bucket.Put(t.Context(), "key", []byte("value")); err != nil {
|
||||
t.Fatalf("the process may not write its own state: %v", err)
|
||||
}
|
||||
if got, err := bucket.Get(t.Context(), "key"); err != nil || string(got.Value()) != "value" {
|
||||
t.Fatalf("the process may not read its own state back: %v, %v", got, err)
|
||||
}
|
||||
|
||||
// 3. Its queue: pulls and acknowledges MAIL_SENDS through its own consumer.
|
||||
js2, err := jetstream.New(admin.Conn())
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := js2.Publish(t.Context(), ModuleStreamSubject("mesh-mail", "mail.send.approved"), []byte("{}")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
reading, err := jetstream.New(sender)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
consumer, err := reading.Consumer(t.Context(), "MAIL_SENDS", QueueConsumer(process, process.Queues[0]).Name)
|
||||
if err != nil {
|
||||
t.Fatalf("the process may not bind its own queue consumer: %v", err)
|
||||
}
|
||||
batch, err := consumer.Fetch(1, jetstream.FetchMaxWait(3*time.Second))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
got := false
|
||||
for msg := range batch.Messages() {
|
||||
got = true
|
||||
if err := msg.Ack(); err != nil {
|
||||
t.Fatalf("the process may not acknowledge its own queue: %v", err)
|
||||
}
|
||||
}
|
||||
if !got {
|
||||
t.Fatal("the process did not pull the message its own queue holds")
|
||||
}
|
||||
|
||||
// 4. Its invoked seat verb, in its own name.
|
||||
if _, err := sender.Request("mesh.seat.operator-channel.tool.tell", []byte(`{}`), 3*time.Second); err != nil {
|
||||
t.Fatalf("the process may not call operator-channel.tell: %v", err)
|
||||
}
|
||||
select {
|
||||
case s := <-heard:
|
||||
if s != "mesh.seat.operator-channel.tool.tell" {
|
||||
t.Fatalf("messenger heard %s", s)
|
||||
}
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("messenger never heard the call")
|
||||
}
|
||||
|
||||
// 5. Refused: answering a tool call (it holds no seat, serves no tool).
|
||||
if _, err := sender.Subscribe("mesh.mod.mesh-mail.tool.>", func(*nats.Msg) {}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = sender.Flush()
|
||||
// 6. Refused: publishing a warrant (only the holder, messenger, may say operator-channel decided).
|
||||
if err := sender.Publish("mesh.seat.operator-channel.event.decided.mesh-mail", []byte("{}")); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
_ = sender.Flush()
|
||||
seen := 0
|
||||
timeout := time.After(3 * time.Second)
|
||||
for seen < 2 {
|
||||
select {
|
||||
case err := <-refusals:
|
||||
if !errors.Is(err, nats.ErrPermissionViolation) {
|
||||
t.Errorf("a refusal was not a permissions violation: %v", err)
|
||||
}
|
||||
seen++
|
||||
case <-timeout:
|
||||
t.Fatalf("saw %d of 2 expected permission violations", seen)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user