// Package announcetest is the test each module runs with its own checks to hold that it announces on the bus every // state it keeps (novox/hq ADR 0495 clause 5; issue 1790): it starts the module against a bus of the test's own and // compares what the module announced with what its code declares. With the bind guard of process.Conn.Keeps, // stdio.Keeps and the TypeScript keeps, which refuses a state the code did not announce, announcement and code agree. // // It replaces the merge gate's check of one module's reads against another's manifest (clause 5); a test using it // carries `//go:build integration`, since it starts a bus. package announcetest import ( "context" "fmt" "reflect" "sort" "testing" "time" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/announce" "git.novox.be/novox/mesh-sdk/go/testbus" ) // Wait is how long Check waits for the module's announcement once it started it. var Wait = 10 * time.Second // Check starts a bus of the test's own, with a stream taking module's events, calls start with its address (start // runs the module, or its part that connects, against it, and answers how to stop it), waits for the module's // announcement and fails the test unless it is module's and names exactly want (in any order), readable as want says. // It answers the announcement. func Check(t testing.TB, module string, want []announce.State, start func(url string) (stop func())) announce.States { t.Helper() url := testbus.URL(t) nc, err := nats.Connect(url) if err != nil { t.Fatalf("announcetest: connecting to the test bus: %v", err) } defer nc.Close() js, err := jetstream.New(nc) if err != nil { t.Fatal(err) } ctx, cancel := context.WithTimeout(context.Background(), Wait) defer cancel() stream, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ANNOUNCETEST", Subjects: []string{"mesh.mod." + module + ".event.>"}}) if err != nil { t.Fatalf("announcetest: making the events stream: %v", err) } stop := start(url) if stop != nil { defer stop() } got, err := waitFor(ctx, stream, module) if err != nil { t.Fatalf("announcetest: %v", err) } if problem := differs(got, module, want); problem != "" { t.Fatalf("announcetest: %s", problem) } return got } // waitFor reads the stream until module's announcement is in it. func waitFor(ctx context.Context, stream jetstream.Stream, module string) (announce.States, error) { for { msg, err := stream.GetLastMsgForSubject(ctx, announce.Subject(module)) if err == nil { return announce.Decode(msg.Data) } select { case <-ctx.Done(): return announce.States{}, fmt.Errorf("%s announced nothing on %s within %s", module, announce.Subject(module), Wait) case <-time.After(50 * time.Millisecond): } } } // differs says how got is not module's announcement of want, or "". func differs(got announce.States, module string, want []announce.State) string { if got.Module != module { return fmt.Sprintf("the announcement names the module %q, not %q", got.Module, module) } w := append([]announce.State{}, want...) sort.Slice(w, func(i, j int) bool { return w[i].Name < w[j].Name }) g := append([]announce.State{}, got.States...) if !reflect.DeepEqual(g, w) && !(len(g) == 0 && len(w) == 0) { return fmt.Sprintf("%s announced %v, and its code declares %v", module, g, w) } return "" }