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

//go:build integration

package main

// The device process over a real bus (novox/hq ADR 0467 clause 4; the shape of mesh-mail's
// sender_integration_test.go, ADR 0436 §7): a JetStream server of the test's own, the work-queue stream
// device-asks and its consumer, the state bucket device-answers, and a fake printer. An ask published on the
// stream is answered in the state under its id and acknowledged. Compiled by the merge gate's
// `go vet -tags integration`; never run on this machine.

import (
	"context"
	"encoding/json"
	"os"
	"path/filepath"
	"strings"
	"testing"
	"time"

	"git.novox.be/novox/mesh-sdk/go/process"
	"github.com/nats-io/nats-server/v2/server"
	"github.com/nats-io/nats.go"
	"github.com/nats-io/nats.go/jetstream"

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

// printerAskedSubject is the wire subject the controller composes for the stream's one event:
// "mesh.stream." + module + ".event." + event (mesh-controller internal/broker/modstreams.go).
const printerAskedSubject = "mesh.stream.brother.event." + EventAsked

func newTestBus(t *testing.T) string {
	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() })
	return s.ClientURL()
}

// 1453: an ask on device-asks is answered under its id in device-answers by the device process, over the
// printer's pinned HTTPS and the login, and acknowledged; stopping it leaves nothing in hand.
func TestTheAdminProcessAnswersAnAskOverARealBus(t *testing.T) {
	url := newTestBus(t)
	ctx := context.Background()
	nc, err := nats.Connect(url)
	if err != nil {
		t.Fatal(err)
	}
	defer nc.Close()
	js, err := jetstream.New(nc)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: AsksStream, Subjects: []string{printerAskedSubject},
		Retention: jetstream.WorkQueuePolicy, Discard: jetstream.DiscardNew, MaxAge: 24 * time.Hour, MaxBytes: 8 << 20}); err != nil {
		t.Fatal(err)
	}
	account := process.Account{URL: url, Node: "g14", Module: "brother", User: "g14.brother." + DeviceProcessID, Password: "pw"}
	if _, err := js.CreateConsumer(ctx, AsksStream, jetstream.ConsumerConfig{Durable: account.QueueConsumerName(),
		AckPolicy: jetstream.AckExplicitPolicy, AckWait: 30 * time.Second, MaxDeliver: 3}); err != nil {
		t.Fatal(err)
	}
	kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: account.Module + "_" + AnswersState, TTL: 10 * time.Minute})
	if err != nil {
		t.Fatal(err)
	}

	f := newFakePrinter(t, "B1891", "pw")
	dir := t.TempDir()
	settings := filepath.Join(dir, "settings.json")
	os.WriteFile(settings, []byte(`{"printer":"`+f.host()+`","printer-certificate":"`+f.print()+`"}`), 0o600)
	pwFile := filepath.Join(dir, "admin-password.secret")
	os.WriteFile(pwFile, []byte("pw\n"), 0o600)
	w := &deviceWorker{
		Settings: func() (Settings, error) { return LoadSettings(settings) },
		Password: func() (string, error) { return secret.Read(pwFile) },
		Now:      time.Now,
	}
	runCtx, cancel := context.WithCancel(ctx)
	done := make(chan error, 1)
	go func() { done <- runDevice(runCtx, account, w) }()

	body, _ := json.Marshal(Ask{ID: "ask-1", Op: OpSettings, Page: "/general/other.html", At: time.Now()})
	if _, err := js.Publish(ctx, printerAskedSubject, body, jetstream.WithMsgID("ask-1")); err != nil {
		t.Fatal(err)
	}
	var answer string
	for deadline := time.Now().Add(15 * time.Second); time.Now().Before(deadline); time.Sleep(50 * time.Millisecond) {
		if e, err := kv.Get(ctx, "ask-1"); err == nil {
			answer = string(e.Value())
			break
		}
	}
	if !strings.Contains(answer, "Synthetic settings") {
		t.Fatalf("the answer under ask-1 is %q", answer)
	}
	if strings.Contains(answer, "shouldnotshow") {
		t.Errorf("a password field's value is in the answer")
	}
	time.Sleep(500 * time.Millisecond)
	cancel()
	select {
	case err := <-done:
		if err == nil {
			t.Fatal("runDevice returned nil, want ctx.Err() once its context was cancelled")
		}
	case <-time.After(10 * time.Second):
		t.Fatal("runDevice did not stop once its context was cancelled")
	}
	cons, err := js.Consumer(ctx, AsksStream, account.QueueConsumerName())
	if err != nil {
		t.Fatal(err)
	}
	info, err := cons.Info(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if info.NumPending != 0 || info.NumAckPending != 0 {
		t.Errorf("the ask was not acknowledged: %d pending, %d awaiting an acknowledgement", info.NumPending, info.NumAckPending)
	}
}