vaultik/internal/storage/conformance_test.go

package storage_test

import (
	"bytes"
	"context"
	"errors"
	"io"
	"reflect"
	"sort"
	"testing"
	"time"

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

// runStorerConformance is the shared Storer contract. Every backend that
// can run in-process is expected to pass it: TestFileStorer runs it against
// file://, TestS3Storer against s3://, TestRcloneStorer against rclone's
// local backend. A new backend inherits this coverage by passing its own
// constructor, so the contract is defined once.
//
// It exercises the public Storer interface: round-trip, stat, list with
// prefix filtering, the modification time ListStream reports, overwrite,
// delete, delete-of-missing, and not-found on Get and Stat. Each section
// takes its own fresh backend instance, so the order of sections never
// matters and no section sees another's objects.
func runStorerConformance(t *testing.T, newStorer func(*testing.T) storage.Storer) {
	t.Helper()

	conformanceRoundTrip(t, newStorer(t))
	conformanceOverwrite(t, newStorer(t))
	conformanceList(t, newStorer(t))
	conformanceListStreamModTime(t, newStorer(t))
	conformanceDelete(t, newStorer(t))
	conformanceNotFound(t, newStorer(t))
}

// conformanceRoundTrip stores a nested key, then reads it back and stats it.
func conformanceRoundTrip(t *testing.T, s storage.Storer) {
	t.Helper()

	ctx := context.Background()
	key := "blobs/aa/bb/object.bin"
	want := []byte("round-trip payload")

	err := s.Put(ctx, key, bytes.NewReader(want))
	if err != nil {
		t.Fatalf("Put: %v", err)
	}

	got := getBytes(t, s, key)
	if !bytes.Equal(got, want) {
		t.Errorf("Get returned %q, want %q", got, want)
	}

	info, err := s.Stat(ctx, key)
	if err != nil {
		t.Fatalf("Stat: %v", err)
	}

	if info.Key != key {
		t.Errorf("Stat key = %q, want %q", info.Key, key)
	}

	if info.Size != int64(len(want)) {
		t.Errorf("Stat size = %d, want %d", info.Size, len(want))
	}
}

// conformanceOverwrite checks that a second Put replaces the first.
func conformanceOverwrite(t *testing.T, s storage.Storer) {
	t.Helper()

	ctx := context.Background()
	key := "meta/snapshot.json"

	err := s.Put(ctx, key, bytes.NewReader([]byte("first")))
	if err != nil {
		t.Fatalf("first Put: %v", err)
	}

	want := []byte("second and longer payload")

	err = s.Put(ctx, key, bytes.NewReader(want))
	if err != nil {
		t.Fatalf("second Put: %v", err)
	}

	got := getBytes(t, s, key)
	if !bytes.Equal(got, want) {
		t.Errorf("after overwrite Get returned %q, want %q", got, want)
	}
}

// conformanceList checks prefix filtering and the empty result for a
// prefix that matches nothing.
func conformanceList(t *testing.T, s storage.Storer) {
	t.Helper()

	ctx := context.Background()
	keys := []string{"blobs/aa/one", "blobs/bb/two", "meta/three"}

	for _, k := range keys {
		err := s.Put(ctx, k, bytes.NewReader([]byte("data")))
		if err != nil {
			t.Fatalf("Put %q: %v", k, err)
		}
	}

	if got := listSorted(t, s, ""); !reflect.DeepEqual(got, keys) {
		t.Errorf("List(\"\") = %v, want %v", got, keys)
	}

	wantBlobs := []string{"blobs/aa/one", "blobs/bb/two"}
	if got := listSorted(t, s, "blobs/"); !reflect.DeepEqual(got, wantBlobs) {
		t.Errorf("List(\"blobs/\") = %v, want %v", got, wantBlobs)
	}

	if got := listSorted(t, s, "absent/"); len(got) != 0 {
		t.Errorf("List(\"absent/\") = %v, want empty", got)
	}
}

// conformanceListStreamModTime checks that ListStream gives an object
// just written a modification time of about now, which prune compares
// with its grace period.
func conformanceListStreamModTime(t *testing.T, s storage.Storer) {
	t.Helper()

	ctx := context.Background()

	err := s.Put(ctx, "blobs/dd/written-now", bytes.NewReader([]byte("data")))
	if err != nil {
		t.Fatalf("Put: %v", err)
	}

	var listed []storage.ObjectInfo

	for info := range s.ListStream(ctx, "blobs/") {
		listed = append(listed, info)
	}

	if len(listed) != 1 || listed[0].Err != nil {
		t.Fatalf("ListStream(\"blobs/\") = %v, want the one object", listed)
	}

	if time.Since(listed[0].ModTime).Abs() > time.Minute {
		t.Errorf("ListStream ModTime = %v, want within a minute of now",
			listed[0].ModTime)
	}
}

// conformanceDelete checks that Delete removes an object and that deleting
// a missing key is not an error.
func conformanceDelete(t *testing.T, s storage.Storer) {
	t.Helper()

	ctx := context.Background()
	key := "blobs/cc/gone.bin"

	err := s.Put(ctx, key, bytes.NewReader([]byte("temporary")))
	if err != nil {
		t.Fatalf("Put: %v", err)
	}

	err = s.Delete(ctx, key)
	if err != nil {
		t.Fatalf("Delete: %v", err)
	}

	_, err = s.Get(ctx, key)
	if !errors.Is(err, storage.ErrNotFound) {
		t.Errorf("Get after Delete error = %v, want ErrNotFound", err)
	}

	err = s.Delete(ctx, key)
	if err != nil {
		t.Errorf("Delete of missing key = %v, want nil", err)
	}
}

// conformanceNotFound checks Get and Stat on an absent key.
func conformanceNotFound(t *testing.T, s storage.Storer) {
	t.Helper()

	ctx := context.Background()
	key := "never/written"

	_, err := s.Get(ctx, key)
	if !errors.Is(err, storage.ErrNotFound) {
		t.Errorf("Get error = %v, want ErrNotFound", err)
	}

	_, err = s.Stat(ctx, key)
	if !errors.Is(err, storage.ErrNotFound) {
		t.Errorf("Stat error = %v, want ErrNotFound", err)
	}
}

// getBytes reads a key fully and closes the reader.
func getBytes(t *testing.T, s storage.Storer, key string) []byte {
	t.Helper()

	rc, err := s.Get(context.Background(), key)
	if err != nil {
		t.Fatalf("Get %q: %v", key, err)
	}

	defer func() { _ = rc.Close() }()

	data, err := io.ReadAll(rc)
	if err != nil {
		t.Fatalf("read %q: %v", key, err)
	}

	return data
}

// listSorted returns the keys under a prefix in a stable order.
func listSorted(t *testing.T, s storage.Storer, prefix string) []string {
	t.Helper()

	keys, err := s.List(context.Background(), prefix)
	if err != nil {
		t.Fatalf("List %q: %v", prefix, err)
	}

	sort.Strings(keys)

	return keys
}