diff --git a/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go new file mode 100644 index 00000000..a143be7d --- /dev/null +++ b/modules/mesh-delivery-pipeline/cmd/mesh-delivery-pipeline/run_integration_test.go @@ -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()) +}