mesh-catalog/modules/mesh-build-node/cmd/mesh-build-node/push_test.go

package main

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"

	registry "git.novox.be/novox/mesh-delivery/contracts/go/seats/containerimageregistry"

	"git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
	builds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds"
)

// pushed are the blobs the fake holder took, by digest, for the tests that read an archive back.
var pushed sync.Map

// fakeStore is the registry seats' holder as m's publish meets it (publish.go; CAT-BN3): publish opens a transfer and
// answers the part it lacks, publish-part takes the chunks in sequence, and the last one keeps the whole part when it
// hashes to its digest. Each file is kept by name and digest; fail, when set, is the holder's refusal at the last chunk.
type fakeStore struct {
	mu        sync.Mutex
	blobs     map[string][]byte // "<name>@<digest>"
	fail      error
	open      map[string]registry.PublishInput
	chunks    map[string][]byte
	published []registry.PublishInput
	seats     []string
}

func withStore(t *testing.T, m *Module) *fakeStore {
	t.Helper()
	s := &fakeStore{blobs: map[string][]byte{}, open: map[string]registry.PublishInput{}, chunks: map[string][]byte{}}
	m.Call = func(key string, body any) (json.RawMessage, error) {
		s.mu.Lock()
		defer s.mu.Unlock()
		raw, _ := json.Marshal(body)
		seat, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".")
		if seat != buildstep.ArtifactSeat && seat != buildstep.NpmSeat {
			return nil, errors.New("nats: no responders available for request")
		}
		switch verb {
		case "publish":
			var in registry.PublishInput
			_ = json.Unmarshal(raw, &in)
			if (seat == buildstep.ArtifactSeat) != (in.Kind == "bundle") || in.Transfer == "" || strings.ContainsAny(in.Transfer, "/. *>") {
				t.Errorf("publish on %s of kind %q, transfer %q", seat, in.Kind, in.Transfer)
			}
			s.open[in.Transfer] = in
			s.seats = append(s.seats, seat)
			return json.Marshal(registry.PublishAnswer{Transfer: in.Transfer, Lacks: []string{in.Digest}})
		case "publish-part":
			var c registry.PublishPartInput
			_ = json.Unmarshal(raw, &c)
			data, _ := base64.StdEncoding.DecodeString(c.Data)
			s.chunks[c.Transfer] = append(s.chunks[c.Transfer], data...)
			if !c.Last {
				return json.Marshal(registry.PublishPartAnswer{Transfer: c.Transfer, Part: c.Part})
			}
			in, whole := s.open[c.Transfer], s.chunks[c.Transfer]
			delete(s.chunks, c.Transfer)
			if s.fail != nil {
				return nil, s.fail
			}
			if sum := sha256.Sum256(whole); "sha256:"+hex.EncodeToString(sum[:]) != in.Digest {
				t.Errorf("published %s with bytes of another digest", in.Digest)
			}
			s.blobs[in.Name+"@"+in.Digest] = whole
			s.published = append(s.published, in)
			pushed.Store(in.Digest, whole)
			return json.Marshal(registry.PublishPartAnswer{Transfer: c.Transfer, Part: c.Part, Complete: true})
		}
		return nil, errors.New("nats: no responders available for request")
	}
	return s
}

// archiveOf is the archive a build published, by its digest.
func archiveOf(t *testing.T, r BuildResult) []byte {
	t.Helper()
	raw, ok := pushed.Load(r.Digest)
	if !ok {
		t.Fatalf("nothing was published under %s", r.Digest)
	}
	return raw.([]byte)
}

// 1587-S6 (ADR 0498 §1; CAT-BN3): a done bundle is published through oci-artifact-registry and answers the seat's
// reference of its archive, under the manifest's module.
func TestADoneBundleAnswersTheReferenceOfTheArchiveItPushed(t *testing.T) {
	m, _, _ := builtMachine(t, ``)
	store := withStore(t, m)
	out, err := m.build(context.Background(), buildArgs())
	if err != nil {
		t.Fatal(err)
	}
	r := out.(BuildResult)
	if want := "oci-artifact-registry/m/tools@" + r.Digest; r.Reference != want {
		t.Errorf("reference %q, want %q", r.Reference, want)
	}
	if store.blobs["m/tools@"+r.Digest] == nil || len(store.seats) != 1 || store.seats[0] != buildstep.ArtifactSeat {
		t.Errorf("the holder holds %d files through %v, not m/tools@%s through %s", len(store.blobs), store.seats, r.Digest, buildstep.ArtifactSeat)
	}
}

// 1587-S7 and S8 (ADR 0498 §2, the lead's decision L9): a publish the holder refuses fails the step `not stored:
// <artifact>: <the holder's words>`, a seat nothing holds too, and whatever it answered, no packed file is left.
func TestARefusedPushFailsNotStoredAndNoPackedFileStays(t *testing.T) {
	for _, c := range []struct {
		fail   error
		noSeat bool
		words  string
	}{{nil, false, ""}, {errors.New("refused: the store did not take it: 507 Insufficient Storage"), false, "507"},
		{nil, true, "no responders"}} {
		m, _, work := builtMachine(t, ``)
		withStore(t, m).fail = c.fail
		if c.noSeat {
			m.Call = func(string, any) (json.RawMessage, error) {
				return nil, errors.New("nats: no responders available for request")
			}
		}
		_, err := m.build(context.Background(), buildArgs())
		if c.words != "" && (err == nil || !strings.HasPrefix(err.Error(), builds.NotStored+"tools: ") || !strings.Contains(err.Error(), c.words)) {
			t.Errorf("a refused publish answered %v", err)
		}
		var archives []string
		filepath.WalkDir(work, func(p string, d os.DirEntry, err error) error {
			if err == nil && strings.HasSuffix(p, ".tar.gz") {
				archives = append(archives, p)
			}
			return nil
		})
		if len(archives) != 0 {
			t.Errorf("publish answered %v: packed files left on the node: %v", c.fail, archives)
		}
	}
}