Merge branch 'pks-backup-fix-test-with-atomic-repo-creation' into 'master'

backup: Context fixes for tests

See merge request gitlab-org/gitaly!4185

8 files changed, 130 insertions, 21 deletions

diff --git a/cmd/gitaly-backup/create_test.go b/cmd/gitaly-backup/create_test.go
index 957005e0b..047da4cbf 100644
--- a/cmd/gitaly-backup/create_test.go
+++ b/cmd/gitaly-backup/create_test.go
@@ -2,7 +2,6 @@ package main
 
 import (
 	"bytes"
-	"context"
 	"encoding/json"
 	"flag"
 	"fmt"
@@ -58,9 +57,12 @@ func TestCreateSubcommand(t *testing.T) {
 	fs := flag.NewFlagSet("create", flag.ContinueOnError)
 	cmd.Flags(fs)
 
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
 	require.NoError(t, fs.Parse([]string{"-path", path}))
 	require.EqualError(t,
-		cmd.Run(context.Background(), &stdin, io.Discard),
+		cmd.Run(ctx, &stdin, io.Discard),
 		"create: pipeline: 1 failures encountered:\n - invalid: manager: isEmpty: could not dial source: invalid connection string: \"invalid\"\n")
 
 	for _, repo := range repos {
diff --git a/cmd/gitaly-backup/restore_test.go b/cmd/gitaly-backup/restore_test.go
index 0b309db30..0c7828c0e 100644
--- a/cmd/gitaly-backup/restore_test.go
+++ b/cmd/gitaly-backup/restore_test.go
@@ -7,12 +7,15 @@ import (
 	"flag"
 	"fmt"
 	"io"
+	"os"
 	"path/filepath"
 	"testing"
 
 	"github.com/stretchr/testify/require"
+	"gitlab.com/gitlab-org/gitaly/v14/client"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/git/gittest"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/gitaly/service/setup"
+	"gitlab.com/gitlab-org/gitaly/v14/internal/metadata/featureflag"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/testhelper"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/testhelper/testcfg"
 	"gitlab.com/gitlab-org/gitaly/v14/internal/testhelper/testserver"
@@ -20,6 +23,16 @@ import (
 )
 
 func TestRestoreSubcommand(t *testing.T) {
+	t.Parallel()
+	testhelper.NewFeatureSets(
+		featureflag.AtomicRemoveRepository,
+		featureflag.TxAtomicRepositoryCreation,
+	).Run(t, testRestoreSubcommand)
+}
+
+func testRestoreSubcommand(t *testing.T, ctx context.Context) {
+	t.Parallel()
+
 	cfg := testcfg.Build(t)
 	testcfg.BuildGitalyHooks(t, cfg)
 
@@ -27,8 +40,28 @@ func TestRestoreSubcommand(t *testing.T) {
 
 	path := testhelper.TempDir(t)
 
-	existingRepo, existRepoPath := gittest.CloneRepo(t, cfg, cfg.Storages[0], gittest.CloneRepoOpts{
+	conn, err := client.Dial(gitalyAddr, nil)
+	require.NoError(t, err)
+	defer testhelper.MustClose(t, conn)
+
+	existingRepo := &gitalypb.Repository{
 		RelativePath: "existing_repo",
+		StorageName:  cfg.Storages[0].Name,
+	}
+
+	// We need to create the repository entry in the database such that we can call
+	// `RemoveRepository()` and have Praefect forward the request to Gitaly as expected.
+	repoClient := gitalypb.NewRepositoryServiceClient(conn)
+	_, err = repoClient.CreateRepository(ctx, &gitalypb.CreateRepositoryRequest{Repository: existingRepo})
+	require.NoError(t, err)
+	existingRepo.RelativePath = testhelper.GetReplicaPath(ctx, t, conn, existingRepo)
+
+	// This is a bit awkward: we have to delete the created repository on-disk again and
+	// recreate it with the seed repository.
+	existRepoPath := filepath.Join(cfg.Storages[0].Path, existingRepo.RelativePath)
+	require.NoError(t, os.RemoveAll(existRepoPath))
+	gittest.CloneRepo(t, cfg, cfg.Storages[0], gittest.CloneRepoOpts{
+		RelativePath: existingRepo.RelativePath,
 	})
 	existingRepoBundlePath := filepath.Join(path, existingRepo.RelativePath+".bundle")
 	gittest.Exec(t, cfg, "-C", existRepoPath, "bundle", "create", existingRepoBundlePath, "--all")
@@ -67,7 +100,7 @@ func TestRestoreSubcommand(t *testing.T) {
 
 	require.NoError(t, fs.Parse([]string{"-path", path}))
 	require.EqualError(t,
-		cmd.Run(context.Background(), &stdin, io.Discard),
+		cmd.Run(ctx, &stdin, io.Discard),
 		"restore: pipeline: 1 failures encountered:\n - invalid: manager: remove repository: could not dial source: invalid connection string: \"invalid\"\n")
 
 	for _, repo := range repos {
diff --git a/internal/backup/backup_test.go b/internal/backup/backup_test.go
index ad46e13f6..9031cf83e 100644
--- a/internal/backup/backup_test.go
+++ b/internal/backup/backup_test.go
@@ -29,6 +29,8 @@ import (
 )
 
 func TestManager_Create(t *testing.T) {
+	t.Parallel()
+
 	const backupID = "abc123"
 
 	cfg := testcfg.Build(t)
@@ -150,6 +152,8 @@ func TestManager_Create(t *testing.T) {
 }
 
 func TestManager_Create_incremental(t *testing.T) {
+	t.Parallel()
+
 	const backupID = "abc123"
 
 	cfg := testcfg.Build(t)
@@ -261,6 +265,8 @@ func TestManager_Create_incremental(t *testing.T) {
 }
 
 func TestManager_Restore(t *testing.T) {
+	t.Parallel()
+
 	cfg := testcfg.Build(t)
 	testcfg.BuildGitalyHooks(t, cfg)
 
@@ -270,6 +276,8 @@ func TestManager_Restore(t *testing.T) {
 }
 
 func TestManager_Restore_praefect(t *testing.T) {
+	t.Parallel()
+
 	gitalyCfg := testcfg.Build(t, testcfg.WithStorages("gitaly-1"))
 
 	testcfg.BuildPraefect(t, gitalyCfg)
@@ -324,11 +332,14 @@ func TestManager_Restore_praefect(t *testing.T) {
 }
 
 func testManagerRestore(t *testing.T, cfg config.Cfg, gitalyAddr string) {
-	testhelper.NewFeatureSets(
-		featureflag.AtomicRemoveRepository,
-		featureflag.TxAtomicRepositoryCreation,
-	).Run(t, func(t *testing.T, ctx context.Context) {
-		testManagerRestoreWithContext(t, ctx, cfg, gitalyAddr)
+	t.Run("parallel", func(t *testing.T) {
+		testhelper.NewFeatureSets(
+			featureflag.AtomicRemoveRepository,
+			featureflag.TxAtomicRepositoryCreation,
+		).Run(t, func(t *testing.T, ctx context.Context) {
+			t.Parallel()
+			testManagerRestoreWithContext(t, ctx, cfg, gitalyAddr)
+		})
 	})
 }
 
@@ -339,12 +350,12 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 
 	repoClient := gitalypb.NewRepositoryServiceClient(cc)
 
-	createRepo := func(t testing.TB, relativePath string) *gitalypb.Repository {
+	createRepo := func(t testing.TB) *gitalypb.Repository {
 		t.Helper()
 
 		repo := &gitalypb.Repository{
 			StorageName:  "default",
-			RelativePath: gittest.NewRepositoryName(t, false) + relativePath,
+			RelativePath: gittest.NewRepositoryName(t, false),
 		}
 
 		for i := 0; true; i++ {
@@ -382,7 +393,7 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			desc:     "existing repo, without hooks",
 			locators: []string{"legacy", "pointer"},
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
-				repo := createRepo(t, "existing")
+				repo := createRepo(t)
 				require.NoError(t, os.MkdirAll(filepath.Join(path, repo.RelativePath), os.ModePerm))
 				bundlePath := filepath.Join(path, repo.RelativePath+".bundle")
 				gittest.BundleTestRepo(t, cfg, "gitlab-test.git", bundlePath)
@@ -395,7 +406,7 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			desc:     "existing repo, with hooks",
 			locators: []string{"legacy", "pointer"},
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
-				repo := createRepo(t, "existing_hooks")
+				repo := createRepo(t)
 				bundlePath := filepath.Join(path, repo.RelativePath+".bundle")
 				customHooksPath := filepath.Join(path, repo.RelativePath, "custom_hooks.tar")
 				require.NoError(t, os.MkdirAll(filepath.Join(path, repo.RelativePath), os.ModePerm))
@@ -415,7 +426,7 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			desc:     "missing bundle",
 			locators: []string{"legacy", "pointer"},
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
-				repo := createRepo(t, "missing_bundle")
+				repo := createRepo(t)
 				return repo, nil
 			},
 			expectedErrAs: ErrSkipped,
@@ -424,7 +435,7 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			desc:     "missing bundle, always create",
 			locators: []string{"legacy", "pointer"},
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
-				repo := createRepo(t, "missing_bundle_always_create")
+				repo := createRepo(t)
 				return repo, new(git.Checksum)
 			},
 			alwaysCreate: true,
@@ -436,8 +447,10 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
 				repo := &gitalypb.Repository{
 					StorageName:  "default",
-					RelativePath: "nonexistent",
+					RelativePath: gittest.NewRepositoryName(t, false),
 				}
+
+				require.NoError(t, os.MkdirAll(filepath.Dir(filepath.Join(path, repo.RelativePath)), os.ModePerm))
 				bundlePath := filepath.Join(path, repo.RelativePath+".bundle")
 				gittest.BundleTestRepo(t, cfg, "gitlab-test.git", bundlePath)
 
@@ -450,7 +463,7 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			locators: []string{"pointer"},
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
 				const backupID = "abc123"
-				repo := createRepo(t, "incremental")
+				repo := createRepo(t)
 				repoBackupPath := filepath.Join(path, repo.RelativePath)
 				backupPath := filepath.Join(repoBackupPath, backupID)
 				require.NoError(t, os.MkdirAll(backupPath, os.ModePerm))
@@ -469,10 +482,10 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 			setup: func(t testing.TB) (*gitalypb.Repository, *git.Checksum) {
 				const backupID = "abc123"
 
-				expected := createRepo(t, "expected")
+				expected := createRepo(t)
 				expectedRepoPath := filepath.Join(cfg.Storages[0].Path, expected.RelativePath)
 
-				repo := createRepo(t, "incremental")
+				repo := createRepo(t)
 				repoBackupPath := filepath.Join(path, repo.RelativePath)
 				backupPath := filepath.Join(repoBackupPath, backupID)
 				require.NoError(t, os.MkdirAll(backupPath, os.ModePerm))
@@ -579,12 +592,17 @@ func testManagerRestoreWithContext(t *testing.T, ctx context.Context, cfg config
 }
 
 func TestResolveSink(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := testhelper.Context()
+	defer cancel()
+
 	isStorageServiceSink := func(expErrMsg string) func(t *testing.T, sink Sink) {
 		return func(t *testing.T, sink Sink) {
 			t.Helper()
 			sssink, ok := sink.(*StorageServiceSink)
 			require.True(t, ok)
-			_, err := sssink.bucket.List(nil).Next(context.TODO())
+			_, err := sssink.bucket.List(nil).Next(ctx)
 			ierr, ok := err.(interface{ Unwrap() error })
 			require.True(t, ok)
 			terr := ierr.Unwrap()
@@ -660,7 +678,7 @@ func TestResolveSink(t *testing.T) {
 			for k, v := range tc.envs {
 				t.Cleanup(testhelper.ModifyEnvironment(t, k, v))
 			}
-			sink, err := ResolveSink(context.TODO(), tc.path)
+			sink, err := ResolveSink(ctx, tc.path)
 			if tc.errMsg != "" {
 				require.EqualError(t, err, tc.errMsg)
 				return
@@ -671,6 +689,8 @@ func TestResolveSink(t *testing.T) {
 }
 
 func TestResolveLocator(t *testing.T) {
+	t.Parallel()
+
 	for _, tc := range []struct {
 		layout      string
 		expectedErr string
diff --git a/internal/backup/filesystem_sink_test.go b/internal/backup/filesystem_sink_test.go
index 86be459a1..6c0fe6030 100644
--- a/internal/backup/filesystem_sink_test.go
+++ b/internal/backup/filesystem_sink_test.go
@@ -13,7 +13,11 @@ import (
 )
 
 func TestFilesystemSink_GetReader(t *testing.T) {
+	t.Parallel()
+
 	t.Run("ok", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
@@ -33,6 +37,8 @@ func TestFilesystemSink_GetReader(t *testing.T) {
 	})
 
 	t.Run("no file", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
@@ -47,7 +53,11 @@ func TestFilesystemSink_GetReader(t *testing.T) {
 }
 
 func TestFilesystemSink_Write(t *testing.T) {
+	t.Parallel()
+
 	t.Run("ok", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
@@ -64,6 +74,8 @@ func TestFilesystemSink_Write(t *testing.T) {
 	})
 
 	t.Run("overrides existing data", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
@@ -84,6 +96,8 @@ func TestFilesystemSink_Write(t *testing.T) {
 	})
 
 	t.Run("dir creation error", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
diff --git a/internal/backup/lazy_test.go b/internal/backup/lazy_test.go
index 5c4db0c1e..e8964faea 100644
--- a/internal/backup/lazy_test.go
+++ b/internal/backup/lazy_test.go
@@ -13,6 +13,8 @@ import (
 )
 
 func TestLazyWrite_noData(t *testing.T) {
+	t.Parallel()
+
 	ctx, cancel := testhelper.Context()
 	defer cancel()
 
@@ -30,6 +32,8 @@ func TestLazyWrite_noData(t *testing.T) {
 }
 
 func TestLazyWrite_data(t *testing.T) {
+	t.Parallel()
+
 	ctx, cancel := testhelper.Context()
 	defer cancel()
 
diff --git a/internal/backup/locator_test.go b/internal/backup/locator_test.go
index 7dd9bc454..17e4f859a 100644
--- a/internal/backup/locator_test.go
+++ b/internal/backup/locator_test.go
@@ -15,10 +15,14 @@ import (
 )
 
 func TestLegacyLocator(t *testing.T) {
+	t.Parallel()
+
 	_, repo, _ := testcfg.BuildWithRepo(t)
 	l := LegacyLocator{}
 
 	t.Run("Begin/Commit Full", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
@@ -36,6 +40,8 @@ func TestLegacyLocator(t *testing.T) {
 	})
 
 	t.Run("FindLatest", func(t *testing.T) {
+		t.Parallel()
+
 		ctx, cancel := testhelper.Context()
 		defer cancel()
 
@@ -58,11 +64,15 @@ func TestLegacyLocator(t *testing.T) {
 }
 
 func TestPointerLocator(t *testing.T) {
+	t.Parallel()
+
 	const backupID = "abc123"
 
 	_, repo, _ := testcfg.BuildWithRepo(t)
 
 	t.Run("Begin/Commit full", func(t *testing.T) {
+		t.Parallel()
+
 		backupPath := testhelper.TempDir(t)
 		var l Locator = PointerLocator{
 			Sink: NewFilesystemSink(backupPath),
@@ -91,6 +101,8 @@ func TestPointerLocator(t *testing.T) {
 	})
 
 	t.Run("Begin/Commit incremental", func(t *testing.T) {
+		t.Parallel()
+
 		const fallbackBackupID = "fallback123"
 
 		for _, tc := range []struct {
@@ -114,6 +126,8 @@ func TestPointerLocator(t *testing.T) {
 			},
 		} {
 			t.Run(tc.desc, func(t *testing.T) {
+				t.Parallel()
+
 				backupPath := testhelper.TempDir(t)
 				sink := NewFilesystemSink(backupPath)
 				var l Locator = PointerLocator{Sink: sink}
@@ -157,7 +171,11 @@ func TestPointerLocator(t *testing.T) {
 	})
 
 	t.Run("FindLatest", func(t *testing.T) {
+		t.Parallel()
+
 		t.Run("no fallback", func(t *testing.T) {
+			t.Parallel()
+
 			backupPath := testhelper.TempDir(t)
 			var l Locator = PointerLocator{
 				Sink: NewFilesystemSink(backupPath),
@@ -200,6 +218,8 @@ func TestPointerLocator(t *testing.T) {
 		})
 
 		t.Run("fallback", func(t *testing.T) {
+			t.Parallel()
+
 			backupPath := testhelper.TempDir(t)
 			var l Locator = PointerLocator{
 				Sink:     NewFilesystemSink(backupPath),
@@ -243,6 +263,8 @@ func TestPointerLocator(t *testing.T) {
 		})
 
 		t.Run("invalid backup LATEST", func(t *testing.T) {
+			t.Parallel()
+
 			backupPath := testhelper.TempDir(t)
 			var l Locator = PointerLocator{
 				Sink: NewFilesystemSink(backupPath),
@@ -261,6 +283,8 @@ func TestPointerLocator(t *testing.T) {
 		})
 
 		t.Run("invalid incremental LATEST", func(t *testing.T) {
+			t.Parallel()
+
 			backupPath := testhelper.TempDir(t)
 			var l Locator = PointerLocator{
 				Sink: NewFilesystemSink(backupPath),
diff --git a/internal/backup/pipeline_test.go b/internal/backup/pipeline_test.go
index 44a90afb7..afe844f9f 100644
--- a/internal/backup/pipeline_test.go
+++ b/internal/backup/pipeline_test.go
@@ -16,12 +16,16 @@ import (
 )
 
 func TestLoggingPipeline(t *testing.T) {
+	t.Parallel()
+
 	testPipeline(t, func() Pipeline {
 		return NewLoggingPipeline(logrus.StandardLogger())
 	})
 }
 
 func TestParallelPipeline(t *testing.T) {
+	t.Parallel()
+
 	testPipeline(t, func() Pipeline {
 		return NewParallelPipeline(NewLoggingPipeline(logrus.StandardLogger()), 2, 0)
 	})
@@ -116,6 +120,8 @@ func (s MockStrategy) Restore(ctx context.Context, req *RestoreRequest) error {
 
 func testPipeline(t *testing.T, init func() Pipeline) {
 	t.Run("create command", func(t *testing.T) {
+		t.Parallel()
+
 		strategy := MockStrategy{
 			CreateFunc: func(_ context.Context, req *CreateRequest) error {
 				switch req.Repository.StorageName {
@@ -148,6 +154,8 @@ func testPipeline(t *testing.T, init func() Pipeline) {
 	})
 
 	t.Run("restore command", func(t *testing.T) {
+		t.Parallel()
+
 		strategy := MockStrategy{
 			RestoreFunc: func(_ context.Context, req *RestoreRequest) error {
 				switch req.Repository.StorageName {
@@ -181,6 +189,8 @@ func testPipeline(t *testing.T, init func() Pipeline) {
 }
 
 func TestPipelineError(t *testing.T) {
+	t.Parallel()
+
 	for _, tc := range []struct {
 		name          string
 		repos         []*gitalypb.Repository
diff --git a/internal/backup/storage_service_sink_test.go b/internal/backup/storage_service_sink_test.go
index 5b740df46..3289bf2a5 100644
--- a/internal/backup/storage_service_sink_test.go
+++ b/internal/backup/storage_service_sink_test.go
@@ -12,6 +12,8 @@ import (
 )
 
 func TestStorageServiceSink(t *testing.T) {
+	t.Parallel()
+
 	ctx, cancel := testhelper.Context()
 	defer cancel()