mesh-catalog/modules/brother/cmd/brother/rig_test.go

package main

import (
	"encoding/json"
	"fmt"
	"github.com/nats-io/nats.go/jetstream"
	"net/http"
	"os"
	"path/filepath"
	"sync"
	"testing"
	"time"

	stdio "git.novox.be/novox/mesh-sdk/go"

	"brother/cmd/brother/internal/secret"
)

// fixedNow is the time the tools and the device worker of a rig read.
var fixedNow = time.Date(2026, 10, 20, 19, 5, 9, 0, time.Local)

// memAnswers is the state device-answers, in memory.
type memAnswers struct {
	mu sync.Mutex
	m  map[string]json.RawMessage
	// fail, when set, is the error every Create answers.
	fail error
}

func (a *memAnswers) Create(key string, v any) (uint64, error) {
	a.mu.Lock()
	defer a.mu.Unlock()
	if a.fail != nil {
		return 0, a.fail
	}
	if _, ok := a.m[key]; ok {
		return 0, fmt.Errorf("%s: %w", key, stdio.ErrChanged)
	}
	if a.m == nil {
		a.m = map[string]json.RawMessage{}
	}
	raw, _ := json.Marshal(v)
	if len(raw) > StateMaxValue {
		return 0, fmt.Errorf("brother's state device-answers, key %s: nats: %w", key, &jetstream.APIError{Code: 400, ErrorCode: jetstream.ErrorCode(10054), Description: "message size exceeds maximum allowed"})
	}
	a.m[key] = raw
	return 1, nil
}

func (a *memAnswers) get(key string) json.RawMessage {
	a.mu.Lock()
	defer a.mu.Unlock()
	return a.m[key]
}

// fakeAskBus is the bus between the tools and the device process: an ask published on it is handed to the
// worker, as the work queue hands it to the device process, and the worker's answers are what the tools read.
type fakeAskBus struct {
	worker    *deviceWorker // nil: nobody consumes the queue
	answers   *memAnswers
	mu        sync.Mutex
	published []Ask
}

func (b *fakeAskBus) Publish(id string, body any) error {
	raw, _ := json.Marshal(body)
	if len(raw) > 1<<20 {
		return fmt.Errorf("a message of %d bytes is over the bus's 1 MiB", len(raw))
	}
	var ask Ask
	json.Unmarshal(raw, &ask)
	b.mu.Lock()
	b.published = append(b.published, ask)
	b.mu.Unlock()
	if b.worker != nil {
		go b.worker.Handle(raw, nil)
	}
	return nil
}

func (b *fakeAskBus) Answer(id string) (json.RawMessage, error) { return b.answers.get(id), nil }

func (b *fakeAskBus) asks() int {
	b.mu.Lock()
	defer b.mu.Unlock()
	return len(b.published)
}

func (b *fakeAskBus) ops() []string {
	b.mu.Lock()
	defer b.mu.Unlock()
	var o []string
	for _, a := range b.published {
		o = append(o, a.Op)
	}
	return o
}

// printerRig is the tools, the bus and the device worker over a settings file, with the secret in a file the way
// the mesh writes it (a trailing newline included; an empty password writes none) and the bucket in memory.
type printerRig struct {
	tl      []stdio.Tool
	run     func(args map[string]any) (any, error) // printer.settings
	bus     *fakeAskBus
	worker  *deviceWorker
	answers *memAnswers
	objects *memObjects
	said    *[]said
	mu      *sync.Mutex
}

// call runs a tool by name.
func (r *printerRig) call(t *testing.T, name string, args map[string]any) (any, error) {
	t.Helper()
	return tool(t, r.tl, name).Run(args)
}

func newRig(t *testing.T, settings, password string) *printerRig {
	t.Helper()
	secretFile := filepath.Join(t.TempDir(), "admin-password.secret")
	if password != "" {
		if err := os.WriteFile(secretFile, []byte(password+"\n"), 0o600); err != nil {
			t.Fatal(err)
		}
	}
	file := settingsFile(t, settings)
	answers, objects := &memAnswers{}, newMemObjects()
	var out []said
	var mu sync.Mutex
	w := &deviceWorker{
		Settings: func() (Settings, error) { return LoadSettings(file) },
		Password: func() (string, error) { return secret.Read(secretFile) },
		Objects:  func() (Objects, error) { return objects, nil },
		Emit: func(body map[string]any, id string) error {
			mu.Lock()
			defer mu.Unlock()
			_, err := objects.Get(t.Context(), body["object"].(string))
			out = append(out, said{body, id, err == nil})
			return nil
		},
		Answers: answers,
		Node:    "shanks",
		Now:     func() time.Time { return fixedNow },
	}
	bus := &fakeAskBus{worker: w, answers: answers}
	old, oldWait, oldScan, oldPoll := askBus, answerWait, scanWait, answerPoll
	askBus, answerWait, scanWait, answerPoll = bus, 5*time.Second, 5*time.Second, 5*time.Millisecond
	oldFetch := fetchClient
	fetchClient = &http.Client{Transport: objects}
	t.Cleanup(func() {
		askBus, answerWait, scanWait, answerPoll = old, oldWait, oldScan, oldPoll
		fetchClient = oldFetch
	})
	tl := tools(file, func() time.Time { return fixedNow })
	return &printerRig{tl: tl, run: tool(t, tl, "printer.settings").Run, bus: bus, worker: w, answers: answers, objects: objects, said: &out, mu: &mu}
}

func printerSettingsJSON(f *fakePrinter) string {
	return `{"printer":"` + f.host() + `","printer-certificate":"` + f.print() + `"}`
}

func printerTools(t *testing.T, f *fakePrinter, password string) (settings string, run func(args map[string]any) (any, error)) {
	t.Helper()
	js := printerSettingsJSON(f)
	return js, newRig(t, js, password).run
}