RN5: the runner's whole-module integration test against the embedded bus
Starts the program, takes a take-in.taken, sees the pipeline placed and its build asked, takes the build's step-ended and answers a session's review through the seat, the proven caller recorded; the pipeline then ends passed. Builds: design 47 G §22, §44, §54, §70, §482, A-106, A-107 Claude-Session: https://claude.ai/code/session_01NcbgfGKaxw984XyAPfLm4h
This commit is contained in:
@@ -0,0 +1,190 @@
|
||||
//go:build integration
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/nats-io/nats.go"
|
||||
"github.com/nats-io/nats.go/jetstream"
|
||||
|
||||
builds "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshbuilds"
|
||||
catalog "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog"
|
||||
seat "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshpipelinerunner"
|
||||
"git.novox.be/novox/mesh-sdk/go/process"
|
||||
"git.novox.be/novox/mesh-sdk/go/testbus"
|
||||
)
|
||||
|
||||
type syncBuffer struct {
|
||||
mu sync.Mutex
|
||||
b bytes.Buffer
|
||||
}
|
||||
|
||||
func (s *syncBuffer) Write(p []byte) (int, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return s.b.Write(p)
|
||||
}
|
||||
func (s *syncBuffer) String() string { s.mu.Lock(); defer s.mu.Unlock(); return s.b.String() }
|
||||
|
||||
// The runner whole against the real bus (design 47 G §22, §44, §54, §70, §482, A-107): started as its process is, it takes a
|
||||
// take-in.taken and places a pipeline of the registered module, asks its build of the seat mesh-builds, takes the step-ended
|
||||
// of the build, and ends the placed review step when a session calls `review`, answered by the process with the caller the
|
||||
// bus proved; the pipeline then ends passed.
|
||||
func TestTheRunnerPlacesAPipelineAsksItsBuildTakesItsEndAndAnswersAReview(t *testing.T) {
|
||||
url := testbus.URL(t)
|
||||
nc, err := nats.Connect(url)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer nc.Close()
|
||||
js, err := jetstream.New(nc)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
|
||||
const module = "mesh-delivery-pipeline"
|
||||
for _, state := range []string{"pipelines", "kept", "toolchains", "findings", "first-views"} {
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_" + state, History: 1}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
if _, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: module + "_bounds", History: 1, LimitMarkerTTL: time.Minute}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: process.EventsStream, Subjects: []string{"mesh.mod.*.event.>", "mesh.seat.*.event.>"}})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if _, err := events.CreateOrUpdateConsumer(ctx, jetstream.ConsumerConfig{Durable: "n1_" + module, AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
asks, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASKS", Subjects: []string{"mesh.seat.*.accept.>"}, Retention: jetstream.WorkQueuePolicy})
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// The seats the runner calls, answered as their holders do.
|
||||
reg := catalog.Registration{Module: "app", Branches: []string{"main"}, Origin: "o/app", Policy: catalog.Policy{Steps: catalog.StepStatuses{Review: "optional"}}}
|
||||
answer := func(subject string, body any) {
|
||||
if _, err := nc.Subscribe(subject, func(m *nats.Msg) {
|
||||
raw, _ := json.Marshal(map[string]any{"result": body})
|
||||
_ = m.Respond(raw)
|
||||
}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
answer("mesh.seat.mesh-catalog.call.registrations.*", catalog.RegistrationsAnswer{Registrations: []catalog.Registration{reg}})
|
||||
answer("mesh.seat.mesh-catalog.call.revision.*", catalog.RevisionAnswer{NoneYet: true})
|
||||
answer("mesh.seat.mesh-hand-acts.call.hand-act.*", map[string]any{})
|
||||
|
||||
dir := t.TempDir()
|
||||
broker := filepath.Join(dir, "broker")
|
||||
raw, _ := json.Marshal(process.Account{URL: url, Node: "n1", Module: module, User: "runner", Password: "x"})
|
||||
if err := os.WriteFile(broker, raw, 0o600); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
logs := &syncBuffer{}
|
||||
runCtx, stop := context.WithCancel(ctx)
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
done <- run(runCtx, func(k string) string {
|
||||
if k == process.BrokerFileEnv {
|
||||
return broker
|
||||
}
|
||||
return ""
|
||||
}, logs)
|
||||
}()
|
||||
defer func() { stop(); <-done }()
|
||||
|
||||
pipelines, err := js.KeyValue(ctx, module+"_pipelines")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// A take-in.taken names one entry of the source revision; the runner places the pipeline of the registered module.
|
||||
taken, _ := json.Marshal(catalog.TakeInTaken{Key: "m.1", Kind: "taken", Branch: "main", Source: 3,
|
||||
Entries: []catalog.TakeInEntry{{Module: "app", Commit: "c1", Fingerprint: "src2:aa"}}})
|
||||
const key = "main.3.app"
|
||||
var p seat.Pipeline
|
||||
waitFor(t, ctx, logs, func() bool {
|
||||
// Said until the runner holds the registration, so the pipeline is placed with its policy's steps.
|
||||
if _, err := js.Publish(ctx, "mesh.mod.mesh-delivery-indexer.event.take-in.taken", taken); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
return readPipeline(ctx, pipelines, key, &p) && len(p.Steps) > 1
|
||||
})
|
||||
build := stepOf(p, seat.StepResultStepBuild)
|
||||
if build.Outcome != seat.StepResultOutcomeRunning || build.Ask == "" {
|
||||
t.Fatalf("the build step = %+v", build)
|
||||
}
|
||||
// The build is asked of the seat mesh-builds: its ask is on its work queue.
|
||||
waitFor(t, ctx, logs, func() bool { info, err := asks.Info(ctx); return err == nil && info.State.Msgs >= 1 })
|
||||
|
||||
// A session's review, while the build still runs, is answered by the runner's process, the caller the one the subject names.
|
||||
in, _ := json.Marshal(seat.ReviewInput{Module: "app", Commit: "c1", Fingerprint: "src2:aa", Outcome: "passed", Said: "reads well"})
|
||||
var reply struct {
|
||||
Result seat.ReviewAnswer `json:"result"`
|
||||
Error string `json:"error"`
|
||||
}
|
||||
waitFor(t, ctx, logs, func() bool {
|
||||
m, err := nc.Request("mesh.seat.mesh-pipeline-runner.call.review.agent~dev", in, 2*time.Second)
|
||||
return err == nil && json.Unmarshal(m.Data, &reply) == nil
|
||||
})
|
||||
if reply.Error != "" || reply.Result.Pipeline != key || reply.Result.Result.By != "agent.dev" || reply.Result.Result.Outcome != "passed" {
|
||||
t.Fatalf("review answered %+v (%s)", reply.Result, reply.Error)
|
||||
}
|
||||
if !readPipeline(ctx, pipelines, key, &p) || stepOf(p, seat.StepResultStepReview).By != "agent.dev" || p.Outcome != seat.PipelineOutcomeRunning {
|
||||
t.Fatalf("pipeline after the review = %+v\n%s", p, logs.String())
|
||||
}
|
||||
|
||||
// The seat's holder says the build ended; the runner takes it from the node's subject.
|
||||
ended, _ := json.Marshal(builds.StepEnded{ID: build.Ask, Pipeline: key, Step: "build", Node: "n1", Outcome: "passed",
|
||||
Entries: []builds.BuildEntry{{Module: "app", Commit: "c1", Fingerprint: "src2:aa"}}})
|
||||
if _, err := js.Publish(ctx, "mesh.seat.mesh-builds.event.step-ended.n1", ended); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
waitFor(t, ctx, logs, func() bool {
|
||||
return readPipeline(ctx, pipelines, key, &p) && stepOf(p, seat.StepResultStepBuild).Outcome == seat.StepResultOutcomePassed
|
||||
})
|
||||
|
||||
waitFor(t, ctx, logs, func() bool { return readPipeline(ctx, pipelines, key, &p) && p.Outcome == seat.PipelineOutcomePassed })
|
||||
}
|
||||
|
||||
func stepOf(p seat.Pipeline, step string) seat.StepResult {
|
||||
for _, s := range p.Steps {
|
||||
if s.Step == step {
|
||||
return s
|
||||
}
|
||||
}
|
||||
return seat.StepResult{}
|
||||
}
|
||||
|
||||
func readPipeline(ctx context.Context, kv jetstream.KeyValue, key string, p *seat.Pipeline) bool {
|
||||
e, err := kv.Get(ctx, key)
|
||||
if err != nil {
|
||||
return false
|
||||
}
|
||||
*p = seat.Pipeline{}
|
||||
return json.Unmarshal(e.Value(), p) == nil
|
||||
}
|
||||
|
||||
// waitFor asks cond until it holds or the test's context ends, the module's journal in the failure.
|
||||
func waitFor(t *testing.T, ctx context.Context, logs *syncBuffer, cond func() bool) {
|
||||
t.Helper()
|
||||
for ctx.Err() == nil {
|
||||
if cond() {
|
||||
return
|
||||
}
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
}
|
||||
t.Fatalf("it did not come about in time; the runner said:\n%s", logs.String())
|
||||
}
|
||||
Reference in New Issue
Block a user