package storage_test import ( "bytes" "context" "errors" "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" // 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: "us-east-1", }) 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) } } // 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: "key", SecretAccessKey: "secret", }, }) 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: "key", SecretAccessKey: "secret", PartSize: partSize, }, }, }, { name: "s3.endpoint", cfg: &config.Config{ S3: config.S3Config{ Endpoint: srv.URL, Bucket: s3TestBucket, AccessKeyID: "key", SecretAccessKey: "secret", 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 }