mesh-catalog/modules/gitea/cmd/forge-account-provider/bus_test.go

package main

// The provider whole, against a real bus (issue 955 F5): a JetStream server is linked into the test and started for
// it alone, and the provider is this test binary started again as the bundle, speaking MCP on its standard streams
// to a runtime played here, which does for the bundle's state what the tool runner does: each state verb and each
// watch on the bucket the controller makes from the manifest. The forge is a fake (an httptest server), and the
// forge's own sshd is a program named ssh-keyscan on the bundle's PATH.
//
// It starts a server and a child process, so it is left to the merge check: run the unit tests with -short.

import (
	"bufio"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net/http/httptest"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

	"git.novox.be/novox/mesh-sdk/go/provider"
	"git.novox.be/novox/mesh-sdk/go/provisions"
)

const bundleVar = "MESH_FORGE_ACCOUNT_TEST_BUNDLE"

func TestMain(m *testing.M) {
	if os.Getenv(bundleVar) == "1" {
		main()
		os.Exit(0)
	}
	os.Exit(m.Run())
}

type runtime struct {
	t   *testing.T
	js  jetstream.JetStream
	kvs map[string]jetstream.KeyValue
	in  io.WriteCloser
	mu  sync.Mutex
	seq int
	ctx context.Context
	// answers to what the runtime asked the bundle, by id.
	wait map[string]chan json.RawMessage
}

func startBus(t *testing.T) (*runtime, context.CancelFunc) {
	t.Helper()
	s, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: server.RANDOM_PORT, JetStream: true,
		StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
	if err != nil {
		t.Fatal(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)
	}
	ctx, cancel := context.WithCancel(context.Background())
	rt := &runtime{t: t, js: js, kvs: map[string]jetstream.KeyValue{}, ctx: ctx, wait: map[string]chan json.RawMessage{}}
	// The controller's job: a bucket per state the module declares (bound) or reads (the others).
	for state, bucket := range map[string]string{"bound": "gitea_bound", "ssh-client.keys": "ssh-client_keys",
		"mesh-provisioner.plans": "mesh-provisioner_plans", "mesh-provisioner.provisions": "mesh-provisioner_provisions"} {
		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 1})
		if err != nil {
			t.Fatal(err)
		}
		rt.kvs[state] = kv
	}
	return rt, cancel
}

type rpc struct {
	ID     json.RawMessage `json:"id,omitempty"`
	Method string          `json:"method,omitempty"`
	Params json.RawMessage `json:"params,omitempty"`
	Result json.RawMessage `json:"result,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) start(env []string) {
	r.t.Helper()
	cmd := exec.Command(os.Args[0], "-test.run=^$")
	cmd.Env = append(os.Environ(), append(env, bundleVar+"=1")...)
	in, err := cmd.StdinPipe()
	if err != nil {
		r.t.Fatal(err)
	}
	out, err := cmd.StdoutPipe()
	if err != nil {
		r.t.Fatal(err)
	}
	cmd.Stderr = testWriter{r.t}
	if err := cmd.Start(); err != nil {
		r.t.Fatal(err)
	}
	r.in = in
	go func() {
		sc := bufio.NewScanner(out)
		sc.Buffer(make([]byte, 1<<20), 1<<24)
		for sc.Scan() {
			var m rpc
			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 <- m.Result
				}
				continue
			}
			go r.answer(m)
		}
	}()
	r.t.Cleanup(func() { _ = in.Close(); _ = cmd.Wait() })
}

type testWriter struct{ t *testing.T }

func (w testWriter) Write(p []byte) (int, error) { w.t.Logf("bundle: %s", p); return len(p), nil }

// push tells the bundle one change of a state it watches, and waits until it has taken it.
func (r *runtime) push(watch string, c map[string]any) {
	r.mu.Lock()
	r.seq++
	id := "runtime-" + strconv.Itoa(r.seq)
	ch := make(chan json.RawMessage, 1)
	r.wait[id] = ch
	r.mu.Unlock()
	r.say(map[string]any{"id": id, "method": "mesh/state", "params": map[string]any{"watch": watch, "change": c}})
	select {
	case <-ch:
	case <-time.After(20 * time.Second):
		r.t.Errorf("the bundle did not take a change of %v in 20 s", c["key"])
	}
}

func (r *runtime) answer(m rpc) {
	var p struct {
		State string          `json:"state"`
		Key   string          `json:"key"`
		Value json.RawMessage `json:"value"`
	}
	_ = json.Unmarshal(m.Params, &p)
	result, err := func() (any, error) {
		kv := r.kvs[p.State]
		if kv == nil {
			return nil, fmt.Errorf("the state %q is not one this bundle declares or reads", p.State)
		}
		switch m.Method {
		case "mesh/state.get":
			e, err := kv.Get(r.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.put":
			rev, err := kv.Put(r.ctx, p.Key, p.Value)
			return map[string]any{"revision": rev}, err
		case "mesh/state.delete":
			err := kv.Delete(r.ctx, p.Key)
			return map[string]any{}, err
		case "mesh/state.watch":
			w, err := kv.WatchAll(r.ctx)
			if err != nil {
				return nil, err
			}
			id := strings.Trim(string(m.ID), `"`)
			for {
				var e jetstream.KeyValueEntry
				select {
				case e = <-w.Updates():
				case <-r.ctx.Done():
					return nil, r.ctx.Err()
				}
				if e == nil { // the current values are all handed
					break
				}
				if e.Operation() == jetstream.KeyValuePut {
					r.push(id, map[string]any{"state": p.State, "key": e.Key(), "op": "put", "revision": e.Revision(),
						"current": true, "value": json.RawMessage(e.Value())})
				}
			}
			go func() {
				for e := range w.Updates() {
					change := map[string]any{"state": p.State, "key": e.Key(), "op": "put", "revision": e.Revision(), "current": false,
						"value": json.RawMessage(e.Value())}
					if e.Operation() != jetstream.KeyValuePut {
						change["op"] = "delete"
						delete(change, "value")
					}
					r.push(id, change)
				}
			}()
			return map[string]any{}, nil
		}
		return nil, fmt.Errorf("the runtime offers no %s to this bundle", m.Method)
	}()
	if err != nil {
		r.say(map[string]any{"id": m.ID, "error": map[string]any{"code": -32000, "message": err.Error()}})
		return
	}
	r.say(map[string]any{"id": m.ID, "result": result})
}

