jschoubben
63a290db3d
unit 302 and integration 34, as at the merge base 45e143f; the buscalls test moves beside busstate's. Builds: design 47 G §419, §626 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
93 lines
2.8 KiB
Go
93 lines
2.8 KiB
Go
//go:build integration
|
|
|
|
package views_test
|
|
|
|
import (
|
|
"context"
|
|
"encoding/json"
|
|
"testing"
|
|
"time"
|
|
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
|
|
"git.novox.be/novox/mesh-sdk/go/testbus"
|
|
"git.novox.be/novox/mesh-sdk/go/views"
|
|
)
|
|
|
|
// Against a real bus: the writer's events are acknowledged by their stream, a start states what the observer observes, an
|
|
// ask published before the writer's consumer existed is never answered, one published after is, and after the connection
|
|
// is made again (Reconnected) the consumer is made again from the newest ask (design 47 G §383, §390, §485).
|
|
func subWriterSaysItsEventsAndHearsAsksOnTheBus(t *testing.T) {
|
|
nc, err := nats.Connect(testbus.URL(t))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer nc.Close()
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
defer cancel()
|
|
subj := views.FindingSubjects("m", "n1")
|
|
for name, subjects := range map[string][]string{"EVENTS": {subj.Written, subj.Cleared, subj.Stated}, "ASKS": {asked}} {
|
|
if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: name, Subjects: subjects}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
if _, err := js.Publish(ctx, asked, []byte(`{"ask":"old","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
listen, err := js.CreateConsumer(ctx, "EVENTS", jetstream.ConsumerConfig{AckPolicy: jetstream.AckNonePolicy})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
state := &fakeState{m: map[string][]byte{}}
|
|
w, err := views.NewWriter(testConfig{Module: "m", State: state, JS: js, Conn: nc, Node: "n1", Asked: asked,
|
|
Observe: func(context.Context) (testObserved, error) {
|
|
return testObserved{Findings: []testFinding{finding("a", "now")}}, nil
|
|
}, Logf: t.Logf})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := w.Start(ctx); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
defer w.Close()
|
|
next := func(want string) testStated {
|
|
t.Helper()
|
|
m, err := listen.Next(jetstream.FetchMaxWait(10 * time.Second))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if m.Subject() != want {
|
|
t.Fatalf("said %s, want %s", m.Subject(), want)
|
|
}
|
|
var st testStated
|
|
if want == subj.Stated {
|
|
if err := json.Unmarshal(m.Data(), &st); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
return st
|
|
}
|
|
if st := next(subj.Stated); st.Asked != "" || len(st.Findings) != 1 {
|
|
t.Fatalf("start statement: %+v", st)
|
|
}
|
|
if err := w.Write(ctx, finding("b", "x")); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
next(subj.Written)
|
|
for _, ask := range []string{"new", "again"} {
|
|
w.Reconnected()
|
|
next(subj.Stated) // the statement after the connection was made again
|
|
if _, err := js.Publish(ctx, asked, []byte(`{"ask":"`+ask+`","at":"2026-11-11T12:00:00Z","by":"keeper"}`)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if st := next(subj.Stated); st.Asked != ask {
|
|
t.Fatalf("answered %q, want %q", st.Asked, ask)
|
|
}
|
|
}
|
|
}
|