mesh/merge-gate head a4e7f480b3b0 merged into trunk 219bedbdd2b3 beside mesh-catalog#337@d2992f81 +1 more — pass, with a note: builds mesh-sdk (+41 depende…
mesh/repo-check head a4e7f480b3b0 merged into trunk 219bedbdd2b3 beside mesh-catalog#337@d2992f81 +1 more — pass: its merge-check.sh passed, each part in i…
mesh/delivery delivered
mesh/delivery-group group feat/providers-after-switch-on delivered: every member is delivered
804 lines
32 KiB
Go
804 lines
32 KiB
Go
package provider
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"encoding/base64"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"reflect"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"testing"
|
|
"time"
|
|
|
|
stdio "git.novox.be/novox/mesh-sdk/go"
|
|
)
|
|
|
|
// standIn is a stand-in bus: buckets of keys and values, and watches that hand the current values first and then
|
|
// every change, as the bus's key-value watch does.
|
|
type standIn struct {
|
|
mu sync.Mutex
|
|
buckets map[string]map[string][]byte
|
|
watchers map[string][]func(string, []byte)
|
|
}
|
|
|
|
func newStandIn() *standIn {
|
|
return &standIn{buckets: map[string]map[string][]byte{}, watchers: map[string][]func(string, []byte){}}
|
|
}
|
|
|
|
func (s *standIn) Watch(ctx context.Context, bucket string, handle func(key string, value []byte)) error {
|
|
s.mu.Lock()
|
|
current := map[string][]byte{}
|
|
for k, v := range s.buckets[bucket] {
|
|
current[k] = v
|
|
}
|
|
s.watchers[bucket] = append(s.watchers[bucket], handle)
|
|
s.mu.Unlock()
|
|
for _, k := range sortedKeys(current) {
|
|
handle(k, current[k])
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *standIn) Put(_ context.Context, bucket, key string, value []byte) error {
|
|
s.mu.Lock()
|
|
if s.buckets[bucket] == nil {
|
|
s.buckets[bucket] = map[string][]byte{}
|
|
}
|
|
s.buckets[bucket][key] = append([]byte(nil), value...)
|
|
ws := slices.Clone(s.watchers[bucket])
|
|
s.mu.Unlock()
|
|
for _, w := range ws {
|
|
w(key, value)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func (s *standIn) Delete(_ context.Context, bucket, key string) error {
|
|
s.mu.Lock()
|
|
delete(s.buckets[bucket], key)
|
|
ws := slices.Clone(s.watchers[bucket])
|
|
s.mu.Unlock()
|
|
for _, w := range ws {
|
|
w(key, nil)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// DeleteIf removes the key only while it still holds what the caller read.
|
|
func (s *standIn) DeleteIf(ctx context.Context, bucket, key string, held []byte) error {
|
|
s.mu.Lock()
|
|
cur := s.buckets[bucket][key]
|
|
s.mu.Unlock()
|
|
if !bytes.Equal(cur, held) {
|
|
return fmt.Errorf("%s: %w", key, stdio.ErrChanged)
|
|
}
|
|
return s.Delete(ctx, bucket, key)
|
|
}
|
|
|
|
func (s *standIn) get(bucket, key string) []byte {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return s.buckets[bucket][key]
|
|
}
|
|
|
|
func (s *standIn) keys(bucket string) []string {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
return sortedKeys(s.buckets[bucket])
|
|
}
|
|
|
|
// everything is every value the bus holds, for a test that no secret is in a bucket.
|
|
func (s *standIn) everything() []byte {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
var all []byte
|
|
for _, b := range s.buckets {
|
|
for k, v := range b {
|
|
all = append(all, k...)
|
|
all = append(all, v...)
|
|
}
|
|
}
|
|
return all
|
|
}
|
|
|
|
// vault is a stand-in vault: a value per name, or nothing yet.
|
|
type vault struct {
|
|
mu sync.Mutex
|
|
values map[string]string
|
|
asked []string
|
|
}
|
|
|
|
func (v *vault) Value(_ context.Context, name string) ([]byte, error) {
|
|
v.mu.Lock()
|
|
defer v.mu.Unlock()
|
|
v.asked = append(v.asked, name)
|
|
val, ok := v.values[name]
|
|
if !ok {
|
|
return nil, fmt.Errorf("the vault holds no %s yet", name)
|
|
}
|
|
return []byte(val), nil
|
|
}
|
|
|
|
// maker records what the provider's adapter was asked to make, with the credential it was handed.
|
|
type maker struct {
|
|
mu sync.Mutex
|
|
made map[string]string
|
|
// contributed is what each consumer's entry carried of its plan's contribution.
|
|
contributed map[string]map[string]any
|
|
// as is the identity each consumer's entry was made under.
|
|
as map[string]string
|
|
gone []string
|
|
fail error
|
|
calls int
|
|
}
|
|
|
|
func (m *maker) Make(_ context.Context, c Consumer) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.calls++
|
|
if m.fail != nil {
|
|
return m.fail
|
|
}
|
|
if m.made == nil {
|
|
m.made = map[string]string{}
|
|
}
|
|
m.made[c.Entry.Key()] = string(c.Credential)
|
|
if m.contributed == nil {
|
|
m.contributed = map[string]map[string]any{}
|
|
}
|
|
m.contributed[c.Entry.Key()] = c.Entry.Contributes
|
|
if m.as == nil {
|
|
m.as = map[string]string{}
|
|
}
|
|
m.as[c.Entry.Key()] = c.Entry.As
|
|
return nil
|
|
}
|
|
|
|
func (m *maker) Gone(_ context.Context, e Entry) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
m.gone = append(m.gone, e.Key())
|
|
return nil
|
|
}
|
|
|
|
// assign puts a consumer's assignment with its plan's provisions into the stand-in mirror.
|
|
func assign(t *testing.T, bus *standIn, node, module string, provisions ...map[string]string) {
|
|
t.Helper()
|
|
ps := []map[string]string{}
|
|
ps = append(ps, provisions...)
|
|
raw, err := json.Marshal(map[string]any{"node": node, "module": module, "assigned": "2026-10-15T00:00:00Z",
|
|
"provisioning": map[string]any{"templates": map[string]string{"source": "x", "digest": "y"},
|
|
"provisions": ps}})
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if err := bus.Put(context.Background(), AssignmentsBucket, node+"."+module, raw); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func bind(provision, local, provider, providerModule string) map[string]string {
|
|
b := map[string]string{"provision": provision, "provider": provider, "provider-module": providerModule}
|
|
if local != "" {
|
|
b["local"] = local
|
|
}
|
|
return b
|
|
}
|
|
|
|
func entryOf(t *testing.T, bus *standIn, key string) Entry {
|
|
t.Helper()
|
|
raw := bus.get(BucketOf("postgres"), key)
|
|
if raw == nil {
|
|
t.Fatalf("no entry %s in %s; it holds %v", key, BucketOf("postgres"), bus.keys(BucketOf("postgres")))
|
|
}
|
|
var e Entry
|
|
if err := json.Unmarshal(raw, &e); err != nil {
|
|
t.Fatalf("entry %s: %v", key, err)
|
|
}
|
|
return e
|
|
}
|
|
|
|
// postgres is a provider of postgres-database on novox with a login per consumer (ADR 0048, ADR 0158 §11), its
|
|
// served values derived from each consumer's identity (ADR 0202 §1).
|
|
func postgres(bus *standIn, v *vault, m *maker) *Provider {
|
|
return New(Options{
|
|
Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Gone: m.Gone,
|
|
Offers: map[string]Offer{
|
|
"postgres-database": {Credential: PerConsumer, Serves: map[string]any{
|
|
"host": "postgres.novox.mesh", "port": 5432, "database": "${consumer:as}"}},
|
|
},
|
|
Logf: func(string, ...any) {},
|
|
})
|
|
}
|
|
|
|
// A provider answers each consumer that binds it with an entry of its own in its own bucket: composed from its
|
|
// served values and the consumer's identity, naming the credential the vault keeps for that consumer, and never
|
|
// the credential itself; its adapter is handed each consumer's own value read from the vault.
|
|
func TestEachConsumerGetsAnEntryOfItsOwnAndTheVaultsValue(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{
|
|
"pair/novox/postgres-database/ace/letta": "letta-secret-1",
|
|
"pair/novox/postgres-database/g14/umami": "umami-secret-2",
|
|
}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
|
p := postgres(bus, v, m)
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"ace.letta.postgres-database",
|
|
"g14.umami.postgres-database"}) {
|
|
t.Fatalf("entries %v", got)
|
|
}
|
|
e := entryOf(t, bus, "ace.letta.postgres-database")
|
|
if e.Provider != "novox" || e.ProviderModule != "postgres" || e.Consumer != "ace" || e.Module != "letta" ||
|
|
e.As != "mesh_ace_letta" || e.User != "mesh_ace_letta" ||
|
|
e.VaultName != "pair/novox/postgres-database/ace/letta" || e.Refused != "" {
|
|
t.Fatalf("entry %+v", e)
|
|
}
|
|
if e.Values["database"] != "mesh_ace_letta" || e.Values["host"] != "postgres.novox.mesh" ||
|
|
e.Values["port"] != float64(5432) {
|
|
t.Fatalf("values %v", e.Values)
|
|
}
|
|
if m.made["ace.letta.postgres-database"] != "letta-secret-1" ||
|
|
m.made["g14.umami.postgres-database"] != "umami-secret-2" {
|
|
t.Fatalf("made %v", m.made)
|
|
}
|
|
if bytes.Contains(bus.everything(), []byte("secret-")) {
|
|
t.Fatal("a credential is in a bucket")
|
|
}
|
|
for _, o := range out {
|
|
if o.Status != Answered {
|
|
t.Fatalf("outcome %+v", o)
|
|
}
|
|
}
|
|
}
|
|
|
|
// A provider whose offer shares one credential (ADR 0158) names the vault's one value for the provision, and its
|
|
// one user name, to every consumer.
|
|
func TestASharedCredentialIsTheVaultsOneValueAndTheProvidersOneUser(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"provision/novox/sonarr-api": "one-value"}}
|
|
assign(t, bus, "ace", "overseerr", bind("sonarr-api", "", "novox", "sonarr"))
|
|
assign(t, bus, "g14", "homepage", bind("sonarr-api", "", "novox", "sonarr"))
|
|
p := New(Options{Node: "novox", Module: "sonarr", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
|
|
Offers: map[string]Offer{"sonarr-api": {Credential: Shared, User: "sonarr",
|
|
Serves: map[string]any{"url": "http://sonarr.novox.mesh:8989"}}}})
|
|
if _, err := p.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
for _, key := range []string{"ace.overseerr.sonarr-api", "g14.homepage.sonarr-api"} {
|
|
var e Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("sonarr"), key), &e); err != nil {
|
|
t.Fatalf("%s: %v", key, err)
|
|
}
|
|
if e.VaultName != "provision/novox/sonarr-api" || e.User != "sonarr" || e.Values["url"] == nil {
|
|
t.Fatalf("%s: %+v", key, e)
|
|
}
|
|
if m.made[key] != "one-value" {
|
|
t.Fatalf("%s made with %q", key, m.made[key])
|
|
}
|
|
}
|
|
if bytes.Contains(bus.everything(), []byte("one-value")) {
|
|
t.Fatal("the shared credential is in a bucket")
|
|
}
|
|
}
|
|
|
|
// A consumer that keeps several holders of one provision is served no derived value: its entry refuses it, naming
|
|
// both ends, so its client can report it (ADR 0349 clause 12, ADR 0202 §10); nothing is made for it.
|
|
func TestAMultiHolderConsumerIsRefusedADerivedValueNamingBothEnds(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta/archive": "x-secret"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "archive", "novox", "postgres"))
|
|
out, err := postgres(bus, v, m).Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
e := entryOf(t, bus, "ace.letta.postgres-database.archive")
|
|
for _, says := range []string{"letta on ace", "postgres on novox", "postgres-database", "archive"} {
|
|
if !strings.Contains(e.Refused, says) {
|
|
t.Fatalf("the refusal does not name %q: %s", says, e.Refused)
|
|
}
|
|
}
|
|
if e.Values != nil || e.VaultName != "" {
|
|
t.Fatalf("a refused entry carries values: %+v", e)
|
|
}
|
|
if m.calls != 0 {
|
|
t.Fatal("the adapter made a refused consumer")
|
|
}
|
|
if len(out) != 1 || out[0].Status != Refused {
|
|
t.Fatalf("outcomes %+v", out)
|
|
}
|
|
}
|
|
|
|
// A consumer bound under a local name is served when the offer derives nothing from it: its credential is its own
|
|
// pair row under that local name.
|
|
func TestALocalNameIsServedWhenNothingIsDerived(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/redis-cache/ace/letta/queue": "q-secret"}}
|
|
assign(t, bus, "ace", "letta", bind("redis-cache", "queue", "novox", "redis"))
|
|
p := New(Options{Node: "novox", Module: "redis", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
|
|
Offers: map[string]Offer{"redis-cache": {Credential: PerConsumer, Serves: map[string]any{"port": 6379}}}})
|
|
if _, err := p.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var e Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("redis"), "ace.letta.redis-cache.queue"), &e); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if e.Refused != "" || e.VaultName != "pair/novox/redis-cache/ace/letta/queue" || e.Local != "queue" {
|
|
t.Fatalf("entry %+v", e)
|
|
}
|
|
if m.made["ace.letta.redis-cache.queue"] != "q-secret" {
|
|
t.Fatalf("made %v", m.made)
|
|
}
|
|
}
|
|
|
|
// A consumer that unbinds loses its entry and the adapter hears it is gone; an entry another machine's instance of
|
|
// the same provider wrote is not this one's to remove.
|
|
func TestAnUnbindRemovesTheEntryAndLeavesAnotherProvidersAlone(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s1"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
other, _ := json.Marshal(Entry{Provider: "shanks", ProviderModule: "postgres", Provision: "postgres-database",
|
|
Consumer: "g14", Module: "gitea"})
|
|
_ = bus.Put(context.Background(), BucketOf("postgres"), "g14.gitea.postgres-database", other)
|
|
p := postgres(bus, v, m)
|
|
if _, err := p.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"g14.gitea.postgres-database"}) {
|
|
t.Fatalf("entries left %v", got)
|
|
}
|
|
if !slices.Equal(m.gone, []string{"ace.letta.postgres-database"}) {
|
|
t.Fatalf("gone %v", m.gone)
|
|
}
|
|
if len(out) != 1 || out[0].Status != Removed {
|
|
t.Fatalf("outcomes %+v", out)
|
|
}
|
|
}
|
|
|
|
// A bind whose credential the vault does not hold yet is waited for: nothing is made and no entry is written, so a
|
|
// consumer is never told of a login its provider has not made; once the vault holds it, it is answered.
|
|
func TestACredentialTheVaultDoesNotHoldYetIsWaitedFor(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
p := postgres(bus, v, m)
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(out) != 1 || out[0].Status != Waiting || !strings.Contains(out[0].Why, "pair/novox/postgres-database/ace/letta") {
|
|
t.Fatalf("outcomes %+v", out)
|
|
}
|
|
if bus.keys(BucketOf("postgres")) != nil || m.calls != 0 {
|
|
t.Fatal("a consumer was answered before its credential was there")
|
|
}
|
|
v.values["pair/novox/postgres-database/ace/letta"] = "late"
|
|
if _, err := p.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if entryOf(t, bus, "ace.letta.postgres-database").Refused != "" || m.made["ace.letta.postgres-database"] != "late" {
|
|
t.Fatal("not answered once the vault held the value")
|
|
}
|
|
}
|
|
|
|
// Only the binds naming this provider, on this machine, are its own; a bind naming it for a provision it does not
|
|
// offer is refused, naming both ends.
|
|
func TestOnlyBindsNamingThisProviderAreAnswered(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "shanks", "postgres"),
|
|
bind("mongodb-database", "", "novox", "postgres"))
|
|
assign(t, bus, "ace", "other", bind("postgres-database", "", "novox", "mssql"))
|
|
out, err := postgres(bus, v, m).Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := bus.keys(BucketOf("postgres")); !slices.Equal(got, []string{"ace.letta.mongodb-database"}) {
|
|
t.Fatalf("entries %v", got)
|
|
}
|
|
e := entryOf(t, bus, "ace.letta.mongodb-database")
|
|
if !strings.Contains(e.Refused, "postgres on novox") || !strings.Contains(e.Refused, "letta on ace") {
|
|
t.Fatalf("refusal %q", e.Refused)
|
|
}
|
|
if len(out) != 1 || m.calls != 0 {
|
|
t.Fatalf("outcomes %+v, %d made", out, m.calls)
|
|
}
|
|
}
|
|
|
|
// An assignment that cannot be read is skipped and said, by its key and why; the rest of the pass goes on (issue 825,
|
|
// finding 3). Its consumer's own entry is left alone, since a bind missed would read as an unbind.
|
|
func TestAnUnreadableAssignmentIsSkippedAndSaidAndTheRestGoesOn(t *testing.T) {
|
|
for name, bad := range map[string]string{"not json": "{not json",
|
|
"another key": `{"node":"g14","module":"umami"}`} {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s",
|
|
"pair/novox/postgres-database/g14/umami": "u"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
p := postgres(bus, v, m)
|
|
if _, err := p.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
_ = bus.Put(context.Background(), AssignmentsBucket, "ace.letta", []byte(bad))
|
|
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
|
var logged []string
|
|
p.opts.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatalf("%s: %v", name, err)
|
|
}
|
|
if bus.get(BucketOf("postgres"), "g14.umami.postgres-database") == nil {
|
|
t.Fatalf("%s: the good consumer was not answered", name)
|
|
}
|
|
if bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil || len(m.gone) != 0 {
|
|
t.Fatalf("%s: the entry of the unreadable assignment's consumer was removed", name)
|
|
}
|
|
said := false
|
|
for _, o := range out {
|
|
if o.Key == "ace.letta" && o.Status == Failed && o.Why != "" {
|
|
said = true
|
|
}
|
|
}
|
|
if !said || !strings.Contains(strings.Join(logged, "\n"), "ace.letta") {
|
|
t.Fatalf("%s: the skipped assignment is not said: %+v, %q", name, out, logged)
|
|
}
|
|
}
|
|
}
|
|
|
|
// An adapter that fails is said with the credential taken out of its words, and the entry is not written.
|
|
func TestAFailedMakeIsSaidWithoutTheCredential(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{fail: errors.New(`password "hunter2-secret" rejected`)}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "hunter2-secret"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
var logged []string
|
|
p := postgres(bus, v, m)
|
|
p.opts.Logf = func(f string, a ...any) { logged = append(logged, fmt.Sprintf(f, a...)) }
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if len(out) != 1 || out[0].Status != Failed || strings.Contains(out[0].Why, "hunter2") {
|
|
t.Fatalf("outcomes %+v", out)
|
|
}
|
|
if strings.Contains(strings.Join(logged, "\n"), "hunter2") || len(logged) == 0 {
|
|
t.Fatalf("logged %q", logged)
|
|
}
|
|
if bus.keys(BucketOf("postgres")) != nil {
|
|
t.Fatal("an entry was written for a consumer the adapter could not make")
|
|
}
|
|
}
|
|
|
|
// Run watches the binds on the stand-in bus: a consumer that binds is answered, and one that unbinds is removed,
|
|
// without a pass asked by hand.
|
|
func TestRunAnswersBindsAndUnbindsAsTheyAreWatched(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}}
|
|
p := postgres(bus, v, m)
|
|
p.opts.Every = time.Hour
|
|
ctx, cancel := context.WithCancel(context.Background())
|
|
done := make(chan error, 1)
|
|
go func() { done <- p.Run(ctx) }()
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
waitFor(t, func() bool { return bus.get(BucketOf("postgres"), "ace.letta.postgres-database") != nil })
|
|
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
|
|
waitFor(t, func() bool { return bus.get(BucketOf("postgres"), "ace.letta.postgres-database") == nil })
|
|
cancel()
|
|
if err := <-done; err != nil && !errors.Is(err, context.Canceled) {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
func waitFor(t *testing.T, ok func() bool) {
|
|
t.Helper()
|
|
deadline := time.Now().Add(3 * time.Second)
|
|
for !ok() {
|
|
if time.Now().After(deadline) {
|
|
t.Fatal("not within 3s")
|
|
}
|
|
time.Sleep(5 * time.Millisecond)
|
|
}
|
|
}
|
|
|
|
// The consumer's identity is the mesh's (ADR 0202 §1): mesh_<node>_<module>, lower case, `_` for anything else,
|
|
// and its DNS form writes `-` for `_`.
|
|
func TestTheIdentityIsTheMeshsAndItsDNSForm(t *testing.T) {
|
|
if got := ConsumerIdentity("Home.Server", "my-app"); got != "mesh_home_server_my_app" {
|
|
t.Fatal(got)
|
|
}
|
|
got, err := ServedTo(map[string]any{"bucket": "b-${consumer:as:dns}", "n": 3}, "mesh_ace_letta")
|
|
if err != nil || got["bucket"] != "b-mesh-ace-letta" || got["n"] != 3 {
|
|
t.Fatalf("%v %v", got, err)
|
|
}
|
|
if _, err := ServedTo(map[string]any{"x": "${consumer:node}"}, "mesh_ace_letta"); err == nil {
|
|
t.Fatal("a fact the mesh does not state was filled")
|
|
}
|
|
}
|
|
|
|
// A consumer's plan entry may carry what its module contributes to the provider (issue 776): the helper hands it to
|
|
// the provider's adapter with the consumer, and does not write it to the bucket.
|
|
func TestAConsumersContributionIsHandedToTheProviderPerConsumer(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{
|
|
"pair/novox/postgres-database/ace/nextcloud": "a", "pair/novox/postgres-database/g14/umami": "b"}}
|
|
raw := `{"node":"ace","module":"nextcloud","provisioning":{"templates":{"source":"x","digest":"y"},` +
|
|
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
|
|
`"contributes":{"name":"nextcloud","extensions":["pg_trgm"]}}]}}`
|
|
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.nextcloud", []byte(raw)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
|
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
want := map[string]any{"name": "nextcloud", "extensions": []any{"pg_trgm"}}
|
|
if got := m.contributed["ace.nextcloud.postgres-database"]; !reflect.DeepEqual(got, want) {
|
|
t.Fatalf("the adapter was handed the contribution %v, want %v", got, want)
|
|
}
|
|
if got := m.contributed["g14.umami.postgres-database"]; got != nil {
|
|
t.Fatalf("a consumer contributing nothing was handed %v", got)
|
|
}
|
|
if bytes.Contains(bus.get(BucketOf("postgres"), "ace.nextcloud.postgres-database"), []byte("pg_trgm")) {
|
|
t.Fatal("a contribution is in the provider's bucket")
|
|
}
|
|
}
|
|
|
|
// A consumer's plan entry carries where its provider is, which machine, and who the consumer is (issue 788): the
|
|
// entry's values carry `at` and `from` as the binding file gives them, and its `as` is the plan's, so a module with a
|
|
// slug keeps the login it has today. An entry of a plan without them is as it was.
|
|
func TestAConsumersEntryCarriesAtFromAndTheAsOfItsPlan(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{
|
|
"pair/novox/postgres-database/ace/claude-code": "a", "pair/novox/postgres-database/g14/umami": "b"}}
|
|
raw := `{"node":"ace","module":"claude-code","provisioning":{"templates":{"source":"x","digest":"y"},` +
|
|
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
|
|
`"at":"10.7.0.1","from":"novox","as":"mesh_ace_agent"}]}}`
|
|
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.claude-code", []byte(raw)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
|
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var e Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "ace.claude-code.postgres-database"), &e); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if e.Values["at"] != "10.7.0.1" || e.Values["from"] != "novox" || e.Values["port"] == nil {
|
|
t.Fatalf("the entry's values are %v", e.Values)
|
|
}
|
|
if e.As != "mesh_ace_agent" || e.User != "mesh_ace_agent" || m.as["ace.claude-code.postgres-database"] != "mesh_ace_agent" {
|
|
t.Fatalf("the entry is %+v, made as %q", e, m.as["ace.claude-code.postgres-database"])
|
|
}
|
|
var plain Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "g14.umami.postgres-database"), &plain); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, has := plain.Values["at"]; has || plain.As != "mesh_g14_umami" {
|
|
t.Fatalf("an entry of a plan without them changed: %+v", plain)
|
|
}
|
|
}
|
|
|
|
// A consumer's plan entry may carry the names the controller composed for it (issue 792): each is copied into the
|
|
// entry's values under its key, over anything the provider serves under it. An entry without them is as it was.
|
|
func TestAConsumersEntryCarriesTheNamesOfItsPlan(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{
|
|
"pair/novox/postgres-database/ace/gitea": "a", "pair/novox/postgres-database/g14/umami": "b"}}
|
|
raw := `{"node":"ace","module":"gitea","provisioning":{"templates":{"source":"x","digest":"y"},` +
|
|
`"provisions":[{"provision":"postgres-database","provider":"novox","provider-module":"postgres",` +
|
|
`"names":{"name":"git.example.org","internal-name":"git.mesh","name-web":"web.example.org"}}]}}`
|
|
if err := bus.Put(context.Background(), AssignmentsBucket, "ace.gitea", []byte(raw)); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
assign(t, bus, "g14", "umami", bind("postgres-database", "", "novox", "postgres"))
|
|
if _, err := postgres(bus, v, m).Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var e Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "ace.gitea.postgres-database"), &e); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if e.Values["name"] != "git.example.org" || e.Values["internal-name"] != "git.mesh" ||
|
|
e.Values["name-web"] != "web.example.org" || e.Values["port"] == nil {
|
|
t.Fatalf("the entry's values are %v", e.Values)
|
|
}
|
|
var plain Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "g14.umami.postgres-database"), &plain); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if _, has := plain.Values["name"]; has {
|
|
t.Fatalf("an entry of a plan without names changed: %+v", plain)
|
|
}
|
|
}
|
|
|
|
// refusedByTheStateGuard mirrors the key rule of the runtime's state guard (mesh-tools node-tools internal/bus
|
|
// state.go): a JSON key whose lowercased name, with `-`, `_` and spaces taken out, ends in one of these is refused.
|
|
func refusedByTheStateGuard(key string) bool {
|
|
k := strings.NewReplacer("-", "", "_", "", " ", "").Replace(strings.ToLower(key))
|
|
for _, end := range []string{"password", "passwd", "secret", "token", "credential", "credentials",
|
|
"authorization", "apikey", "privatekey", "accesskey", "cookie"} {
|
|
if strings.HasSuffix(k, end) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func keysOf(v any, into *[]string) {
|
|
switch x := v.(type) {
|
|
case map[string]any:
|
|
for k, c := range x {
|
|
*into = append(*into, k)
|
|
keysOf(c, into)
|
|
}
|
|
case []any:
|
|
for _, c := range x {
|
|
keysOf(c, into)
|
|
}
|
|
}
|
|
}
|
|
|
|
// Every entry the helper writes passes the state guard's key rule (issue 945), and names where the vault keeps the
|
|
// credential under `vault-name`, for a login per consumer and for a shared one.
|
|
func TestAnEntryHasNoKeyTheStateGuardRefuses(t *testing.T) {
|
|
for name, form := range map[string]CredentialForm{"pair": PerConsumer, "shared": Shared} {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s1",
|
|
"provision/novox/postgres-database": "s2"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
|
|
Offers: map[string]Offer{"postgres-database": {Credential: form, User: "shared-user"}}})
|
|
if _, err := p.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var doc map[string]any
|
|
if err := json.Unmarshal(bus.get(BucketOf("postgres"), "ace.letta.postgres-database"), &doc); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
var keys []string
|
|
keysOf(doc, &keys)
|
|
for _, k := range keys {
|
|
if refusedByTheStateGuard(k) {
|
|
t.Fatalf("%s: the entry carries the key %q, which the state guard refuses", name, k)
|
|
}
|
|
}
|
|
want := map[string]string{"pair": "pair/novox/postgres-database/ace/letta",
|
|
"shared": "provision/novox/postgres-database"}[name]
|
|
if doc["vault-name"] != want {
|
|
t.Fatalf("%s: vault-name is %v, want %s", name, doc["vault-name"], want)
|
|
}
|
|
}
|
|
}
|
|
|
|
// The plan's credential form and holder name win over the offer's; the offer's form is the default (issue 825,
|
|
// finding 4).
|
|
func TestThePlansCredentialFormWinsOverTheOffers(t *testing.T) {
|
|
run := func(offer CredentialForm, plan map[string]string) Entry {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "p",
|
|
"provision/novox/postgres-database": "s", "provision/novox/postgres-database/ace/letta": "h"}}
|
|
b := bind("postgres-database", "", "novox", "postgres")
|
|
for k, val := range plan {
|
|
b[k] = val
|
|
}
|
|
assign(t, bus, "ace", "letta", b)
|
|
p := New(Options{Node: "novox", Module: "postgres", Bus: bus, Vault: v, Make: m.Make, Logf: func(string, ...any) {},
|
|
Offers: map[string]Offer{"postgres-database": {Credential: offer, User: "shared-user"}}})
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil || len(out) != 1 || out[0].Status != Answered {
|
|
t.Fatalf("%v %+v", err, out)
|
|
}
|
|
return entryOf(t, bus, "ace.letta.postgres-database")
|
|
}
|
|
if e := run(PerConsumer, map[string]string{"credential-form": "shared"}); e.VaultName !=
|
|
"provision/novox/postgres-database" || e.User != "shared-user" {
|
|
t.Fatalf("shared over pair: %+v", e)
|
|
}
|
|
if e := run(Shared, map[string]string{"credential-form": "pair"}); e.VaultName !=
|
|
"pair/novox/postgres-database/ace/letta" || e.User != "mesh_ace_letta" {
|
|
t.Fatalf("pair over shared: %+v", e)
|
|
}
|
|
for _, form := range []string{"issued", "given"} {
|
|
if e := run(PerConsumer, map[string]string{"credential-form": form, "name": "main"}); e.VaultName !=
|
|
"provision/novox/postgres-database/ace/letta" || e.User != "mesh_ace_letta" {
|
|
t.Fatalf("%s holder: %+v", form, e)
|
|
}
|
|
}
|
|
if e := run(Shared, nil); e.VaultName != "provision/novox/postgres-database" {
|
|
t.Fatalf("the offer's form is the default: %+v", e)
|
|
}
|
|
}
|
|
|
|
// An instance deletes only an entry it wrote, and only while the entry is as it read it: one that another instance of
|
|
// the provider wrote after the consumer moved stands (issue 825, finding 5).
|
|
func TestAnEntryAnotherInstanceWroteAfterTheMoveStands(t *testing.T) {
|
|
bus, m := newStandIn(), &maker{}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": "s"}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
a := postgres(bus, v, m)
|
|
if _, err := a.Pass(context.Background()); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
// The consumer moves to the instance on shanks: A's pass reads its own entry and the unbind, and while it lets
|
|
// the consumer go, B writes the consumer's entry.
|
|
_ = bus.Delete(context.Background(), AssignmentsBucket, "ace.letta")
|
|
theirs, _ := json.Marshal(Entry{Provider: "shanks", ProviderModule: "postgres", Provision: "postgres-database",
|
|
Consumer: "ace", Module: "letta", VaultName: "pair/shanks/postgres-database/ace/letta"})
|
|
a.opts.Gone = func(ctx context.Context, e Entry) error {
|
|
return bus.Put(ctx, BucketOf("postgres"), e.Key(), theirs)
|
|
}
|
|
out, err := a.Pass(context.Background())
|
|
if err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
if got := bus.get(BucketOf("postgres"), "ace.letta.postgres-database"); !bytes.Equal(got, theirs) {
|
|
t.Fatalf("the other instance's entry was removed: %s", got)
|
|
}
|
|
for _, o := range out {
|
|
if o.Status == Removed {
|
|
t.Fatalf("outcomes %+v", out)
|
|
}
|
|
}
|
|
}
|
|
|
|
// A failed Make is said with the credential taken out in the forms it can come back in: as it is, in base64 (standard
|
|
// and URL, with and without padding) and as a JSON string says it (issue 825, finding 6).
|
|
func TestAFailedMakeIsSaidWithoutTheCredentialInAnyOfItsForms(t *testing.T) {
|
|
cred := "pa/ss+w\"rd?>>x"
|
|
b := []byte(cred)
|
|
jsonQuoted, _ := json.Marshal(cred)
|
|
forms := map[string]string{
|
|
"raw": cred, "std": base64.StdEncoding.EncodeToString(b), "std unpadded": base64.RawStdEncoding.EncodeToString(b),
|
|
"url": base64.URLEncoding.EncodeToString(b), "url unpadded": base64.RawURLEncoding.EncodeToString(b),
|
|
"json": string(jsonQuoted[1 : len(jsonQuoted)-1]),
|
|
}
|
|
if forms["std"] == forms["url"] || forms["json"] == forms["raw"] {
|
|
t.Fatal("the test credential does not tell the forms apart")
|
|
}
|
|
for name, form := range forms {
|
|
bus, m := newStandIn(), &maker{fail: errors.New("login failed for " + form + " at the server")}
|
|
v := &vault{values: map[string]string{"pair/novox/postgres-database/ace/letta": cred}}
|
|
assign(t, bus, "ace", "letta", bind("postgres-database", "", "novox", "postgres"))
|
|
out, err := postgres(bus, v, m).Pass(context.Background())
|
|
if err != nil || len(out) != 1 {
|
|
t.Fatalf("%v %+v", err, out)
|
|
}
|
|
if strings.Contains(out[0].Why, form) || !strings.Contains(out[0].Why, "<credential>") {
|
|
t.Fatalf("%s form: %q", name, out[0].Why)
|
|
}
|
|
}
|
|
}
|
|
|
|
// An offer that declares no credential is answered with none whatever form the plan says (the controller writes
|
|
// "pair" for such an offer): no vault name, and the Vault is never asked, so a provider with no Vault works (issue 825,
|
|
// finding 4; issue 923).
|
|
func TestAnOfferWithNoCredentialIgnoresThePlansForm(t *testing.T) {
|
|
bus := newStandIn()
|
|
b := bind("route", "", "novox", "route-proxy")
|
|
b["credential-form"] = "pair"
|
|
assign(t, bus, "ace", "letta", b)
|
|
p := New(Options{Node: "novox", Module: "route-proxy", Bus: bus, Logf: func(string, ...any) {},
|
|
Offers: map[string]Offer{"route": {Credential: NoCredential}}})
|
|
out, err := p.Pass(context.Background())
|
|
if err != nil || len(out) != 1 || out[0].Status != Answered {
|
|
t.Fatalf("%v %+v", err, out)
|
|
}
|
|
var e Entry
|
|
if err := json.Unmarshal(bus.get(BucketOf("route-proxy"), "ace.letta.route"), &e); err != nil || e.VaultName != "" ||
|
|
e.User != "" {
|
|
t.Fatalf("%v %+v", err, e)
|
|
}
|
|
}
|