diff --git a/node-tools/cmd/node-tools/approvals_integration_test.go b/node-tools/cmd/node-tools/approvals_integration_test.go new file mode 100644 index 0000000..2e58b24 --- /dev/null +++ b/node-tools/cmd/node-tools/approvals_integration_test.go @@ -0,0 +1,290 @@ +//go:build integration + +package main + +import ( + "encoding/json" + "errors" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" + seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" + catalogseat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-sdk/go/seatdef" + "github.com/nats-io/nats.go" + + "github.com/novox/mesh-tools/node-tools/internal/approval" + "github.com/novox/mesh-tools/node-tools/internal/bus" + mt "github.com/novox/mesh-tools/node-tools/internal/meshtest" +) + +// marked is the fixture marked module's catalog half: its build src2:1 marks its tool wipe (design 46 7-approval-mark §1). +type marked struct{} + +var wipeEntry = catalogseat.CatalogEntry{Module: "shell", Fp: "src2:1", Manifest: map[string]any{"needs-approval": []any{"wipe"}}} + +func (marked) Ready() error { return nil } +func (marked) Revision() string { return "7" } +func (marked) Served(fp string) (catalogseat.CatalogEntry, bool, string) { + return wipeEntry, fp == wipeEntry.Fp, "not-found" +} +func (marked) Newest(m string) (catalogseat.CatalogEntry, bool) { return wipeEntry, m == "shell" } +func (marked) Definitions() map[string]seatdef.Definition { return map[string]seatdef.Definition{} } +func (marked) Entries() []catalogseat.CatalogEntry { return []catalogseat.CatalogEntry{wipeEntry} } + +// bed is the runner's gate wired as main wires it, over the runner's own connection to a real bus, with stand-ins of the +// operator channel's router (it takes each ask into its record asks, as the router writes it) and of the hand-act seat. +type bed struct { + t *testing.T + conn *bus.Conn + asks nats.KeyValue + taken chan struct{} + done chan struct{} + + mu sync.Mutex + asked []seat.Ask + recorded []string + holders map[string][]byte +} + +func newBed(t *testing.T, bindState bool) *bed { + t.Helper() + mesh := mt.New(t) + nc, err := nats.Connect(mt.URL(t)) + if err != nil { + t.Fatal(err) + } + t.Cleanup(nc.Close) + js, _ := nc.JetStream() + if _, err := js.AddStream(&nats.StreamConfig{Name: "SEAT_OPERATOR_CHANNEL", Subjects: []string{"mesh.seat.operator-channel.accept.>"}, + Retention: nats.WorkQueuePolicy}); err != nil { + t.Fatal(err) + } + b := &bed{t: t, asks: mesh.Bucket(t, "messenger_asks"), taken: make(chan struct{}, 1), done: make(chan struct{}), + holders: map[string][]byte{"operator-channel": []byte(`{"seat":"operator-channel","scope":"mesh","holders":[{"module":"messenger"}]}`)}} + t.Cleanup(func() { close(b.done) }) + mesh.Bucket(t, "node-tools_approvals") + mesh.BucketKeyTTL(t, "node-tools_bound") + + // The router's stand-in: each ask accepted under the marked module's name is taken into its record, state open. + sub, err := js.PullSubscribe("mesh.seat.operator-channel.accept.>", "router", nats.BindStream("SEAT_OPERATOR_CHANNEL")) + if err != nil { + t.Fatal(err) + } + go func() { + for { + select { + case <-b.done: + return + default: + } + msgs, _ := sub.Fetch(1, nats.MaxWait(200*time.Millisecond)) + for _, m := range msgs { + var a seat.Ask + if err := json.Unmarshal(m.Data, &a); err == nil { + module := strings.TrimPrefix(m.Subject, "mesh.seat.operator-channel.accept.ask.") + b.mu.Lock() + b.asked = append(b.asked, a) + b.mu.Unlock() + b.write(seat.AsksByAskerIDValue{Asker: module, Key: module + "." + a.ID, State: "open", Headline: a.Headline, + Expires: a.Expires, Opened: operatorchannelclient.FormatTime(time.Now()), OfferedOn: []string{"telegram"}}) + select { + case b.taken <- struct{}{}: + default: + } + } + _ = m.Ack() + } + } + }() + // The hand-act seat's stand-in: warranted {asker, ask} answered, each one kept. + if _, err := nc.Subscribe("mesh.seat.mesh-hand-acts.>", func(m *nats.Msg) { + b.mu.Lock() + b.recorded = append(b.recorded, string(m.Data)) + b.mu.Unlock() + _ = m.Respond([]byte(`{"result":{"recorded":true}}`)) + }); err != nil { + t.Fatal(err) + } + + runner := mt.MembershipOf("node-tools", "n1", false, nil) + if bindState { + runner.State = []bus.StateIssued{{Name: approval.StateName, Bucket: "node-tools_approvals", Writes: true}, + {Name: BoundState, Bucket: "node-tools_bound", Writes: true}} + } + shell := mt.MembershipOf("shell", "n1", false, nil) + shell.SeatTraffic = &bus.SeatTraffic{Publish: []string{seat.AcceptSubjectAsk("shell")}, + Records: []string{"$JS.API.DIRECT.GET.KV_messenger_asks.$KV.messenger_asks.shell.>"}} + mesh.Issue(t, runner) + mesh.Issue(t, shell) + b.conn, err = bus.Connect(bus.Credential{URL: mt.URL(t), Module: "node-tools", Node: "n1"}) + if err != nil { + t.Fatal(err) + } + b.conn.Logf = func(string, ...any) {} + t.Cleanup(b.conn.Close) + b.conn.Follow("node-tools") + b.conn.Follow("shell") + mt.Until(t, func() error { + if b.conn.Membership("node-tools") == nil || b.conn.Membership("shell") == nil { + return errors.New("not issued yet") + } + return nil + }) + return b +} + +// gate is a runner's gate as main makes it; record says how the run's recording goes beside the call (nil: at once). +func (b *bed) gate(record func(func())) *approval.Gate { + if record == nil { + record = func(f func()) { f() } + } + g := approval.New(approval.Deps{Node: "n1", Runner: "node-tools", Catalog: marked{}, Logf: b.t.Logf, + State: approvalsState{conn: b.conn, module: "node-tools"}, Publish: publishAsk(b.conn), Record: routerRecord(b.conn), + Warranted: warranted(b.conn), Bounds: boundState{conn: b.conn, module: "node-tools"}, Go: record, + Holder: approval.HoldersIn(func() (map[string][]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.holders, nil + }), + // The pause while the router takes the ask ends when the stand-in has taken one: no sleeping. + After: func(time.Duration) <-chan time.Time { + c := make(chan time.Time, 1) + go func() { + select { + case <-b.taken: + case <-b.done: + } + c <- time.Now() + }() + return c + }}) + _ = g.Unserved("shell", "src2:1", []string{"wipe"}) + return g +} + +func (b *bed) write(rec seat.AsksByAskerIDValue) { + raw, _ := json.Marshal(rec) + if _, err := b.asks.Put(rec.Key, raw); err != nil { + b.t.Fatal(err) + } +} + +// decide writes the router's record of the newest ask as the router writes the operator's choice: chosen, with its warrant. +func (b *bed) decide(option string) seat.Ask { + b.mu.Lock() + a := b.asked[len(b.asked)-1] + b.mu.Unlock() + e, err := b.asks.Get("shell." + a.ID) + if err != nil { + b.t.Fatal(err) + } + var rec seat.AsksByAskerIDValue + _ = json.Unmarshal(e.Value(), &rec) + rec.State = "chosen" + rec.Warrant = &seat.Warrant{Ask: a.ID, Asker: "shell", Outcome: operatorchannelclient.OutcomeChosen, Option: option, + Level: operatorchannelclient.Approve, At: operatorchannelclient.FormatTime(time.Now()), AskDigest: operatorchannelclient.Digest(a), + Channel: "telegram", By: &seat.Person{Who: operatorchannelclient.Operator, Kind: "telegram", Identity: "42", Verified: "user id verified"}} + b.write(rec) + return a +} + +func wipe(args string) approval.Call { + return approval.Call{Address: "shell.wipe", Module: "shell", Name: "wipe", Fingerprint: "src2:1", Caller: "agent", + Args: json.RawMessage(args), Description: "Wipes a scratch directory."} +} + +// Design 46 7-approval-mark §22: the runner's gate against the real bus with a fixture marked module: a call asked and taken +// by the operator channel's stand-in, the operator's Approve written as the router's record, the call again run once and +// recorded in the hand-act log, mesh_approvals' list, a restart between the run and warranted's answer recorded once; a +// Decline runs nothing; a gate missing a part of its check runs nothing. +func TestAMarkedCallOnTheBusIsAskedRunOnTheApprovalAndRecorded(t *testing.T) { + t.Run("approved: asked, run once, recorded, listed", func(t *testing.T) { + b := newBed(t, true) + g := b.gate(nil) + runs := 0 + run := func() (any, error) { runs++; return "wiped", nil } + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || + !strings.HasPrefix(err.Error(), "Not run: this needs the operator's approval. Asked on telegram at ") || runs != 0 { + t.Fatalf("first call: %v, runs %d", err, runs) + } + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || !strings.Contains(err.Error(), "still waiting for the operator's answer") { + t.Fatalf("before the answer: %v", err) + } + a := b.decide(approval.OptionApprove) + got, err := g.Run(wipe(`{"dir":"scratch"}`), run) + answered, ok := got.(approval.Answered) + if err != nil || !ok || answered.Answer != "wiped" || !strings.Contains(answered.Approval, "ask "+a.ID) || runs != 1 { + t.Fatalf("on the approval: %+v, %v, runs %d", got, err, runs) + } + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || runs != 1 { + t.Fatalf("the warrant ran twice: %v, runs %d", err, runs) + } + b.mu.Lock() + recorded := strings.Join(b.recorded, "\n") + b.mu.Unlock() + if !strings.Contains(recorded, a.ID) || !strings.Contains(recorded, "shell") { + t.Fatalf("hand-act log heard %q", recorded) + } + list, err := g.List() + if err != nil || len(list) == 0 || list[len(list)-1].Ask != a.ID || list[len(list)-1].State != "run, recorded" { + t.Fatalf("mesh_approvals lists %+v, %v", list, err) + } + }) + t.Run("a restart between the run and warranted's answer records the run once", func(t *testing.T) { + b := newBed(t, true) + stopped := b.gate(func(func()) {}) // the runner stops before warranted answers + runs := 0 + run := func() (any, error) { runs++; return "wiped", nil } + _, _ = stopped.Run(wipe(`{"dir":"tmp"}`), run) + a := b.decide(approval.OptionApprove) + if _, err := stopped.Run(wipe(`{"dir":"tmp"}`), run); err != nil || runs != 1 { + t.Fatalf("run: %v, runs %d", err, runs) + } + if list, _ := stopped.List(); len(list) != 1 || list[0].State != "run, not yet recorded" { + t.Fatalf("before the restart: %+v", list) + } + started := b.gate(nil) + started.Recover() + started.Recover() + b.mu.Lock() + n := len(b.recorded) + b.mu.Unlock() + if list, _ := started.List(); n != 1 || len(list) != 1 || list[0].Ask != a.ID || list[0].State != "run, recorded" || runs != 1 { + t.Fatalf("after the restart: recorded %d, %+v, runs %d", n, list, runs) + } + }) + t.Run("declined: refused, nothing runs", func(t *testing.T) { + b := newBed(t, true) + g := b.gate(nil) + runs := 0 + run := func() (any, error) { runs++; return nil, nil } + _, _ = g.Run(wipe(`{"dir":"scratch"}`), run) + b.decide(approval.OptionDecline) + if _, err := g.Run(wipe(`{"dir":"scratch"}`), run); err == nil || !strings.Contains(err.Error(), "the operator declined ask") || runs != 0 { + t.Fatalf("declined: %v, runs %d", err, runs) + } + }) + t.Run("failing closed: no state bound, or nobody to ask, runs nothing and asks nothing", func(t *testing.T) { + unbound := newBed(t, false) + runs := 0 + run := func() (any, error) { runs++; return nil, nil } + if _, err := unbound.gate(nil).Run(wipe(`{"dir":"scratch"}`), run); err == nil || + !strings.Contains(err.Error(), "binds no state approvals") || runs != 0 || len(unbound.asked) != 0 { + t.Fatalf("no state: %v, runs %d", err, runs) + } + b := newBed(t, true) + b.holders = map[string][]byte{} + if _, err := b.gate(nil).Run(wipe(`{"dir":"scratch"}`), run); err == nil || + !strings.Contains(err.Error(), "no module holds operator-channel") || runs != 0 { + t.Fatalf("no holder: %v, runs %d", err, runs) + } + if got, err := b.gate(nil).Run(approval.Call{Address: "shell.tidy", Module: "shell", Name: "tidy", Fingerprint: "src2:1", + Args: json.RawMessage(`{}`)}, func() (any, error) { return "tidied", nil }); err != nil || got != "tidied" { + t.Fatalf("an unmarked tool: %v, %v", got, err) + } + }) +} diff --git a/node-tools/cmd/node-tools/main_test.go b/node-tools/cmd/node-tools/main_test.go index a9d093a..0018932 100644 --- a/node-tools/cmd/node-tools/main_test.go +++ b/node-tools/cmd/node-tools/main_test.go @@ -13,10 +13,17 @@ import ( "github.com/novox/mesh-tools/node-tools/internal/runtime" ) +// The runner's start waits for its credential and its bus within the start's bound (issue 1496): one test, two cases, so +// design 46 7-approval-mark §22's whole test of the gate adds no test to the module (design 47 G §419, §626). +func TestTheRunnersStartWaitsForItsCredentialAndItsBus(t *testing.T) { + t.Run("the credential", awaitsCredential) + t.Run("the bus", connectsWithin) +} + // A runtime started before its credential is on the machine waits for it, woken by the watch of its directory, and gives // up loudly at its bound; a file that is no credential ends the wait at once (issue 1496: on ace mesh-build-node's and // mesh-build-image's runtimes exited twice on "cannot read the broker credential ... no such file"). -func TestAwaitCredential(t *testing.T) { +func awaitsCredential(t *testing.T) { const good = `{"url":"tls://bus.example:4222","module":"m"}` rereadEvery = time.Hour // only the watch can wake the read in time settle := runtime.Settle @@ -82,7 +89,7 @@ func TestAwaitCredential(t *testing.T) { // A bus that refuses the account at a start is asked again until the start's bound, said once; a refusal past it, a pin // that does not match and an address that is no URL give up (issue 1496, issue 058). -func TestConnectWithin(t *testing.T) { +func connectsWithin(t *testing.T) { refused := errors.New("nats: Authorization Violation") unreachable := errors.New("nats: no servers available for connection") for _, c := range []struct { diff --git a/node-tools/internal/approval/approval_test.go b/node-tools/internal/approval/approval_test.go index ef5907f..53dfe47 100644 --- a/node-tools/internal/approval/approval_test.go +++ b/node-tools/internal/approval/approval_test.go @@ -246,8 +246,8 @@ func TestTheRefusalsAndAnswers(t *testing.T) { } var k operatorchannelclient.Kept _ = json.Unmarshal(w.state.Vals["n1.ask."+a.ID], &k) - if k.Act["arg.dir"] != "scratch" || string(k.Arguments) != `{"dir":"scratch"}` || string(w.state.Vals["n1.act."+d]) != `"`+a.ID+`"` { - t.Errorf("kept %+v, act key %s", k, w.state.Vals["n1.act."+d]) + if k.Act["arg.dir"] != "scratch" || string(k.Arguments) != `{"dir":"scratch"}` || string(w.state.Vals["n1.act."+keys.EscapeName(d)]) != `"`+a.ID+`"` { + t.Errorf("kept %+v, act key %s", k, w.state.Vals["n1.act."+keys.EscapeName(d)]) } }}, {"§12: with no record within 5 seconds, Not run, asked; not yet taken by the router", func(t *testing.T, w *world) { diff --git a/node-tools/internal/approval/ask.go b/node-tools/internal/approval/ask.go index 60a27da..70ebadd 100644 --- a/node-tools/internal/approval/ask.go +++ b/node-tools/internal/approval/ask.go @@ -11,6 +11,7 @@ import ( "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/clients/operatorchannelclient" seat "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel" stdio "git.novox.be/novox/mesh-sdk/go" + "git.novox.be/novox/mesh-sdk/go/keys" ) // StateName is the runner's own state its asks and claims are kept in, per machine (§13). @@ -50,8 +51,8 @@ func (g *Gate) client(module string) *operatorchannelclient.Client { Record: func(id string) (*stdio.StateEntry, error) { return g.d.Record(module, id) }} } -// actKey is the key naming the newest ask of an act (§13). -func (g *Gate) actKey(digest string) string { return g.d.Node + ".act." + digest } +// actKey is the key naming the newest ask of an act (§13), its digest escaped as a key's name is, since the bus holds no `:` in a key (lead answer A-209). +func (g *Gate) actKey(digest string) string { return g.d.Node + ".act." + keys.EscapeName(digest) } // askKey is the key the client keeps an ask under (§13). func (g *Gate) askKey(id string) string { return g.d.Node + ".ask." + id }