Test the forge-account provider whole against a real bus and a fake forge (issue 955 F5)
This commit is contained in:
@@ -0,0 +1,335 @@
|
||||
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"
|
||||
)
|
||||
|
||||
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-controller.assignments": "mesh-controller_assignments"} {
|
||||
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-controller.assignments", "g14.ssh-client", map[string]any{"node": "g14", "module": "ssh-client",
|
||||
"provisioning": map[string]any{"provisions": []map[string]any{{"provision": "forge-account", "provider": "novox",
|
||||
"provider-module": "gitea", "as": "mesh_g14_ssh_client"}}}})
|
||||
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"] != "agent-g14" || e.Values["ssh_host"] != "novox.internal" ||
|
||||
e.Values["known_host"] != "[novox.internal]:222 ssh-ed25519 AAAAC3HOSTKEY" ||
|
||||
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("agent-g14"); 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["agent-g14"]
|
||||
return len(k) == 1 && strings.HasPrefix(k[0]["key"].(string), pubB)
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user