mesh-sdk/go/process/account_test.go

package process

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

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

// 1171/AC2: given an account file broker-<process id> in the controller's form {url, fingerprint, node,
// module, user, password, claims}, when the process package reads it, then it yields those fields and
// tolerates claims.
func TestReadAccountYieldsEveryFieldAndToleratesClaims(t *testing.T) {
	path := writeAccount(t, map[string]any{
		"url": "nats://novox:4222", "fingerprint": "sha256:ab", "node": "novox", "module": "mesh-mail",
		"user": "novox.mesh-mail.send", "password": "secret-pw",
		"claims": []map[string]any{{"seat": "operator-channel", "verbs": []string{"tell"}}},
	})
	a, err := ReadAccount(path)
	if err != nil {
		t.Fatalf("a well-formed account file was refused: %v", err)
	}
	if a.URL != "nats://novox:4222" || a.Fingerprint != "sha256:ab" || a.Node != "novox" ||
		a.Module != "mesh-mail" || a.User != "novox.mesh-mail.send" || a.Password != "secret-pw" {
		t.Fatalf("the account read back: %+v", a)
	}
}

// 1171/AC2: a process account carries no claims at all (heldProcessAccount); reading it is not refused for
// the field being absent.
func TestReadAccountToleratesNoClaimsAtAll(t *testing.T) {
	path := writeAccount(t, map[string]any{
		"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail",
		"user": "novox.mesh-mail.send", "password": "secret-pw",
	})
	a, err := ReadAccount(path)
	if err != nil {
		t.Fatalf("a process account with no claims was refused: %v", err)
	}
	if a.User != "novox.mesh-mail.send" {
		t.Fatalf("the account read back: %+v", a)
	}
}

// 1171/AC2: a file missing url, user or password is refused, naming the field; the password never appears
// in the error even when it is the field present and another is missing.
func TestReadAccountRefusesAFileMissingAField(t *testing.T) {
	t.Run("1945: a refused bucket stops a start only when the part cannot work without it", func(t *testing.T) {
		refusal := errors.New("nats: permissions violation for publish to $JS.API.STREAM.INFO.KV_mesh-delivery-gate_source")
		for _, c := range []struct {
			name string
			err  error
		}{{"refused", refusal}, {"not answered", context.DeadlineExceeded}} {
			var said []string
			logf := func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
			if kv := OptionalBucket(context.Background(), refusingBus{c.err}, "b", "b is not heard", logf); kv != nil {
				t.Errorf("%s: an optional bucket answered %v", c.name, kv)
			}
			NotWatched(logf, "b", c.err, "b is not heard")
			if len(said) != 2 || !strings.Contains(said[0], c.err.Error()) || !strings.Contains(said[1], "cannot be watched") {
				t.Errorf("%s: said %q", c.name, said)
			}
			if _, err := EssentialBucket(context.Background(), refusingBus{c.err}, "builder", "b", "build"); err == nil ||
				!strings.Contains(err.Error(), "grant of b") || !strings.Contains(err.Error(), "the builder cannot build") {
				t.Errorf("%s: an essential bucket: %v", c.name, err)
			}
		}
	})

	t.Run("1790: a bucket missing at start is bound when the bus says it was created, and what was written meanwhile is kept", func(t *testing.T) {
		bus := &laterBus{}
		hear := &heldSubscriber{}
		var said []string
		logf := func(f string, a ...any) { said = append(said, fmt.Sprintf(f, a...)) }
		b := AwaitBucket(context.Background(), bus, hear, "m_hearing", "the health tool cannot say it", logf)
		if hear.subject != "$JS.EVENT.ADVISORY.STREAM.CREATED.KV_m_hearing" {
			t.Fatalf("heard %q", hear.subject)
		}
		ready := make(chan jetstream.KeyValue, 1)
		b.Ready(func(kv jetstream.KeyValue) { ready <- kv })
		for _, v := range []string{"one", "two"} {
			if _, err := b.Put(context.Background(), "dead-letters", []byte(v)); !errors.Is(err, ErrNotYet) {
				t.Fatalf("a write while missing: %v", err)
			}
		}
		if b.KV() != nil || len(ready) > 0 {
			t.Fatal("bound before the bucket exists")
		}
		bus.create()
		hear.deliver()
		select {
		case <-ready:
		case <-time.After(5 * time.Second):
			t.Fatal("not bound within 5s of the bus saying the bucket was created")
		}
		if got := bus.kv.puts; strings.Join(got, ",") != "dead-letters=two" {
			t.Errorf("written when it appeared: %v, want the latest value only", got)
		}
		if _, err := b.Put(context.Background(), "dead-letters", []byte("three")); err != nil {
			t.Errorf("a write once bound: %v", err)
		}
		if len(said) != 2 || !strings.Contains(said[0], "cannot be read yet") || !strings.Contains(said[1], "appeared") {
			t.Errorf("said %q", said)
		}
		// The bus refuses the subscription to the creation: said once, loudly; a reconnect hears it again and binds.
		refusedBus, refusing := &laterBus{}, &heldSubscriber{refused: true}
		var loud []string
		r := AwaitBucket(context.Background(), refusedBus, refusing, "m_reasks", "x",
			func(f string, a ...any) { loud = append(loud, fmt.Sprintf(f, a...)) })
		bound := make(chan struct{})
		r.Ready(func(jetstream.KeyValue) { close(bound) })
		if len(loud) != 2 || !strings.Contains(loud[0], "cannot hear the creation of m_reasks") ||
			!strings.Contains(loud[0], "permissions violation") {
			t.Fatalf("a refused subscription was said as %q", loud)
		}
		refusing.mu.Lock()
		refusing.refused = false
		refusing.mu.Unlock()
		refusedBus.create()
		refusing.status <- nats.CONNECTED
		select {
		case <-bound:
		case <-time.After(5 * time.Second):
			t.Fatal("not bound after the connection connected again")
		}
		if refusing.subscribed != 2 {
			t.Errorf("subscribed %d times, want again at the reconnect", refusing.subscribed)
		}

		// Without the advisory heard, the next use binds it.
		quiet := &laterBus{}
		c := AwaitBucket(context.Background(), quiet, nil, "m_hearing", "x", nil)
		quiet.create()
		if c.KV() == nil {
			t.Error("a use after the bucket appeared did not bind it")
		}
	})

	for _, tc := range []struct {
		field  string
		values map[string]any
	}{
		{"url", map[string]any{"node": "novox", "module": "mesh-mail", "user": "novox.mesh-mail.send", "password": "secret-pw"}},
		{"user", map[string]any{"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail", "password": "secret-pw"}},
		{"password", map[string]any{"url": "nats://novox:4222", "node": "novox", "module": "mesh-mail", "user": "novox.mesh-mail.send"}},
	} {
		t.Run(tc.field, func(t *testing.T) {
			path := writeAccount(t, tc.values)
			_, err := ReadAccount(path)
			if err == nil {
				t.Fatalf("an account file missing %s was accepted", tc.field)
			}
			if !strings.Contains(err.Error(), tc.field) {
				t.Errorf("the error %q does not name the missing field %s", err.Error(), tc.field)
			}
			if strings.Contains(err.Error(), "secret-pw") {
				t.Errorf("the password appears in the error: %q", err.Error())
			}
		})
	}
}

