mesh-controller/internal/link/meshcli_test.go

package link

import (
	"context"
	"encoding/json"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/nats-io/nats.go"

	"github.com/novox/mesh-controller/internal/testbus"
)

// The node-engine's request and the answer it hands mesh-cli hold these field names; the engine's side holds the
// same list (mesh-host internal/meshcli, TestTheRequestToTheControllerKeepsItsFieldNames).
func TestTheMeshCLIRequestAndAnswerKeepTheirFieldNames(t *testing.T) {
	body, _ := json.Marshal(CLIAsked{Line: []string{"status"}, Account: "a", UID: 1})
	if got := keysIn(t, body); got != "account line uid" {
		t.Fatalf("the request's fields are %q", got)
	}
	body, _ = json.Marshal(CLIAnswer{Stdout: []byte("o"), Stderr: []byte("e"), Exit: 1, Terminal: true, Why: "w",
		Refused: "r", Cut: true})
	if got := keysIn(t, body); got != "cut exit refused stderr stdout terminal why" {
		t.Fatalf("the answer's fields are %q", got)
	}
	body, _ = json.Marshal(CLIAsked{Follow: "call-1", Account: "a", UID: 1})
	if got := keysIn(t, body); got != "account follow uid" {
		t.Fatalf("a follow's fields are %q", got)
	}
	// What a line still running answers, in the envelope every call's answer is in: `result`, then these.
	var env struct {
		Result json.RawMessage `json:"result"`
	}
	_ = json.Unmarshal(running(&Call{ID: "call-1", Seat: CLISeat, Verb: "a"}, AnswerWithin, false, ""), &env)
	if got := keysIn(t, env.Result); got != "call output running started" {
		t.Fatalf("a running answer's fields are %q", got)
	}
}

