796 lines
28 KiB
Go
796 lines
28 KiB
Go
//go:build integration
|
|
|
|
package main
|
|
|
|
import (
|
|
"bufio"
|
|
"context"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"os"
|
|
"os/exec"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
oc "git.novox.be/novox/mesh-contracts/operator-and-conversation/go/seats/operatorchannel"
|
|
mdc "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshdeliveryclient"
|
|
"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-forge-bridge/forgemsg"
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
"github.com/nats-io/nats-server/v2/server"
|
|
"github.com/nats-io/nats.go"
|
|
"github.com/nats-io/nats.go/jetstream"
|
|
)
|
|
|
|
// **The acceptance of acting on events** (novox/hq ADR 0406, design 47 milestone 2, ADR 0391 phase 3, issue 975; ADR
|
|
// 0421, issue 1022): a pull request whose verdict passes is asked of the operator within seconds of the verdict's event,
|
|
// with no label, comment or mark written by anyone, and the merge is asked within five seconds of the Approve's event —
|
|
// with no tick, no poll and no reconcile to find it.
|
|
// Proved over a real bus: a JetStream server is linked into the test and started for it alone; the module is this
|
|
// test binary started again as the bundle, speaking MCP on its standard streams to a runtime played here, which does
|
|
// what the tool runner does for a bundle — its state on the buckets the controller would make from the manifest, its
|
|
// events delivered from the bus's streams, the controller's verbs and the forge's tools answered by the test (the
|
|
// pattern of mesh-issues' bus_test).
|
|
|
|
const bundleVar = "MESH_DELIVERY_TEST_BUNDLE"
|
|
|
|
func TestMain(m *testing.M) {
|
|
if os.Getenv(bundleVar) == "1" {
|
|
main()
|
|
os.Exit(0)
|
|
}
|
|
os.Exit(m.Run())
|
|
}
|
|
|
|
type testBus struct {
|
|
nc *nats.Conn
|
|
js jetstream.JetStream
|
|
kvs map[string]jetstream.KeyValue
|
|
}
|
|
|
|
func newTestBus(t *testing.T) *testBus {
|
|
t.Helper()
|
|
opts := &server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true, StoreDir: t.TempDir(),
|
|
NoLog: true, NoSigs: true}
|
|
s, err := server.NewServer(opts)
|
|
if err != nil {
|
|
t.Fatalf("a bus for this test could not be made: %v", err)
|
|
}
|
|
go s.Start()
|
|
if !s.ReadyForConnections(30 * time.Second) {
|
|
s.Shutdown()
|
|
t.Fatal("a bus for this test did not come up within 30s")
|
|
}
|
|
t.Cleanup(func() { s.Shutdown(); s.WaitForShutdown() })
|
|
nc, err := nats.Connect(s.ClientURL())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Cleanup(nc.Close)
|
|
js, err := jetstream.New(nc)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b := &testBus{nc: nc, js: js, kvs: map[string]jetstream.KeyValue{}}
|
|
// The controller's job: a bucket per state the manifest declares.
|
|
m, _ := readManifest(t)
|
|
for _, name := range m.buckets() {
|
|
kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: m.Module + "_" + name, History: 10})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.kvs[name] = kv
|
|
}
|
|
// The buckets of other modules the manifest reads (`<module>.<name>` is the bucket `<module>_<name>`): the teller's answers
|
|
// and the checker's verdicts.
|
|
for _, name := range m.Reads {
|
|
kv, err := js.CreateKeyValue(context.Background(), jetstream.KeyValueConfig{Bucket: strings.Replace(name, ".", "_", 1), History: 10})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
b.kvs[name] = kv
|
|
}
|
|
// The streams the events are on (mesh.mod.<module>.event.<name>, mesh.seat.<seat>.event.<name>).
|
|
if _, err := js.CreateStream(context.Background(), jetstream.StreamConfig{Name: "EVENTS",
|
|
Subjects: []string{"mesh.mod.gitea.event.>", "mesh.seat." + oc.Seat + ".event.>",
|
|
"mesh.mod." + forgeBridgeModule + ".event.>"}}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return b
|
|
}
|
|
|
|
// runtime is the half of the conversation the tool runner plays for one running bundle.
|
|
type runtime struct {
|
|
t *testing.T
|
|
b *testBus
|
|
cmd *exec.Cmd
|
|
in io.WriteCloser
|
|
mu sync.Mutex
|
|
next int
|
|
wait map[string]chan json.RawMessage
|
|
|
|
// what the bundle subscribed to, and asked of the mesh.
|
|
subs []string
|
|
asked []askedOfMesh
|
|
notify chan struct{}
|
|
// the asks the operator was asked, and the controller's walks and the forge's open pull requests as the test plays
|
|
// them.
|
|
operatorAsks []oc.Ask
|
|
// operatorCancels are the ids of the asks taken back from the router, in order (issue 1079).
|
|
operatorCancels []string
|
|
openPulls []OpenPull
|
|
}
|
|
|
|
type askedOfMesh struct {
|
|
Key string
|
|
Body map[string]any
|
|
At time.Time
|
|
}
|
|
|
|
func (b *testBus) start(t *testing.T) *runtime {
|
|
t.Helper()
|
|
cmd := exec.Command(os.Args[0], "-test.run=^$")
|
|
cmd.Env = append(os.Environ(), bundleVar+"=1", "MESH_SERVED_MODULE=mesh-delivery-gate", "MESH_MODULE=mesh-delivery-gate", "MESH_NODE=test")
|
|
in, err := cmd.StdinPipe()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
out, err := cmd.StdoutPipe()
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
cmd.Stderr = testLog{t}
|
|
if err := cmd.Start(); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
r := &runtime{t: t, b: b, cmd: cmd, in: in, wait: map[string]chan json.RawMessage{}, notify: make(chan struct{}, 64)}
|
|
go r.read(out)
|
|
t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() })
|
|
return r
|
|
}
|
|
|
|
type testLog struct{ t *testing.T }
|
|
|
|
func (w testLog) Write(p []byte) (int, error) {
|
|
defer func() { _ = recover() }() // the bundle may speak after the test ended
|
|
w.t.Logf("bundle: %s", strings.TrimRight(string(p), "\n"))
|
|
return len(p), nil
|
|
}
|
|
|
|
type rpcMessage struct {
|
|
ID json.RawMessage `json:"id,omitempty"`
|
|
Method string `json:"method,omitempty"`
|
|
Params json.RawMessage `json:"params,omitempty"`
|
|
Result json.RawMessage `json:"result,omitempty"`
|
|
Error *struct {
|
|
Code int `json:"code"`
|
|
Message string `json:"message"`
|
|
} `json:"error,omitempty"`
|
|
}
|
|
|
|
func (r *runtime) say(m map[string]any) {
|
|
m["jsonrpc"] = "2.0"
|
|
raw, _ := json.Marshal(m)
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
_, _ = r.in.Write(append(raw, '\n'))
|
|
}
|
|
|
|
func (r *runtime) read(out io.Reader) {
|
|
sc := bufio.NewScanner(out)
|
|
sc.Buffer(make([]byte, 1<<20), 1<<24)
|
|
for sc.Scan() {
|
|
var m rpcMessage
|
|
if json.Unmarshal(sc.Bytes(), &m) != nil {
|
|
continue
|
|
}
|
|
if m.Method == "" {
|
|
var id string
|
|
_ = json.Unmarshal(m.ID, &id)
|
|
r.mu.Lock()
|
|
ch := r.wait[id]
|
|
delete(r.wait, id)
|
|
r.mu.Unlock()
|
|
if ch != nil {
|
|
ch <- sc.Bytes()
|
|
}
|
|
continue
|
|
}
|
|
go r.answer(m)
|
|
}
|
|
}
|
|
|
|
// untilListening waits for the gate to have read its kept state back and subscribed to every event it follows: it takes no
|
|
// event before (loadKept), so a subscription is its word that it is ready.
|
|
func (r *runtime) untilListening() {
|
|
r.t.Helper()
|
|
r.until("the gate to listen to every event it follows", 20*time.Second, func() bool {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
return len(r.subs) >= len(followed)
|
|
})
|
|
}
|
|
|
|
// noControllerAsk fails the test when the gate asked the controller anything: it asks it nothing (ADR 0539 §17, R6-C3).
|
|
func (r *runtime) noControllerAsk() {
|
|
r.t.Helper()
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
for _, a := range r.asked {
|
|
if strings.HasPrefix(a.Key, "seat:mesh-controller.") {
|
|
r.t.Fatalf("the gate asked the controller %s: it asks it nothing", a.Key)
|
|
}
|
|
}
|
|
}
|
|
|
|
func (r *runtime) ping() {
|
|
select {
|
|
case r.notify <- struct{}{}:
|
|
default:
|
|
}
|
|
}
|
|
|
|
// asksOf are the asks of the mesh made so far under a key.
|
|
func (r *runtime) asksOf(key string) []askedOfMesh {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
var out []askedOfMesh
|
|
for _, a := range r.asked {
|
|
if a.Key == key {
|
|
out = append(out, a)
|
|
}
|
|
}
|
|
return out
|
|
}
|
|
|
|
// until waits for what the test needs to be true, woken by every message of the bundle.
|
|
func (r *runtime) until(what string, within time.Duration, ok func() bool) {
|
|
r.t.Helper()
|
|
deadline := time.After(within)
|
|
for {
|
|
if ok() {
|
|
return
|
|
}
|
|
select {
|
|
case <-r.notify:
|
|
case <-time.After(50 * time.Millisecond):
|
|
case <-deadline:
|
|
if ok() {
|
|
return
|
|
}
|
|
r.t.Fatalf("within %s: %s did not happen", within, what)
|
|
}
|
|
}
|
|
}
|
|
|
|
// toolAnswer is a tool's or a verb's answer as the runtime hands it over.
|
|
func toolAnswer(answer any) any {
|
|
return map[string]any{"ok": true, "output": "", "answer": answer}
|
|
}
|
|
|
|
func (r *runtime) answer(m rpcMessage) {
|
|
var p struct {
|
|
State string `json:"state"`
|
|
Key string `json:"key"`
|
|
Value json.RawMessage `json:"value"`
|
|
Revision uint64 `json:"revision"`
|
|
Pattern string `json:"pattern"`
|
|
Subject string `json:"subject"`
|
|
Body json.RawMessage `json:"body"`
|
|
}
|
|
_ = json.Unmarshal(m.Params, &p)
|
|
ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
|
|
defer cancel()
|
|
defer r.ping()
|
|
result, err := func() (any, error) {
|
|
switch m.Method {
|
|
case "mesh/subscribe":
|
|
r.mu.Lock()
|
|
r.subs = append(r.subs, p.Pattern)
|
|
r.mu.Unlock()
|
|
return map[string]any{}, nil
|
|
case "mesh/publish":
|
|
return map[string]any{}, nil // the module's own events: not needed here
|
|
case "mesh/seat.publish":
|
|
// The ask for the operator, published under the module's name.
|
|
if strings.Contains(p.Subject, ".accept.ask.") {
|
|
var a oc.Ask
|
|
if err := json.Unmarshal(p.Body, &a); err != nil {
|
|
return nil, err
|
|
}
|
|
r.mu.Lock()
|
|
r.operatorAsks = append(r.operatorAsks, a)
|
|
r.mu.Unlock()
|
|
}
|
|
if strings.Contains(p.Subject, ".accept.cancel.") {
|
|
var c struct {
|
|
ID string `json:"id"`
|
|
}
|
|
if err := json.Unmarshal(p.Body, &c); err != nil {
|
|
return nil, err
|
|
}
|
|
r.mu.Lock()
|
|
r.operatorCancels = append(r.operatorCancels, c.ID)
|
|
r.mu.Unlock()
|
|
}
|
|
return map[string]any{}, nil
|
|
case "mesh/seat.record":
|
|
return nil, nil
|
|
case "mesh/state.watch":
|
|
// The runtime's watch (ADR 0201): each current value, then each change, handed to the bundle as mesh/state under the
|
|
// id the watch was asked with.
|
|
var id string
|
|
_ = json.Unmarshal(m.ID, &id)
|
|
kv := r.b.kvs[p.State]
|
|
if kv == nil {
|
|
return nil, fmt.Errorf("the state %q is not declared", p.State)
|
|
}
|
|
go r.watch(kv, id, p.State)
|
|
return map[string]any{}, nil
|
|
case "mesh/ask":
|
|
var a struct {
|
|
Key string `json:"key"`
|
|
Body map[string]any `json:"body"`
|
|
}
|
|
if err := json.Unmarshal(m.Params, &a); err != nil {
|
|
return nil, err
|
|
}
|
|
r.mu.Lock()
|
|
r.asked = append(r.asked, askedOfMesh{Key: a.Key, Body: a.Body, At: time.Now()})
|
|
pulls := append([]OpenPull{}, r.openPulls...)
|
|
r.mu.Unlock()
|
|
switch a.Key {
|
|
case "":
|
|
return nil, errors.New("an ask with no key")
|
|
case "gitea.gitea_open_pulls":
|
|
return toolAnswer(map[string]any{"pulls": pulls}), nil
|
|
case "gitea.gitea_commit_statuses":
|
|
// Green: mesh/check a success at the head asked, as the gitea module answers.
|
|
sha, _ := a.Body["sha"].(string)
|
|
if sha == "" {
|
|
sha = "0000000"
|
|
}
|
|
return toolAnswer(map[string]any{"commit": sha, "statuses": []any{
|
|
map[string]any{"context": "mesh/check", "state": "success"}},
|
|
"merge": map[string]any{"branch": "main", "protected": true, "required": []string{"mesh/check"},
|
|
"blocking": []any{}, "mergeable": true, "says": "main requires mesh/check"}}), nil
|
|
case "gitea.gitea_note_append", "gitea.gitea_delivery_view", "gitea.gitea_commit_status", "gitea.gitea_merge_pull_request",
|
|
"gitea.gitea_add_comment":
|
|
return toolAnswer(map[string]any{}), nil
|
|
}
|
|
if strings.HasPrefix(a.Key, "seat:mesh-controller.") {
|
|
// The fake controller of this test fails on any call: the gate has no business with it.
|
|
return nil, fmt.Errorf("the controller is not to be called by the gate: %s", a.Key)
|
|
}
|
|
return nil, fmt.Errorf("the runtime answers no ask %s in this test", a.Key)
|
|
}
|
|
kv := r.b.kvs[p.State]
|
|
if kv == nil {
|
|
return nil, fmt.Errorf("the state %q is not declared", p.State)
|
|
}
|
|
switch m.Method {
|
|
case "mesh/state.get":
|
|
e, err := kv.Get(ctx, p.Key)
|
|
if errors.Is(err, jetstream.ErrKeyNotFound) {
|
|
return nil, nil
|
|
}
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
return map[string]any{"key": p.Key, "value": json.RawMessage(e.Value()), "revision": e.Revision()}, nil
|
|
case "mesh/state.create":
|
|
rev, err := kv.Create(ctx, p.Key, p.Value)
|
|
return map[string]any{"revision": rev}, err
|
|
case "mesh/state.update":
|
|
rev, err := kv.Update(ctx, p.Key, p.Value, p.Revision)
|
|
return map[string]any{"revision": rev}, err
|
|
case "mesh/state.put":
|
|
rev, err := kv.Put(ctx, p.Key, p.Value)
|
|
return map[string]any{"revision": rev}, err
|
|
case "mesh/state.delete":
|
|
return map[string]any{}, kv.Delete(ctx, p.Key)
|
|
case "mesh/state.keys":
|
|
keys, err := kv.Keys(ctx)
|
|
if errors.Is(err, jetstream.ErrNoKeysFound) {
|
|
return []string{}, nil
|
|
}
|
|
return keys, err
|
|
}
|
|
return nil, fmt.Errorf("the runtime offers no %s to this bundle", m.Method)
|
|
}()
|
|
if err != nil {
|
|
code := -32000
|
|
var api *jetstream.APIError
|
|
if errors.Is(err, jetstream.ErrKeyExists) || errors.As(err, &api) && api.ErrorCode == jetstream.JSErrCodeStreamWrongLastSequence {
|
|
code = -32010
|
|
}
|
|
r.say(map[string]any{"id": m.ID, "error": map[string]any{"code": code, "message": err.Error()}})
|
|
return
|
|
}
|
|
r.say(map[string]any{"id": m.ID, "result": result})
|
|
}
|
|
|
|
// watch hands the bundle each key of a bucket and each change of it, as the runtime does for a state watch.
|
|
func (r *runtime) watch(kv jetstream.KeyValue, id, state string) {
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
r.t.Cleanup(cancel)
|
|
w, err := kv.Watch(ctx, ">")
|
|
if err != nil {
|
|
return
|
|
}
|
|
defer w.Stop()
|
|
current := true
|
|
for {
|
|
select {
|
|
case <-ctx.Done():
|
|
return
|
|
case e, ok := <-w.Updates():
|
|
if !ok {
|
|
return
|
|
}
|
|
if e == nil {
|
|
current = false
|
|
continue
|
|
}
|
|
op := "put"
|
|
if e.Operation() != jetstream.KeyValuePut {
|
|
op = "delete"
|
|
}
|
|
r.mu.Lock()
|
|
n := r.next
|
|
r.next++
|
|
r.mu.Unlock()
|
|
r.say(map[string]any{"id": fmt.Sprintf("watch-%d", n), "method": "mesh/state", "params": map[string]any{"watch": id,
|
|
"change": stdio.StateChange{State: state, Key: e.Key(), Op: op, Value: e.Value(), Revision: e.Revision(), Current: current}}})
|
|
}
|
|
}
|
|
}
|
|
|
|
// deliverEvents is the runtime reading the events streams and handing each event to the bundle, once, as
|
|
// `mesh/event`, the way the tool runner does for a module's consumer.
|
|
func (r *runtime) deliverEvents(ctx context.Context) {
|
|
stream, err := r.b.js.Stream(ctx, "EVENTS")
|
|
if err != nil {
|
|
r.t.Fatal(err)
|
|
}
|
|
cons, err := stream.OrderedConsumer(ctx, jetstream.OrderedConsumerConfig{})
|
|
if err != nil {
|
|
r.t.Fatal(err)
|
|
}
|
|
cc, err := cons.Consume(func(msg jetstream.Msg) {
|
|
key := keyOfSubject(msg.Subject())
|
|
r.mu.Lock()
|
|
id := fmt.Sprintf("event-%d", r.next)
|
|
r.next++
|
|
r.mu.Unlock()
|
|
r.say(map[string]any{"id": id, "method": "mesh/event", "params": map[string]any{"envelope": stdio.Envelope{
|
|
Key: key, Node: "test", Body: msg.Data(), Headers: map[string]string{"x-source": strings.SplitN(key, ".", 2)[0]}}}})
|
|
})
|
|
if err != nil {
|
|
r.t.Fatal(err)
|
|
}
|
|
r.t.Cleanup(cc.Stop)
|
|
}
|
|
|
|
// keyOfSubject is an event's key, as the runtime gives it to a module: mesh.mod.<module>.event.<name> is
|
|
// <module>.<name>, and mesh.seat.<seat>.event.<name> is <seat>.<name>.
|
|
func keyOfSubject(subject string) string {
|
|
parts := strings.Split(subject, ".")
|
|
if len(parts) < 5 || parts[3] != "event" {
|
|
return subject
|
|
}
|
|
return parts[2] + "." + strings.Join(parts[4:], ".")
|
|
}
|
|
|
|
func (b *testBus) publish(t *testing.T, subject string, body any) time.Time {
|
|
t.Helper()
|
|
raw, err := json.Marshal(body)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.js.Publish(context.Background(), subject, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return time.Now()
|
|
}
|
|
|
|
func TestAPassingVerdictAsksTheApproveAndItsApproveMergesWithinSecondsOverARealBus(t *testing.T) {
|
|
b := newTestBus(t)
|
|
ctx := context.Background()
|
|
rt := b.start(t)
|
|
evctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
rt.deliverEvents(evctx)
|
|
// The module has started when it hears what it follows.
|
|
rt.untilListening()
|
|
for _, k := range followed {
|
|
if strings.Contains(k, "labeled") || strings.Contains(k, "commented") {
|
|
t.Fatalf("the module follows %s: a label or a comment starts nothing here (ADR 0421 §4)", k)
|
|
}
|
|
}
|
|
|
|
// A pull request into main is announced, and the controller's verdict on its head passes.
|
|
const head = "1111111111111111111111111111111111111111"
|
|
announced := pr("novox/app", 7, head, "fix/x", "src/a.go")
|
|
announced.BaseSHA = "a1a1a1a1a1a1"
|
|
announced.Title = "Fix the apostrophe (issue 975)"
|
|
announced.Body = "The description, with no review line yet."
|
|
b.publish(t, "mesh.mod.gitea.event.pull.updated", announced)
|
|
rt.until("the pull request's key to hold its head", eventWithin, func() bool {
|
|
e, err := b.kvs["pulls"].Get(ctx, "novox.app.7")
|
|
var k PullKey
|
|
return err == nil && json.Unmarshal(e.Value(), &k) == nil && k.Head == head && k.Trunk == "a1a1a1a1a1a1"
|
|
})
|
|
b.checkPasses(t, rt, "fix/x", "a1a1a1a1a1a1", judgedPull{"novox/app", 7, head})
|
|
|
|
// The description holds no review line and nothing is waited for: the ask is written into the gate's bucket asks within
|
|
// seconds of the verdict, with no label, comment or mark written by anyone (the gate reads no description, ADR 0493 §6, ADR 0659).
|
|
edit := time.Now()
|
|
var key string
|
|
rt.until("the ask to be written", eventWithin, func() bool { key = onlyAsk(b); return key != "" })
|
|
ask, err := askOf(askAt(t, b, key))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
t.Logf("the ask was written %s after the verdict", time.Since(edit))
|
|
if strings.Contains(strings.ToLower(ask.Explanation), "review") || ask.ID != mdc.AskID(key) {
|
|
t.Fatalf("the ask names a review or has the wrong id: %s", ask.Explanation)
|
|
}
|
|
if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 {
|
|
t.Fatalf("merged before the Approve")
|
|
}
|
|
// The pull request's key holds its head, the trunk commit it was asked against and the verdict.
|
|
rt.until("the pull request's key to hold head, trunk and verdict", eventWithin, func() bool {
|
|
e, err := b.kvs["pulls"].Get(ctx, "novox.app.7")
|
|
var k PullKey
|
|
return err == nil && json.Unmarshal(e.Value(), &k) == nil && k == PullKey{Head: head, Trunk: "a1a1a1a1a1a1", Verdict: "pass"}
|
|
})
|
|
|
|
// What the transitions owe the forge is written as owed keys; the gate calls no tool of the forge that writes (issue 1349).
|
|
for _, tool := range []string{"gitea.gitea_commit_status", "gitea.gitea_delivery_view", "gitea.gitea_note_append", "gitea.gitea_add_comment"} {
|
|
if n := len(rt.asksOf(tool)); n != 0 {
|
|
t.Fatalf("the gate asked %s %d times: the bridge writes to the forge", tool, n)
|
|
}
|
|
}
|
|
rt.until("the owed keys to be written", eventWithin, func() bool {
|
|
keys, err := b.kvs["owed"].Keys(ctx)
|
|
return err == nil && len(keys) > 0
|
|
})
|
|
// The operator approves; the teller holds the warrant and writes the answer: the merge is handed to the bridge within
|
|
// seconds, with the Approve's heads, and the gate asks the forge nothing.
|
|
approved := b.answer(t, key, OptionApprove)
|
|
var handed Merge
|
|
rt.until("the merge to be handed to the bridge", eventWithin, func() bool { handed = mergeAt(b, "fix/x"); return len(handed.Pulls) > 0 })
|
|
if len(handed.Pulls) != 1 || handed.Pulls[0].Head.Head != head || handed.Pulls[0].Number != 7 || handed.Pulls[0].State != mdc.MergeMerging ||
|
|
handed.Approved[mdc.ApprovedKey("novox/app", 7)] != head || handed.Ask != key {
|
|
t.Fatalf("the merge handed is %+v; wanted pull request 7 at the approved head, merging, under the ask %s", handed, key)
|
|
}
|
|
t.Logf("the merge was handed %s after the answer", time.Since(approved))
|
|
if n := len(rt.asksOf("gitea.gitea_merge_pull_request")); n != 0 {
|
|
t.Fatalf("the gate asked the forge to merge %d times: the bridge merges", n)
|
|
}
|
|
// The bridge says it merged; the mesh's source then holds the merge, in one write.
|
|
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: "novox/app", Number: 7, Head: head, Merge: strings.Repeat("b", 40),
|
|
Root: "r", Trees: map[string]string{"app": "t"}})
|
|
// The forge's own word that the pull request merged follows the bridge's (its webhook): the delivery ends merged.
|
|
b.publish(t, "mesh.mod.gitea.event.pull.merged", merged("novox/app", 7, head, strings.Repeat("b", 40)))
|
|
rt.until("source.main to hold the merge", eventWithin, func() bool {
|
|
e, err := b.kvs["source"].Get(ctx, mdc.SourceMainKey)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
src, err := mdc.DecodeSource(e.Value())
|
|
return err == nil && src.Group == "fix/x" && src.Heads["novox/app"] == strings.Repeat("b", 40)
|
|
})
|
|
// The delivery ends merged (ADR 0539 §17), and the gate asked the controller nothing (R6-C3).
|
|
rt.until("the delivery to be merged", eventWithin, func() bool {
|
|
e, err := b.kvs["deliveries"].Get(ctx, kvKey(IDOf("novox/app", head)))
|
|
var d Delivery
|
|
return err == nil && json.Unmarshal(e.Value(), &d) == nil && d.State == Merged
|
|
})
|
|
rt.noControllerAsk()
|
|
// Quiet follows: nothing asks the forge or the controller again with no event to say so.
|
|
quiet := len(rt.asked)
|
|
time.Sleep(1500 * time.Millisecond)
|
|
rt.mu.Lock()
|
|
more := len(rt.asked) - quiet
|
|
rt.mu.Unlock()
|
|
if more > 3 {
|
|
t.Fatalf("%d asks of the mesh in a quiet second and a half after the merge: something polls", more)
|
|
}
|
|
}
|
|
|
|
// onlyAsk is the key of the one ask in the gate's bucket asks, or empty.
|
|
func onlyAsk(b *testBus) string {
|
|
keys, err := b.kvs["asks"].Keys(context.Background())
|
|
if err != nil || len(keys) != 1 {
|
|
return ""
|
|
}
|
|
return keys[0]
|
|
}
|
|
|
|
// askNow is the ask under a key at this moment; false in the instant a rewrite has taken the key away and not yet written it.
|
|
func askNow(b *testBus, key string) (Ask, bool) {
|
|
e, err := b.kvs["asks"].Get(context.Background(), key)
|
|
if err != nil {
|
|
return Ask{}, false
|
|
}
|
|
a, err := DecodeAsk(e.Value())
|
|
return a, err == nil
|
|
}
|
|
|
|
func askNowOrFatal(t *testing.T, b *testBus, key string) Ask {
|
|
t.Helper()
|
|
a, ok := askNow(b, key)
|
|
if !ok {
|
|
t.Fatalf("no ask under %s to answer", key)
|
|
}
|
|
return a
|
|
}
|
|
|
|
func askAt(t *testing.T, b *testBus, key string) Ask {
|
|
t.Helper()
|
|
e, err := b.kvs["asks"].Get(context.Background(), key)
|
|
if err != nil {
|
|
t.Fatalf("the ask %s: %v", key, err)
|
|
}
|
|
a, err := DecodeAsk(e.Value())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return a
|
|
}
|
|
|
|
func mergeAt(b *testBus, group string) Merge {
|
|
e, err := b.kvs["merges"].Get(context.Background(), group)
|
|
if err != nil {
|
|
return Merge{}
|
|
}
|
|
m, _ := DecodeMerge(e.Value())
|
|
return m
|
|
}
|
|
|
|
// answer is the teller's answer to an ask, written under the ask's own key in its bucket answers.
|
|
func (b *testBus) answer(t *testing.T, key, option string) time.Time {
|
|
t.Helper()
|
|
// The warrant's id is the router id of the asking answered and the time (the teller's warrantID).
|
|
asked, err := askOf(askNowOrFatal(t, b, key))
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
raw, err := mdc.EncodeAnswer(Answer{Option: option, Warrant: asked.ID + "@2026-10-20T03:00:00Z", Words: "the operator, via telegram, chose " + option})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.kvs["mesh-delivery-teller.answers"].Put(context.Background(), key, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return time.Now()
|
|
}
|
|
|
|
// forgeSays is the forge bridge's message on FORGE, as its stream carries it.
|
|
func (b *testBus) forgeSays(t *testing.T, f forgemsg.Forge) {
|
|
t.Helper()
|
|
raw, err := forgemsg.EncodeForge(f)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
event, err := forgemsg.ForgeEvent(f.Repository, f.Kind)
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, err := b.js.Publish(context.Background(), "mesh.mod."+forgeBridgeModule+".event."+event, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// **The acceptance of one group, one Approve** (novox/hq ADR 0418 §4, ADR 0421, ADR 0441 §5, design 47 milestone 3, issues 978
|
|
// and 1022): a group of two pull requests on one branch name, in two repositories, with green checks, gets one ask of the
|
|
// operator — no label, comment or mark written by anyone — and on the Approve the members are handed to the bridge one at a time,
|
|
// the second after the first merged, each with the approved heads of both.
|
|
func TestAGroupOfTwoWithGreenChecksGetsOneAskAndOnItsApproveBothMergeOneAfterTheOtherOverARealBus(t *testing.T) {
|
|
b := newTestBus(t)
|
|
ctx := context.Background()
|
|
rt := b.start(t)
|
|
evctx, cancel := context.WithCancel(ctx)
|
|
defer cancel()
|
|
rt.deliverEvents(evctx)
|
|
rt.untilListening()
|
|
const ctlHead, catHead = "c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1c1", "d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2d2"
|
|
for _, p := range []PullEvent{pr("novox/mesh-controller", 1, ctlHead, "feat/x", "internal/a.go"),
|
|
pr("novox/mesh-catalog", 2, catHead, "feat/x", "modules/app/module.json")} {
|
|
p.BaseSHA = "a1a1a1a1a1a1"
|
|
p.Body = "Reviewed-by: lead@shanks at " + p.HeadSHA
|
|
b.publish(t, "mesh.mod.gitea.event.pull.updated", p)
|
|
}
|
|
// Both members pass at the checker, whose one verdict covers both heads: the group is asked once, no composed check is asked
|
|
// of the controller (issue 1318).
|
|
b.checkPasses(t, rt, "feat/x", "a1a1a1a1a1a1", judgedPull{"novox/mesh-controller", 1, ctlHead}, judgedPull{"novox/mesh-catalog", 2, catHead})
|
|
var key string
|
|
rt.until("the operator to be asked once for the group", 10*time.Second, func() bool { key = onlyAsk(b); return key != "" })
|
|
time.Sleep(500 * time.Millisecond)
|
|
if keys, _ := b.kvs["asks"].Keys(ctx); len(keys) != 1 {
|
|
t.Fatalf("the operator was asked %d times for one group", len(keys))
|
|
}
|
|
if g := askAt(t, b, key).Group; g != "feat/x" {
|
|
t.Fatalf("the ask is for the group %q", g)
|
|
}
|
|
if len(rt.asksOf("gitea.gitea_merge_pull_request")) != 0 || mergeAt(b, "feat/x").Group != "" {
|
|
t.Fatalf("merged before the Approve")
|
|
}
|
|
approved := b.answer(t, key, OptionApprove)
|
|
|
|
// One member is handed, with both heads approved, within seconds of the answer; the second only after the bridge says the
|
|
// first merged.
|
|
var first Merge
|
|
rt.until("the first member to be handed", eventWithin, func() bool { first = mergeAt(b, "feat/x"); return len(first.Pulls) > 0 })
|
|
if len(first.Pulls) != 1 || first.Pulls[0].State != mdc.MergeMerging || len(first.Approved) != 2 ||
|
|
first.Approved[mdc.ApprovedKey("novox/mesh-controller", 1)] != ctlHead || first.Approved[mdc.ApprovedKey("novox/mesh-catalog", 2)] != catHead {
|
|
t.Fatalf("the first merge handed is %+v", first)
|
|
}
|
|
t.Logf("the first member was handed %s after the answer", time.Since(approved))
|
|
time.Sleep(500 * time.Millisecond)
|
|
if again := mergeAt(b, "feat/x"); len(again.Pulls) != 1 {
|
|
t.Fatalf("the second member was handed before the first merged: %+v", again.Pulls)
|
|
}
|
|
one := first.Pulls[0]
|
|
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: one.Repository, Number: one.Number, Head: one.Head.Head,
|
|
Merge: strings.Repeat("a", 40), Root: "r", Trees: map[string]string{"x": "t"}})
|
|
var both Merge
|
|
rt.until("the second member to be handed", eventWithin, func() bool { both = mergeAt(b, "feat/x"); return len(both.Pulls) == 2 })
|
|
if both.Pulls[0].State != mdc.MergeMerged || both.Pulls[1].State != mdc.MergeMerging || both.Pulls[1].Repository == one.Repository {
|
|
t.Fatalf("the second merge handed is %+v", both.Pulls)
|
|
}
|
|
two := both.Pulls[1]
|
|
b.forgeSays(t, forgemsg.Forge{Kind: forgemsg.ForgeMerged, Repository: two.Repository, Number: two.Number, Head: two.Head.Head,
|
|
Merge: strings.Repeat("b", 40), Root: "r", Trees: map[string]string{"x": "t"}})
|
|
rt.until("source.main to hold both merges", eventWithin, func() bool {
|
|
e, err := b.kvs["source"].Get(ctx, mdc.SourceMainKey)
|
|
if err != nil {
|
|
return false
|
|
}
|
|
src, err := mdc.DecodeSource(e.Value())
|
|
return err == nil && src.Group == "feat/x" && len(src.Heads) == 2
|
|
})
|
|
if keys, _ := b.kvs["asks"].Keys(ctx); len(keys) != 1 {
|
|
t.Fatalf("a second ask was made: %v", keys)
|
|
}
|
|
// The forge's own word that each pull request merged follows the bridge's (its webhook), after the second was handed (the
|
|
// checker judges the members still open by the trunk the first merge moved, which this test does not play): both members
|
|
// end merged, the group with them, and the controller was never called (R6-C3).
|
|
b.publish(t, "mesh.mod.gitea.event.pull.merged", merged(one.Repository, one.Number, one.Head.Head, strings.Repeat("a", 40)))
|
|
b.publish(t, "mesh.mod.gitea.event.pull.merged", merged(two.Repository, two.Number, two.Head.Head, strings.Repeat("b", 40)))
|
|
for _, id := range []string{IDOf("novox/mesh-controller", ctlHead), IDOf("novox/mesh-catalog", catHead)} {
|
|
id := id
|
|
rt.until(id+" to be merged", eventWithin, func() bool {
|
|
e, err := b.kvs["deliveries"].Get(ctx, kvKey(id))
|
|
var d Delivery
|
|
return err == nil && json.Unmarshal(e.Value(), &d) == nil && d.State == Merged
|
|
})
|
|
}
|
|
rt.until("the group to read merged", eventWithin, func() bool {
|
|
e, err := b.kvs["groups"].Get(ctx, kvKey("feat/x"))
|
|
var g Group
|
|
return err == nil && json.Unmarshal(e.Value(), &g) == nil && g.Said == "merged"
|
|
})
|
|
rt.noControllerAsk()
|
|
}
|
|
|
|
func mustDecode[T any](t *testing.T, raw []byte) T {
|
|
t.Helper()
|
|
var v T
|
|
if err := json.Unmarshal(raw, &v); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
return v
|
|
}
|
|
|
|
// eventWithin is how long a test waits for one step of the gate's path over the real bus. On an idle machine each step
|
|
// takes milliseconds (the tests log it), but a slow, loaded build node ran the ask step in 6.69 s against the old 5 s
|
|
// (issue 1404). A failed first ask is asked again only after askAgainAfter (30 s, react.go), and 20 s deliberately does not
|
|
// cover that: a failed first ask still fails the test.
|
|
const eventWithin = 20 * time.Second
|