vaultik/internal/storage/s3_test.go

package storage_test

import (
	"bytes"
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"slices"
	"strings"
	"sync/atomic"
	"testing"

	"github.com/johannesboyne/gofakes3"
	"github.com/johannesboyne/gofakes3/backend/s3mem"

	"sneak.berlin/go/vaultik/internal/config"
	"sneak.berlin/go/vaultik/internal/s3"
	"sneak.berlin/go/vaultik/internal/storage"
)

// s3TestBucket is the bucket created for each in-process S3 server.
const s3TestBucket = "test-bucket"

// Credentials for the tests' S3 clients. The test servers accept any.
const (
	s3TestAccessKeyID     = "key"
	s3TestSecretAccessKey = "secret"
)

// s3TestRegion is the region of the S3 clients built here with
// s3.NewClient. The test servers ignore it.
const s3TestRegion = "us-east-1"

// newS3Storer builds an s3:// backend backed by a fresh in-process
// S3 server. It reuses the same in-memory S3 harness (gofakes3 + s3mem
// over httptest) that internal/s3 and the not-found regression test use,
// so no new mock or dependency is introduced. Each call gets its own
// server, bucket, and client, so the conformance suite's per-section
// instances stay isolated.
//
//nolint:ireturn // conformance runs against the Storer interface by design
func newS3Storer(t *testing.T) storage.Storer {
	t.Helper()

	backend := s3mem.New()

	err := backend.CreateBucket(s3TestBucket)
	if err != nil {
		t.Fatalf("create bucket: %v", err)
	}

	srv := httptest.NewServer(gofakes3.New(backend).Server())
	t.Cleanup(srv.Close)

	client, err := s3.NewClient(context.Background(), s3.Config{
		Endpoint:        srv.URL,
		Bucket:          s3TestBucket,
		AccessKeyID:     "test",
		SecretAccessKey: "test",
		Region:          s3TestRegion,
	})
	if err != nil {
		t.Fatalf("new client: %v", err)
	}

	return storage.NewS3Storer(client)
}

// TestS3Storer runs the shared Storer contract against the s3:// backend,
// so it is held to the same round-trip, list, delete, and not-found
// behaviour as the file:// backend.
func TestS3Storer(t *testing.T) {
	t.Parallel()
	runStorerConformance(t, newS3Storer)
}

// TestS3StorerMissingKeyMapsToErrNotFound pins the specific contract that a
// missing object surfaces as storage.ErrNotFound rather than the raw AWS SDK
// error. Without the mapping, errors.Is(err, storage.ErrNotFound) is false on
// s3 and callers would branch differently per backend.
func TestS3StorerMissingKeyMapsToErrNotFound(t *testing.T) {
	t.Parallel()

	storer := newS3Storer(t)
	ctx := context.Background()

	_, err := storer.Get(ctx, "does-not-exist")
	if !errors.Is(err, storage.ErrNotFound) {
		t.Errorf("Get on missing key: got %v, want ErrNotFound", err)
	}

	_, err = storer.Stat(ctx, "does-not-exist")
	if !errors.Is(err, storage.ErrNotFound) {
		t.Errorf("Stat on missing key: got %v, want ErrNotFound", err)
	}
}

// TestS3ListingCutShortIsAnError checks that List and ListStream fail on a
// listing page marked truncated that carries no continuation token, as a
// server answering V2 listings V1-style sends it. Taken as the last page,
// it would hide every later key, and prune would treat the blobs of the
// snapshots listed there as unreferenced.
func TestS3ListingCutShortIsAnError(t *testing.T) {
	t.Parallel()

	const (
		prefix        = "metadata/"
		truncatedPage = `<?xml version="1.0" encoding="UTF-8"?>
<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
  <Name>test-bucket</Name>
  <Prefix>metadata/</Prefix>
  <KeyCount>1</KeyCount>
  <MaxKeys>1</MaxKeys>
  <IsTruncated>true</IsTruncated>
  <Contents>
    <Key>metadata/snap/manifest.json.zst</Key>
    <Size>8</Size>
  </Contents>
</ListBucketResult>`
	)

	srv := httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, _ *http.Request) {
			_, _ = io.WriteString(w, truncatedPage)
		}))
	t.Cleanup(srv.Close)

	client, err := s3.NewClient(context.Background(), s3.Config{
		Endpoint:        srv.URL,
		Bucket:          s3TestBucket,
		AccessKeyID:     s3TestAccessKeyID,
		SecretAccessKey: s3TestSecretAccessKey,
		Region:          s3TestRegion,
	})
	if err != nil {
		t.Fatalf("new client: %v", err)
	}

	storer := storage.NewS3Storer(client)
	ctx := context.Background()

	keys, err := storer.List(ctx, prefix)
	if !errors.Is(err, s3.ErrNoContinuationToken) {
		t.Errorf("List = %q, %v; want ErrNoContinuationToken", keys, err)
	}

	var streamErr error

	for obj := range storer.ListStream(ctx, prefix) {
		if obj.Err != nil {
			streamErr = obj.Err
		}
	}

	if !errors.Is(streamErr, s3.ErrNoContinuationToken) {
		t.Errorf("ListStream error = %v, want ErrNoContinuationToken", streamErr)
	}
}