func writeAccount(t *testing.T, fields map[string]any) string {
	t.Helper()
	b, err := json.Marshal(fields)
	if err != nil {
		t.Fatal(err)
	}
	path := filepath.Join(t.TempDir(), "broker-send")
	if err := os.WriteFile(path, b, 0o600); err != nil {
		t.Fatal(err)
	}
	return path
}

// refusingBus refuses every bucket with err.
type refusingBus struct{ err error }

func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }

// laterBus has no bucket until create, and then one.
type laterBus struct {
	mu sync.Mutex
	kv *keptKV
}

func (l *laterBus) create() { l.mu.Lock(); l.kv = &keptKV{}; l.mu.Unlock() }

func (l *laterBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	if l.kv == nil {
		return nil, jetstream.ErrBucketNotFound
	}
	return l.kv, nil
}

// keptKV records each put as key=value.
type keptKV struct {
	jetstream.KeyValue
	puts []string
}

func (k *keptKV) Put(_ context.Context, key string, value []byte) (uint64, error) {
	k.puts = append(k.puts, key+"="+string(value))
	return uint64(len(k.puts)), nil
}

// heldSubscriber keeps the last subscription made, for the test to deliver to; refused, the bus refuses it as a
// *nats.Conn says it (the connection's last error after a round trip). status stands for the connection's CONNECTED.
type heldSubscriber struct {
	mu         sync.Mutex
	subject    string
	handle     nats.MsgHandler
	subscribed int
	refused    bool
	last       error
	status     chan nats.Status
}

func (h *heldSubscriber) Subscribe(subject string, handle nats.MsgHandler) (*nats.Subscription, error) {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.subject, h.handle = subject, handle
	h.subscribed++
	if h.refused {
		h.last = fmt.Errorf("%w: Permissions Violation for Subscription to %q", nats.ErrPermissionViolation, subject)
	}
	return nil, nil
}

func (h *heldSubscriber) FlushTimeout(time.Duration) error { return nil }

func (h *heldSubscriber) LastError() error { h.mu.Lock(); defer h.mu.Unlock(); return h.last }

func (h *heldSubscriber) StatusChanged(...nats.Status) chan nats.Status {
	h.mu.Lock()
	defer h.mu.Unlock()
	h.status = make(chan nats.Status, 1)
	return h.status
}

func (h *heldSubscriber) RemoveStatusListener(chan nats.Status) {}

func (h *heldSubscriber) deliver() {
	h.mu.Lock()
	handle, subject := h.handle, h.subject
	h.mu.Unlock()
	handle(&nats.Msg{Subject: subject})
}