seaweedfs/weed/s3api/s3api_versioning_list_test.go

package s3api

import (
	"path"
	"testing"

	"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
	"github.com/seaweedfs/seaweedfs/weed/s3api/s3_constants"
	"github.com/stretchr/testify/assert"
)

// TestVersionedObjectListingPathConstruction tests that versioned objects are listed with correct paths.
// This is an integration test that validates the fix for GitHub discussion #7573.
//
// Issue: When using Velero with Kopia and S3 bucket versioning enabled, list operations were
// returning doubled paths like "kopia/logpaste/kopia/logpaste/file" instead of "kopia/logpaste/file".
// This caused Kopia to fail with "unable to load pack indexes despite 10 retries".
//
// Root cause: getLatestVersionEntryForListOperation was setting entry.Name to the full object path
// instead of just the base filename. When eachEntryFn combined dir + entry.Name, paths got doubled.
//
// Fix: Use path.Base(object) to extract only the filename portion.
func TestVersionedObjectListingPathConstruction(t *testing.T) {
	// Test that getLatestVersionEntryForListOperation creates entries with correct Name field
	t.Run("entry name should be base filename for versioned objects", func(t *testing.T) {
		// Simulate the fix: when creating logical entries for versioned object listing,
		// we use path.Base(object) instead of the full path
		testCases := []struct {
			name           string
			objectPath     string
			expectedName   string
			dir            string
			bucketPrefix   string
			expectedKey    string
		}{
			{
				name:         "kopia file in nested directory",
				objectPath:   "kopia/logpaste/kopia.blobcfg",
				expectedName: "kopia.blobcfg",
				dir:          "/buckets/velero/kopia/logpaste",
				bucketPrefix: "/buckets/velero/",
				expectedKey:  "kopia/logpaste/kopia.blobcfg",
			},
			{
				name:         "kopia repository file",
				objectPath:   "kopia/logpaste/kopia.repository",
				expectedName: "kopia.repository",
				dir:          "/buckets/velero/kopia/logpaste",
				bucketPrefix: "/buckets/velero/",
				expectedKey:  "kopia/logpaste/kopia.repository",
			},
			{
				name:         "backup file in nested directory",
				objectPath:   "backups/test1/velero-backup.json",
				expectedName: "velero-backup.json",
				dir:          "/buckets/velero/backups/test1",
				bucketPrefix: "/buckets/velero/",
				expectedKey:  "backups/test1/velero-backup.json",
			},
			{
				name:         "file at bucket root",
				objectPath:   "file.txt",
				expectedName: "file.txt",
				dir:          "/buckets/velero",
				bucketPrefix: "/buckets/velero/",
				expectedKey:  "file.txt",
			},
			{
				name:         "deeply nested file",
				objectPath:   "a/b/c/d/e/file.json",
				expectedName: "file.json",
				dir:          "/buckets/velero/a/b/c/d/e",
				bucketPrefix: "/buckets/velero/",
				expectedKey:  "a/b/c/d/e/file.json",
			},
		}

		for _, tc := range testCases {
			t.Run(tc.name, func(t *testing.T) {
				// Simulate what the fix does: use path.Base to get just the filename
				entryName := path.Base(tc.objectPath)
				assert.Equal(t, tc.expectedName, entryName, "entry.Name should be base filename only")

				// Verify that combining dir + entry.Name produces correct key
				// This is what eachEntryFn does in the list operation
				key := (tc.dir + "/" + entryName)[len(tc.bucketPrefix):]
				assert.Equal(t, tc.expectedKey, key, "final key should not have doubled paths")
			})
		}
	})

	t.Run("verify path doubling bug is fixed", func(t *testing.T) {
		// This test demonstrates the exact bug that was fixed
		objectPath := "kopia/logpaste/kopia.blobcfg"
		dir := "/buckets/velero/kopia/logpaste"
		bucketPrefix := "/buckets/velero/"

		// WRONG behavior (before fix): entry.Name = full path
		wrongEntryName := objectPath
		wrongKey := (dir + "/" + wrongEntryName)[len(bucketPrefix):]
		assert.Equal(t, "kopia/logpaste/kopia/logpaste/kopia.blobcfg", wrongKey,
			"bug: using full path as entry.Name causes path doubling")

		// CORRECT behavior (after fix): entry.Name = base filename only
		correctEntryName := path.Base(objectPath)
		correctKey := (dir + "/" + correctEntryName)[len(bucketPrefix):]
		assert.Equal(t, "kopia/logpaste/kopia.blobcfg", correctKey,
			"fix: using path.Base for entry.Name produces correct path")
	})
}

// TestVersionedEntryCreation tests that versioned entries are created correctly
// with all necessary metadata for list operations.
func TestVersionedEntryCreation(t *testing.T) {
	t.Run("logical entry should have correct attributes", func(t *testing.T) {
		// Simulate creating a logical entry for a versioned object
		objectPath := "kopia/logpaste/kopia.repository"
		
		// Create a mock versioned entry (as would be returned from .versions directory)
		mockVersionedEntry := &filer_pb.Entry{
			Name: "v_1880a8e61c70fd480b47df7f530b796b", // Version file name
			Attributes: &filer_pb.FuseAttributes{
				Mtime:    1234567890,
				FileSize: 1024,
			},
			Extended: map[string][]byte{
				s3_constants.ExtVersionIdKey: []byte("1880a8e61c70fd480b47df7f530b796b"),
				s3_constants.ExtETagKey:      []byte("\"abc123\""),
			},
		}

		// Simulate what getLatestVersionEntryForListOperation does with the fix
		logicalEntry := &filer_pb.Entry{
			Name:        path.Base(objectPath), // This is the fix!
			IsDirectory: false,
			Attributes:  mockVersionedEntry.Attributes,
			Extended:    mockVersionedEntry.Extended,
			Chunks:      mockVersionedEntry.Chunks,
		}

		// Verify the logical entry has the correct name
		assert.Equal(t, "kopia.repository", logicalEntry.Name,
			"logical entry should have base filename as Name")
		assert.False(t, logicalEntry.IsDirectory,
			"logical entry should not be a directory")
		assert.NotNil(t, logicalEntry.Extended,
			"logical entry should preserve extended attributes")
	})
}

// TestVersionsFolderPathHandling tests that .versions folder paths are handled correctly
func TestVersionsFolderPathHandling(t *testing.T) {
	t.Run("versions folder suffix handling", func(t *testing.T) {
		testCases := []struct {
			versionsDir    string
			expectedObject string
		}{
			{
				versionsDir:    "kopia.blobcfg.versions",
				expectedObject: "kopia.blobcfg",
			},
			{
				versionsDir:    "kopia.repository.versions",
				expectedObject: "kopia.repository",
			},
			{
				versionsDir:    "file.txt.versions",
				expectedObject: "file.txt",
			},
		}

		for _, tc := range testCases {
			t.Run(tc.versionsDir, func(t *testing.T) {
				// This is what the list handler does to extract object name
				baseObjectName := tc.versionsDir[:len(tc.versionsDir)-len(s3_constants.VersionsFolder)]
				assert.Equal(t, tc.expectedObject, baseObjectName,
					"object name should be correctly extracted from versions directory")
			})
		}
	})
}