// TestS3URLPrefixKeyLayout pins the bucket keys an s3:// URL reads and
// writes: the README's remote storage layout, with the prefix joined to
// each key by one "/". s3://b/p and s3://b/p/ must be the same
// destination, or a host that writes the URL the other way finds no
// snapshots. The listed object is put straight into the bucket, as
// another host would have written it. Both List and ListStream are
// checked: ListStream is what every snapshot listing goes through.
func TestS3URLPrefixKeyLayout(t *testing.T) {
	t.Parallel()

	const (
		blobKey      = "blobs/aa/bb/aabbccdd"
		listPrefix   = "metadata/"
		manifestKey  = listPrefix + "snap/manifest.json.zst"
		manifestBody = "manifest"
	)

	cases := []struct {
		urlPath   string // URL path after the bucket name
		keyPrefix string // what every key in the bucket must start with
	}{
		{urlPath: "/p", keyPrefix: "p/"},
		{urlPath: "/p/", keyPrefix: "p/"},
		{urlPath: "", keyPrefix: ""},
	}

	for _, tc := range cases {
		storageURL := "s3://" + s3TestBucket + tc.urlPath

		t.Run(storageURL, func(t *testing.T) {
			t.Parallel()

			backend := s3mem.New()

			err := backend.CreateBucket(s3TestBucket)
			if err != nil {
				t.Fatalf("create bucket: %v", err)
			}

			srv := httptest.NewServer(gofakes3.New(backend).Server())
			t.Cleanup(srv.Close)

			storer, err := storage.NewStorer(&config.Config{
				StorageURL: storageURL + "?endpoint=" + srv.URL,
				S3: config.S3Config{
					AccessKeyID:     s3TestAccessKeyID,
					SecretAccessKey: s3TestSecretAccessKey,
				},
			})
			if err != nil {
				t.Fatalf("NewStorer: %v", err)
			}

			ctx := context.Background()

			err = storer.Put(ctx, blobKey, strings.NewReader("blob"))
			if err != nil {
				t.Fatalf("Put: %v", err)
			}

			_, err = backend.HeadObject(s3TestBucket, tc.keyPrefix+blobKey)
			if err != nil {
				t.Errorf("blob not stored at %q: %v", tc.keyPrefix+blobKey, err)
			}

			_, err = backend.PutObject(s3TestBucket, tc.keyPrefix+manifestKey,
				nil, strings.NewReader(manifestBody), int64(len(manifestBody)))
			if err != nil {
				t.Fatalf("seed manifest: %v", err)
			}

			keys, err := storer.List(ctx, listPrefix)
			if err != nil {
				t.Fatalf("List: %v", err)
			}

			if !slices.Equal(keys, []string{manifestKey}) {
				t.Errorf("List(%q) = %q, want [%q]", listPrefix, keys, manifestKey)
			}

			streamed := listStreamKeys(t, storer, listPrefix)
			if !slices.Equal(streamed, []string{manifestKey}) {
				t.Errorf("ListStream(%q) = %q, want [%q]", listPrefix, streamed, manifestKey)
			}
		})
	}
}

// TestS3UploadUsesConfiguredPartSize checks that s3.part_size reaches the
// multipart uploader, through storage_url and through the s3.* fields. An
// object three parts long must arrive as three parts; at the SDK's default
// of 5 MiB it would arrive as four.
func TestS3UploadUsesConfiguredPartSize(t *testing.T) {
	t.Parallel()

	const (
		partSize  = 6 * 1024 * 1024
		wantParts = 3
	)

	backend := s3mem.New()

	err := backend.CreateBucket(s3TestBucket)
	if err != nil {
		t.Fatalf("create bucket: %v", err)
	}

	// Every part of a multipart upload is one request with a partNumber.
	var parts atomic.Int32

	fake := gofakes3.New(backend).Server()
	srv := httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, r *http.Request) {
			if r.URL.Query().Has("partNumber") {
				parts.Add(1)
			}

			fake.ServeHTTP(w, r)
		}))
	t.Cleanup(srv.Close)

	cases := []struct {
		name string
		cfg  *config.Config
	}{
		{
			name: "storage_url",
			cfg: &config.Config{
				StorageURL: "s3://" + s3TestBucket + "?endpoint=" + srv.URL,
				S3: config.S3Config{
					AccessKeyID:     s3TestAccessKeyID,
					SecretAccessKey: s3TestSecretAccessKey,
					PartSize:        partSize,
				},
			},
		},
		{
			name: "s3.endpoint",
			cfg: &config.Config{
				S3: config.S3Config{
					Endpoint:        srv.URL,
					Bucket:          s3TestBucket,
					AccessKeyID:     s3TestAccessKeyID,
					SecretAccessKey: s3TestSecretAccessKey,
					PartSize:        partSize,
				},
			},
		},
	}

	for _, tc := range cases {
		parts.Store(0)

		storer, err := storage.NewStorer(tc.cfg)
		if err != nil {
			t.Fatalf("%s: NewStorer: %v", tc.name, err)
		}

		data := bytes.NewReader(make([]byte, wantParts*partSize))

		err = storer.PutWithProgress(
			context.Background(), "blob", data, data.Size(), nil)
		if err != nil {
			t.Fatalf("%s: PutWithProgress: %v", tc.name, err)
		}

		if got := parts.Load(); got != wantParts {
			t.Errorf("%s: uploaded in %d parts, want %d", tc.name, got, wantParts)
		}
	}
}

// listStreamKeys returns the keys ListStream yields under a prefix, and
// fails the test on a listing error.
func listStreamKeys(t *testing.T, s storage.Storer, prefix string) []string {
	t.Helper()

	var keys []string

	for obj := range s.ListStream(context.Background(), prefix) {
		if obj.Err != nil {
			t.Fatalf("ListStream %q: %v", prefix, obj.Err)
		}

		keys = append(keys, obj.Key)
	}

	return keys
}