TLS-5 integration test: the owners' own views streams, the runner's consumers on them and a membership naming them, events on views.Emit's subject
Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
This commit is contained in:
@@ -11,6 +11,8 @@ import (
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"git.novox.be/novox/mesh-sdk/go/catalogue"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/views"
|
||||
"github.com/nats-io/nats.go"
|
||||
|
||||
@@ -19,13 +21,15 @@ import (
|
||||
)
|
||||
|
||||
// issue 1966 TLS-5 [i] (ADR 0615 §2, §4, §6): against the real bus, with the owners down and then up, the runner's own
|
||||
// consumer carries the manifest's four `consumes` to the owners' views events: the first views are asked until answered,
|
||||
// consumers on the owners' own views streams carry the manifest's four `consumes`: the first views are asked until answered,
|
||||
// the unanswered ones are findings that the answer clears, and each owner's next event reaches its picture.
|
||||
func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThenUp(t *testing.T) {
|
||||
mesh := mt.New(t)
|
||||
url := mt.URL(t)
|
||||
|
||||
// The consumer is the mesh's, made from the manifest's `consumes` as the mesh makes it: one filter per family.
|
||||
// The owners publish their views on their own streams (views.Emit: `mesh.stream.<owner>.event.views.<state>`, on
|
||||
// assigner-views and indexer-views), never on EVENTS. The consumers on them are the mesh's, one filter per family the
|
||||
// manifest's `consumes` names, and the runner's membership names them as the mesh issues it (ADR 0363 §8).
|
||||
raw, err := os.ReadFile("../../module.json")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
@@ -36,15 +40,30 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen
|
||||
if err := json.Unmarshal(raw, &manifest); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
var filters []string
|
||||
streamOf := map[string]string{Assigner: catalogue.StreamBusName("assigner-views"), Indexer: catalogue.StreamBusName("indexer-views")}
|
||||
filters := map[string][]string{}
|
||||
families := 0
|
||||
for _, c := range manifest.Consumes {
|
||||
emitter, event, _ := strings.Cut(c, ".views.")
|
||||
filters = append(filters, "mesh.mod."+emitter+".event.views."+event)
|
||||
stream := streamOf[emitter]
|
||||
if stream == "" {
|
||||
t.Fatalf("the manifest consumes %s, whose views stream the test does not know", c)
|
||||
}
|
||||
filters[stream] = append(filters[stream], process.ModuleStreamSubject(emitter, views.EventPrefix+event))
|
||||
families++
|
||||
}
|
||||
if len(filters) != len(Specs) {
|
||||
t.Fatalf("the manifest consumes %d families, the runner holds %d first views", len(filters), len(Specs))
|
||||
if families != len(Specs) {
|
||||
t.Fatalf("the manifest consumes %d families, the runner holds %d first views", families, len(Specs))
|
||||
}
|
||||
mesh.Consumer(t, "anchor", "node-tools", filters, 2*time.Second)
|
||||
runner := mt.MembershipOf("node-tools", "anchor", false, nil)
|
||||
for _, owner := range []string{Assigner, Indexer} {
|
||||
stream := streamOf[owner]
|
||||
mesh.ModuleStream(t, owner, stream)
|
||||
mesh.ConsumerOn(t, stream, "anchor", "node-tools", filters[stream], 2*time.Second)
|
||||
runner.Hears = append(runner.Hears, bus.HeardIssued{Stream: stream, Consumer: "anchor_node-tools"})
|
||||
}
|
||||
mesh.Consumer(t, "anchor", "node-tools", []string{"mesh.mod.mesh-knowledge.event.reviewed"}, 2*time.Second) // its `waits`, on EVENTS
|
||||
mesh.Issue(t, runner)
|
||||
|
||||
conn, err := bus.Connect(bus.Credential{URL: url, Module: "node-tools", Node: "anchor"})
|
||||
if err != nil {
|
||||
@@ -52,6 +71,13 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen
|
||||
}
|
||||
conn.Logf = func(string, ...any) {}
|
||||
t.Cleanup(conn.Close)
|
||||
conn.Follow("node-tools")
|
||||
mt.Until(t, func() error {
|
||||
if conn.Membership("node-tools") == nil {
|
||||
return errFindings(-1)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
|
||||
var mu sync.Mutex
|
||||
findings := map[string]bool{}
|
||||
@@ -110,7 +136,7 @@ func TestTheRunnersConsumerCarriesTheOwnersViewsFamiliesWhenTheOwnersAreDownThen
|
||||
for _, s := range Specs {
|
||||
ev, _ := json.Marshal(views.Event{V: 1, Owner: s.Owner, State: s.State, Key: "next", Revision: 5, Created: time.Unix(1000, 0).UTC(),
|
||||
Value: json.RawMessage(`{"n":2}`)})
|
||||
mesh.Emit(t, "mesh.mod."+s.Owner+".event.views."+s.State, json.RawMessage(ev))
|
||||
mesh.Emit(t, process.ModuleStreamSubject(s.Owner, views.EventPrefix+s.State), json.RawMessage(ev))
|
||||
}
|
||||
for _, s := range Specs {
|
||||
name := s.Name()
|
||||
|
||||
Reference in New Issue
Block a user