func keysIn(t *testing.T, body []byte) string {
	t.Helper()
	var m map[string]any
	if err := json.Unmarshal(body, &m); err != nil {
		t.Fatal(err)
	}
	var keys []string
	for k := range m {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	return strings.Join(keys, " ")
}

// A node's mesh-cli subject names the node, and no other subject does.
func TestTheMeshCLISubjectNamesItsNode(t *testing.T) {
	if node, ok := CLINode(CLISubject("laptop")); !ok || node != "laptop" {
		t.Fatalf("CLINode(%q) = %q %v", CLISubject("laptop"), node, ok)
	}
	for _, s := range []string{"mesh.control.laptop.report", "mesh.control..cli", "mesh.control.a.b.cli", "mesh.node.a.cli"} {
		if _, ok := CLINode(s); ok {
			t.Fatalf("%s was read as a mesh-cli subject", s)
		}
	}
}

// An answer larger than one bus message is cut to fit, and the cut is said.
func TestALargeMeshCLIAnswerIsCutAndSaysSo(t *testing.T) {
	a := CLIAnswer{Stdout: []byte(strings.Repeat("o", 200000)), Stderr: []byte(strings.Repeat("e", 1000)), Why: "the terminal"}
	body := FitCLIAnswer(a, 64<<10)
	if len(body) > 64<<10 {
		t.Fatalf("the answer is %d bytes, over the bound", len(body))
	}
	var got CLIAnswer
	if err := json.Unmarshal(body, &got); err != nil {
		t.Fatal(err)
	}
	if !got.Cut || got.Why != "the terminal" || len(got.Stdout) == 0 || string(got.Stderr) != strings.Repeat("e", 1000) {
		t.Fatalf("the cut answer is %+v", got)
	}
	if small := FitCLIAnswer(CLIAnswer{Stdout: []byte("ok")}, 64<<10); strings.Contains(string(small), `"cut"`) {
		t.Fatal("an answer that fits was said to be cut")
	}
}

// cliBus serves mesh-cli on a bus of the test's own with a call log of its own, and returns a connection to ask on.
func cliBus(t *testing.T, l *CallLog, atOnce int, handle CLIHandler) *nats.Conn {
	t.Helper()
	conn, err := nats.Connect(testbus.URL(t))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(conn.Close)
	stop, err := OverNATS{Conn: conn}.serveCLI(l, atOnce, handle, nil)
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(stop)
	return conn
}

// askCLI asks one request on a node's subject and reads the envelope.
func askCLI(t *testing.T, conn *nats.Conn, node string, asked CLIAsked) (CLIAnswer, map[string]any, string) {
	t.Helper()
	body, _ := json.Marshal(asked)
	msg, err := conn.Request(CLISubject(node), body, 5*time.Second)
	if err != nil {
		t.Fatal(err)
	}
	var env struct {
		Result json.RawMessage `json:"result"`
		Error  string          `json:"error"`
	}
	if err := json.Unmarshal(msg.Data, &env); err != nil {
		t.Fatalf("not an envelope: %s", msg.Data)
	}
	var a CLIAnswer
	var raw map[string]any
	_ = json.Unmarshal(env.Result, &a)
	_ = json.Unmarshal(env.Result, &raw)
	return a, raw, env.Error
}

// **A line that outlasts the bus's window for an answer is followed to its answer, never lost** (review of ADR
// 0272): the first answer says it is running and names its call, and following the call by the same account on
// the same node gives what the line printed once it ends. Another account, or another node, is told no such call.
func TestALongMeshCLILineIsFollowedToItsAnswer(t *testing.T) {
	was := AnswerWithin
	AnswerWithin = 50 * time.Millisecond
	t.Cleanup(func() { AnswerWithin = was })
	release := make(chan struct{})
	conn := cliBus(t, NewCallLog(), 4, func(ctx context.Context, node string, asked CLIAsked) CLIAnswer {
		<-release
		return CLIAnswer{Stdout: []byte("done on " + node), Terminal: true}
	})
	_, first, failed := askCLI(t, conn, "laptop", CLIAsked{Line: []string{"push", "laptop"}, Account: "op", UID: 1000})
	call, _ := first["call"].(string)
	if failed != "" || first["running"] != true || call == "" {
		t.Fatalf("a long line was not answered as running with its call: %v %q", first, failed)
	}
	_, still, _ := askCLI(t, conn, "laptop", CLIAsked{Follow: call, Account: "op", UID: 1000})
	if still["running"] != true {
		t.Fatalf("following a running line answered %v", still)
	}
	close(release)
	deadline := time.Now().Add(5 * time.Second)
	for {
		a, raw, failed := askCLI(t, conn, "laptop", CLIAsked{Follow: call, Account: "op", UID: 1000})
		if failed != "" {
			t.Fatalf("following answered %q", failed)
		}
		if raw["running"] != true {
			if string(a.Stdout) != "done on laptop" || !a.Terminal {
				t.Fatalf("followed to %+v", a)
			}
			break
		}
		if time.Now().After(deadline) {
			t.Fatal("the line never finished for its follower")
		}
		time.Sleep(20 * time.Millisecond)
	}
	if _, _, failed := askCLI(t, conn, "laptop", CLIAsked{Follow: call, Account: "agent", UID: 1001}); failed == "" {
		t.Fatal("another account followed the operator's line")
	}
	if _, _, failed := askCLI(t, conn, "desktop", CLIAsked{Follow: call, Account: "op", UID: 1000}); failed == "" {
		t.Fatal("another node followed the line")
	}
}

// Lines running at once are bounded: one over the bound is answered busy at once, and nothing ran.
func TestMeshCLILinesAtOnceAreBounded(t *testing.T) {
	was := AnswerWithin
	AnswerWithin = 50 * time.Millisecond
	t.Cleanup(func() { AnswerWithin = was })
	release := make(chan struct{})
	t.Cleanup(func() { close(release) })
	ran := make(chan struct{}, 4)
	conn := cliBus(t, NewCallLog(), 1, func(context.Context, string, CLIAsked) CLIAnswer {
		ran <- struct{}{}
		<-release
		return CLIAnswer{}
	})
	if _, first, _ := askCLI(t, conn, "laptop", CLIAsked{Line: []string{"status"}, Account: "op"}); first["running"] != true {
		t.Fatalf("the first line answered %v", first)
	}
	a, _, _ := askCLI(t, conn, "laptop", CLIAsked{Line: []string{"status"}, Account: "op"})
	if !strings.Contains(a.Refused, "busy") || a.Exit == 0 {
		t.Fatalf("a line over the bound answered %+v", a)
	}
	if len(ran) != 1 {
		t.Fatalf("%d lines ran, one was bound", len(ran))
	}
}

// A mesh-cli line's record keeps its first word, never the rest of its line, and its answer is never kept on the
// bus, where `calls` answers anyone who may call the seat.
func TestAMeshCLIRecordKeepsNeitherItsLineNorItsAnswerOnTheBus(t *testing.T) {
	l, a := NewCallLog(), newAnswers(t)
	writes := make(chan Call, 4)
	l.writes = writes
	asked, _ := json.Marshal(CLIAsked{Line: []string{"settings", "set", "x", `{"password":"s3cret"}`}, Account: "op"})
	l.serveCall(CLISeat, "laptop", asked, "_INBOX.node.laptop.abcdefghijklmnopqrstuv",
		func(context.Context, json.RawMessage) (any, error) {
			return CLIAnswer{Stdout: []byte("token s3cret-join")}, nil
		},
		a.respond, nil)
	_ = a.only()
	close(writes)
	for c := range writes {
		if strings.Contains(string(c.Args), "s3cret") || strings.Contains(string(c.Answer), "s3cret") {
			t.Fatalf("the bus was sent %s / %s", c.Args, c.Answer)
		}
		if !strings.Contains(string(c.Args), "settings") || !strings.Contains(string(c.Args), `"op"`) {
			t.Fatalf("the record does not say what was asked and by whom: %s", c.Args)
		}
	}
}

// `calls` answers anyone who may call the seat, agents among them: a mesh-cli line's answer is never shown there,
// even from the memory of the controller that ran it; every other call's is (review of ADR 0272).
func TestCallsNeverShowsAMeshCLILinesAnswer(t *testing.T) {
	l, a := NewCallLog(), newAnswers(t)
	asked, _ := json.Marshal(CLIAsked{Line: []string{"token", "issue", "x"}, Account: "operator"})
	l.serveCall(CLISeat, "control", asked, "_INBOX.node.control.abcdefghijklmnopqrstuv",
		func(context.Context, json.RawMessage) (any, error) {
			return CLIAnswer{Stdout: []byte("s3cret-join")}, nil
		},
		a.respond, nil)
	_ = a.only()
	recent, _ := l.Recent()
	shown, found, err := l.Shown(recent[0].ID)
	if err != nil || !found {
		t.Fatalf("the line is not shown at all: %v %v", found, err)
	}
	if strings.Contains(string(shown.Answer), "s3cret-join") {
		t.Fatalf("calls shows a mesh-cli line's answer: %s", shown.Answer)
	}
	b := newAnswers(t)
	l.serveCall("mesh-controller", "status", nil, "_INBOX.x.abcdefghijklmnopqrstuv",
		func(context.Context, json.RawMessage) (any, error) { return "all well", nil }, b.respond, nil)
	_ = b.only()
	recent, _ = l.Recent()
	if shown, _, _ := l.Shown(recent[0].ID); !strings.Contains(string(shown.Answer), "all well") {
		t.Fatalf("another call's answer is withheld: %s", shown.Answer)
	}
}