package main import ( "encoding/json" "errors" "strings" "testing" "github.com/nats-io/nats.go" "github.com/novox/mesh-controller/internal/broker" ) func verifiedSenderRecords(free map[string]bool) (broker.Records, broker.Declared) { tg := broker.Seat{Name: "channel", Scope: "mesh", Accepts: []string{"show"}, Kinded: true, Kind: "telegram", DeclaredBy: "messenger", Capabilities: []string{"choice", "verified-sender"}} router := broker.Declared{Module: "messenger", RunsAs: "messenger", Holds: []broker.Seat{{Name: "operator-channel", Scope: "mesh", Accepts: []string{"ask"}}}, Uses: []broker.Seat{{Name: "channel", Scope: "mesh", Kinded: true, DeclaredBy: "messenger"}}} return broker.Records{Nodes: []string{"novox"}, RootFree: free, Assigned: map[string][]broker.Declared{ "novox": {router, {Module: "telegram", RunsAs: "telegram", Holds: []broker.Seat{tg}}}}}, router } type fakeBus struct { last map[string][]byte published map[string][]byte seq map[string]uint64 // between runs once, after a read and before the write: a push issuing meanwhile. between func() } func (f *fakeBus) get(node, module string) ([]byte, uint64, error) { if b, ok := f.last[node+"/"+module]; ok { return b, f.seq[node+"/"+module], nil } return nil, 0, nats.ErrMsgNotFound } func (f *fakeBus) put(node, module string, body []byte, seq uint64) error { if f.between != nil { f.between() f.between = nil } if seq != f.seq[node+"/"+module] { return errors.New("nats: wrong last sequence") } f.published[node+"/"+module] = body f.last[node+"/"+module] = body f.seq[node+"/"+module]++ return nil } func hasVerified(t *testing.T, membership []byte) bool { t.Helper() var m broker.Membership if err := json.Unmarshal(membership, &m); err != nil { t.Fatal(err) } for _, k := range m.SeatTraffic.Kinds { if k.Kind == "telegram" { return strings.Contains(strings.Join(k.Capabilities, ","), "verified-sender") } } t.Fatal("telegram not in the router's membership") return false } // A membership composed while the judgement failed is issued again once it passes, with the verified sender // (novox/hq issue 930); while it fails, nothing is given back (ADR 0259 ยง8). func TestAMembershipThatLostItsVerifiedSenderGetsItBackWhenTheJudgementPasses(t *testing.T) { failed, router := verifiedSenderRecords(nil) whereFailed := broker.PlacementsOf(failed, nil) bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}} body, _ := json.Marshal(broker.MembershipFor("novox", router, whereFailed)) bus.last["novox/messenger"] = body if hasVerified(t, body) { t.Fatal("composed with a failed judgement, and the verified sender is kept") } say := func(string, ...any) {} if n := reissueVerifiedSenders(failed, whereFailed, bus.get, bus.put, say); n != 0 || len(bus.published) != 0 { t.Fatalf("issued %d while the judgement still fails", n) } passing, _ := verifiedSenderRecords(map[string]bool{"novox": true}) wherePassing := broker.PlacementsOf(passing, nil) if n := reissueVerifiedSenders(passing, wherePassing, bus.get, bus.put, say); n != 1 { t.Fatalf("issued %d memberships after the judgement passed, want 1", n) } if !hasVerified(t, bus.published["novox/messenger"]) { t.Fatal("the membership issued again lacks the verified sender") } if n := reissueVerifiedSenders(passing, wherePassing, bus.get, bus.put, say); n != 0 { t.Fatalf("issued %d again with nothing changed", n) } } // What a composition drops it says, naming the kind, the machine and the judgement's reason. func TestADroppedVerifiedSenderIsSaidWithItsReason(t *testing.T) { records, _ := verifiedSenderRecords(nil) where := broker.PlacementsOf(records, nil) lines := droppedVerifiedSenders(records, where, map[string]string{"novox": "not measured: no verb heard"}) if len(lines) != 1 || !strings.Contains(lines[0], "telegram") || !strings.Contains(lines[0], "novox") || !strings.Contains(lines[0], "not measured: no verb heard") { t.Fatalf("said %q", lines) } records, _ = verifiedSenderRecords(map[string]bool{"novox": true}) if lines := droppedVerifiedSenders(records, broker.PlacementsOf(records, nil), nil); len(lines) != 0 { t.Fatalf("said %q of a kind that kept its verified sender", lines) } } func TestAMembershipNeverIssuedIsLeftToItsOwnSend(t *testing.T) { records, _ := verifiedSenderRecords(map[string]bool{"novox": true}) bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}} var said []string n := reissueVerifiedSenders(records, broker.PlacementsOf(records, nil), bus.get, bus.put, func(f string, a ...any) { said = append(said, f) }) if n != 0 || len(said) != 0 { t.Fatalf("issued %d, said %v", n, said) } } // **On a real bus** (novox/hq issue 930): a router's membership issued without the verified sender because the // judgement failed is, once the judgement passes, published again to the AssignmentsStream with it in its seat // traffic's kinds, and a second pass publishes nothing. // A push issuing a newer membership between the self-check's read and its write is not overwritten (issue 930). func TestANewerMembershipIssuedMeanwhileIsNotOverwritten(t *testing.T) { failed, router := verifiedSenderRecords(nil) bus := &fakeBus{last: map[string][]byte{}, published: map[string][]byte{}, seq: map[string]uint64{}} old, _ := json.Marshal(broker.MembershipFor("novox", router, broker.PlacementsOf(failed, nil))) bus.last["novox/messenger"], bus.seq["novox/messenger"] = old, 1 newer := []byte(`{"node":"novox","module":"messenger","newer":true}`) bus.between = func() { bus.last["novox/messenger"], bus.seq["novox/messenger"] = newer, 2 } passing, _ := verifiedSenderRecords(map[string]bool{"novox": true}) if n := reissueVerifiedSenders(passing, broker.PlacementsOf(passing, nil), bus.get, bus.put, func(string, ...any) {}); n != 0 { t.Fatalf("issued %d over a newer membership", n) } if string(bus.last["novox/messenger"]) != string(newer) || len(bus.published) != 0 { t.Fatalf("the newer membership was overwritten: %s", bus.last["novox/messenger"]) } }