func (r *runtime) put(state, key string, v any) {
	r.t.Helper()
	raw, _ := json.Marshal(v)
	if _, err := r.kvs[state].Put(r.ctx, key, raw); err != nil {
		r.t.Fatal(err)
	}
}

func (r *runtime) boundEntry(key string) (provider.Entry, bool) {
	e, err := r.kvs["bound"].Get(r.ctx, key)
	if err != nil {
		return provider.Entry{}, false
	}
	var out provider.Entry
	_ = json.Unmarshal(e.Value(), &out)
	return out, true
}

func eventually(t *testing.T, what string, ok func() bool) {
	t.Helper()
	for deadline := time.Now().Add(40 * time.Second); time.Now().Before(deadline); time.Sleep(100 * time.Millisecond) {
		if ok() {
			return
		}
	}
	t.Fatalf("not within 40 s: %s", what)
}

func TestTheProviderWholeAnswersAMachineThatReportedItsKey(t *testing.T) {
	if testing.Short() {
		t.Skip("starts a bus and a process: left to the merge check")
	}
	forge := newForge()
	srv := httptest.NewServer(forge)
	t.Cleanup(srv.Close)
	dir := t.TempDir()
	bin := filepath.Join(dir, "bin")
	if err := os.Mkdir(bin, 0o755); err != nil {
		t.Fatal(err)
	}
	files := map[string]string{
		"pw":              "x\n",
		"host":            "novox.internal\n",
		"origins":         "https://git.novox.be\n",
		"bin/ssh-keyscan": "#!/bin/sh\necho '[127.0.0.1]:222 ssh-ed25519 AAAAC3HOSTKEY'\n",
	}
	for name, c := range files {
		mode := os.FileMode(0o600)
		if strings.HasPrefix(name, "bin/") {
			mode = 0o755
		}
		if err := os.WriteFile(filepath.Join(dir, name), []byte(c), mode); err != nil {
			t.Fatal(err)
		}
	}
	rt, cancel := startBus(t)
	t.Cleanup(cancel)
	rt.put("mesh-provisioner.plans", "g14.ssh-client", map[string]any{"v": 1, "node": "g14", "module": "ssh-client",
		"plan": map[string]any{"provisions": []map[string]any{{"provision": "forge-account", "provider": "novox",
			"provider-module": "gitea", "as": "mesh_g14_ssh_client"}}}})
	rt.put("mesh-provisioner.provisions", "g14.ssh-client.forge-account",
		json.RawMessage(provisionKey("g14", "ssh-client", provisions.Asked, nil)))
	rt.put("ssh-client.keys", "g14.nox", map[string]string{"node": "g14", "account": "nox", "kind": "machine",
		"public": pubA + " nox@g14 (mesh)", "fingerprint": "SHA256:x", "at": "2026-10-16T10:00:00Z"})
	rt.start([]string{"MESH_NODE=novox", "MESH_MODULE=gitea", "MESH_SERVED_MODULE=gitea",
		"PATH=" + bin + ":" + os.Getenv("PATH"),
		"MESH_GITEA_URL=" + srv.URL, "MESH_GITEA_ADMIN_USER=mesh-admin",
		"MESH_GITEA_ADMIN_PASSWORD_FILE=" + filepath.Join(dir, "pw"),
		"MESH_GITEA_ORIGINS_FILE=" + filepath.Join(dir, "origins"),
		"MESH_FORGE_HOST_FILE=" + filepath.Join(dir, "host"), "MESH_GITEA_SSH_PORT=222"})

	var e provider.Entry
	eventually(t, "the bound entry of g14's ssh-client", func() bool {
		var ok bool
		e, ok = rt.boundEntry("g14.ssh-client.forge-account")
		return ok
	})
	if e.Values["user"] != "nox-agent" || e.Values["ssh_host"] != "novox.internal" ||
		e.Values["known_host"] != "[novox.internal]:222 ssh-ed25519 AAAAC3HOSTKEY" || e.Values["host_key"] != "ssh-ed25519 AAAAC3HOSTKEY" || e.Values["ssh_alias"] != "mesh-forge" ||
		e.Values["ssh_base"] != "ssh://git@novox.internal:222/" || e.Values["https_base"] != "https://git.novox.be/" {
		t.Fatalf("values = %v", e.Values)
	}
	if got := forge.titles("nox-agent"); len(got) != 1 || got[0] != "mesh:g14.nox" {
		t.Fatalf("registered keys = %v", got)
	}

	// A key that changes is answered without waiting for the next pass.
	rt.put("ssh-client.keys", "g14.nox", map[string]string{"node": "g14", "account": "nox", "kind": "machine",
		"public": pubB + " nox@g14 (mesh)", "fingerprint": "SHA256:y", "at": "2026-10-16T11:00:00Z"})
	eventually(t, "the new key registered and the old one gone", func() bool {
		forge.mu.Lock()
		defer forge.mu.Unlock()
		k := forge.held["nox-agent"]
		return len(k) == 1 && strings.HasPrefix(k[0]["key"].(string), pubB)
	})
}