Extract Esti utils

.github/workflows/esti.yaml

@@ -769,7 +769,6 @@ jobs:
       REGISTRY: ${{ needs.login-to-amazon-ecr.outputs.registry }}
       AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
       AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}
-      LAKEFS_DATABASE_CONNECTION_STRING: "" # Override lakeFS docker compose settings
     steps:
       - name: Check-out code
         uses: actions/checkout@v4

esti/auth_test.go

@@ -219,7 +219,7 @@ func TestCreateRepo_Unauthorized(t *testing.T) {
 
 	name := generateUniqueRepositoryName()
 	storageNamespace := generateUniqueStorageNamespace(name)
-	name = makeRepositoryName(name)
+	name = MakeRepositoryName(name)
 	groups := []string{"Readers", "Viewers"}
 
 	// map group names to IDs

esti/commit_test.go

@@ -168,7 +168,7 @@ func TestCommitReadOnlyRepo(t *testing.T) {
 	ctx := context.Background()
 	name := strings.ToLower(t.Name())
 	storageNamespace := generateUniqueStorageNamespace(name)
-	repoName := makeRepositoryName(name)
+	repoName := MakeRepositoryName(name)
 	resp, err := client.CreateRepositoryWithResponse(ctx, &apigen.CreateRepositoryParams{}, apigen.CreateRepositoryJSONRequestBody{
 		DefaultBranch:    apiutil.Ptr(mainBranch),
 		Name:             repoName,

esti/copy_test.go

@@ -131,7 +131,7 @@ func TestCopyObject(t *testing.T) {
 	t.Run("read-only repository", func(t *testing.T) {
 		name := strings.ToLower(t.Name())
 		storageNamespace := generateUniqueStorageNamespace(name)
-		repoName := makeRepositoryName(name)
+		repoName := MakeRepositoryName(name)
 		resp, err := client.CreateRepositoryWithResponse(ctx, &apigen.CreateRepositoryParams{}, apigen.CreateRepositoryJSONRequestBody{
 			DefaultBranch:    apiutil.Ptr(mainBranch),
 			Name:             repoName,

esti/delete_test.go

@@ -78,7 +78,7 @@ func TestDeleteObjectsReadOnlyRepository(t *testing.T) {
 	ctx := context.Background()
 	name := strings.ToLower(t.Name())
 	storageNamespace := generateUniqueStorageNamespace(name)
-	repoName := makeRepositoryName(name)
+	repoName := MakeRepositoryName(name)
 	resp, err := client.CreateRepositoryWithResponse(ctx, &apigen.CreateRepositoryParams{}, apigen.CreateRepositoryJSONRequestBody{
 		DefaultBranch:    apiutil.Ptr(mainBranch),
 		Name:             repoName,

esti/system_test.go → esti/esti_utils.go

@@ -11,10 +11,13 @@ import (
 	"net/url"
 	"path/filepath"
 	"regexp"
+	"slices"
 	"strings"
 	"testing"
 
+	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/go-openapi/swag"
+	"github.com/hashicorp/go-multierror"
 	"github.com/rs/xid"
 	"github.com/spf13/viper"
 	"github.com/stretchr/testify/require"
@@ -24,14 +27,32 @@ import (
 	"github.com/treeverse/lakefs/pkg/api/helpers"
 	"github.com/treeverse/lakefs/pkg/config"
 	"github.com/treeverse/lakefs/pkg/logging"
-	"golang.org/x/exp/slices"
 )
 
-const mainBranch = "main"
-
-const minHTTPErrorStatusCode = 400
+type ArrayFlags []string
+
+var (
+	logger      logging.Logger
+	client      apigen.ClientWithResponsesInterface
+	endpointURL string
+	svc         *s3.Client
+	server      *WebhookServer
+
+	metaClientJarPath  string
+	sparkImageTag      string
+	repositoriesToKeep ArrayFlags
+	groupsToKeep       ArrayFlags
+	usersToKeep        ArrayFlags
+	policiesToKeep     ArrayFlags
+)
 
 const (
+	DefaultAdminAccessKeyID     = "AKIAIOSFDNN7EXAMPLEQ"
+	DefaultAdminSecretAccessKey = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
+	AdminUsername               = "esti"
+
+	mainBranch               = "main"
+	minHTTPErrorStatusCode   = 400
 	ViperStorageNamespaceKey = "storage_namespace"
 	ViperBlockstoreType      = "blockstore_type"
 )
@@ -40,6 +61,177 @@ var errNotVerified = errors.New("lakeFS failed")
 
 var nonAlphanumericSequence = regexp.MustCompile("[^a-zA-Z0-9]+")
 
+func (i *ArrayFlags) String() string {
+	return strings.Join(*i, " ")
+}
+
+func (i *ArrayFlags) Set(value string) error {
+	*i = append(*i, value)
+	return nil
+}
+
+func EnvCleanup(client apigen.ClientWithResponsesInterface, repositoriesToKeep, groupsToKeep, usersToKeep, policiesToKeep ArrayFlags) error {
+	ctx := context.Background()
+	errRepos := DeleteAllRepositories(ctx, client, repositoriesToKeep)
+	errGroups := DeleteAllGroups(ctx, client, groupsToKeep)
+	errPolicies := DeleteAllPolicies(ctx, client, policiesToKeep)
+	errUsers := DeleteAllUsers(ctx, client, usersToKeep)
+	return multierror.Append(errRepos, errGroups, errPolicies, errUsers).ErrorOrNil()
+}
+
+func DeleteAllRepositories(ctx context.Context, client apigen.ClientWithResponsesInterface, repositoriesToKeep ArrayFlags) error {
+	// collect repositories to delete
+	var (
+		repositoriesToDelete []string
+		nextOffset           string
+	)
+
+	for {
+		resp, err := client.ListRepositoriesWithResponse(ctx, &apigen.ListRepositoriesParams{After: apiutil.Ptr(apigen.PaginationAfter(nextOffset))})
+		if err != nil {
+			return fmt.Errorf("list repositories: %w", err)
+		}
+		if resp.StatusCode() != http.StatusOK {
+			return fmt.Errorf("list repositories: status: %s", resp.Status())
+		}
+		for _, repo := range resp.JSON200.Results {
+			if !slices.Contains(repositoriesToKeep, repo.Id) {
+				repositoriesToDelete = append(repositoriesToDelete, repo.Id)
+			}
+		}
+		if !resp.JSON200.Pagination.HasMore {
+			break
+		}
+		nextOffset = resp.JSON200.Pagination.NextOffset
+	}
+
+	var errs *multierror.Error
+	for _, id := range repositoriesToDelete {
+		resp, err := client.DeleteRepositoryWithResponse(ctx, id, &apigen.DeleteRepositoryParams{})
+		if err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("delete repository: %s, err: %w", id, err))
+		} else if resp.StatusCode() != http.StatusNoContent {
+			errs = multierror.Append(errs, fmt.Errorf("delete repository: %s, status: %s", id, resp.Status()))
+		}
+
+	}
+	return errs.ErrorOrNil()
+}
+
+func DeleteAllGroups(ctx context.Context, client apigen.ClientWithResponsesInterface, groupsToKeep ArrayFlags) error {
+	// list groups to delete
+	var (
+		groupsToDelete []string
+		nextOffset     string
+	)
+	for {
+		resp, err := client.ListGroupsWithResponse(ctx, &apigen.ListGroupsParams{After: apiutil.Ptr(apigen.PaginationAfter(nextOffset))})
+		if err != nil {
+			return fmt.Errorf("list groups: %w", err)
+		}
+		if resp.StatusCode() != http.StatusOK {
+			return fmt.Errorf("list groups: status: %s", resp.Status())
+		}
+		for _, group := range resp.JSON200.Results {
+			if !slices.Contains(groupsToKeep, swag.StringValue(group.Name)) {
+				groupsToDelete = append(groupsToDelete, group.Id)
+			}
+		}
+		if !resp.JSON200.Pagination.HasMore {
+			break
+		}
+		nextOffset = resp.JSON200.Pagination.NextOffset
+	}
+
+	// delete groups
+	var errs *multierror.Error
+	for _, id := range groupsToDelete {
+		resp, err := client.DeleteGroupWithResponse(ctx, id)
+		if err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("delete group: %s, err: %w", id, err))
+		} else if resp.StatusCode() != http.StatusNoContent {
+			errs = multierror.Append(errs, fmt.Errorf("delete group: %s, status: %s", id, resp.Status()))
+		}
+	}
+	return errs.ErrorOrNil()
+}
+
+func DeleteAllUsers(ctx context.Context, client apigen.ClientWithResponsesInterface, usersToKeep ArrayFlags) error {
+	// collect users to delete
+	var (
+		usersToDelete []string
+		nextOffset    string
+	)
+	for {
+		resp, err := client.ListUsersWithResponse(ctx, &apigen.ListUsersParams{After: apiutil.Ptr(apigen.PaginationAfter(nextOffset))})
+		if err != nil {
+			return fmt.Errorf("list users: %s", err)
+		}
+		if resp.JSON200 == nil {
+			return fmt.Errorf("list users, status: %s", resp.Status())
+		}
+		for _, user := range resp.JSON200.Results {
+			if !slices.Contains(usersToKeep, user.Id) {
+				usersToKeep = append(usersToDelete, user.Id)
+			}
+		}
+		if !resp.JSON200.Pagination.HasMore {
+			break
+		}
+		nextOffset = resp.JSON200.Pagination.NextOffset
+	}
+
+	// delete users
+	var errs *multierror.Error
+	for _, id := range usersToDelete {
+		resp, err := client.DeleteUserWithResponse(ctx, id)
+		if err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("delete user %s: %w", id, err))
+		} else if resp.StatusCode() != http.StatusNoContent {
+			errs = multierror.Append(errs, fmt.Errorf("delete user %s, status: %s", id, resp.Status()))
+		}
+	}
+	return errs.ErrorOrNil()
+}
+
+func DeleteAllPolicies(ctx context.Context, client apigen.ClientWithResponsesInterface, policiesToKeep ArrayFlags) error {
+	// list policies to delete
+	var (
+		policiesToDelete []string
+		nextOffset       string
+	)
+	for {
+		resp, err := client.ListPoliciesWithResponse(ctx, &apigen.ListPoliciesParams{After: apiutil.Ptr(apigen.PaginationAfter(nextOffset))})
+		if err != nil {
+			return fmt.Errorf("list policies: %w", err)
+		}
+		if resp.JSON200 == nil {
+			return fmt.Errorf("list policies, status: %s", resp.Status())
+		}
+		for _, policy := range resp.JSON200.Results {
+			if !slices.Contains(policiesToKeep, policy.Id) {
+				policiesToDelete = append(policiesToDelete, policy.Id)
+			}
+		}
+		if !resp.JSON200.Pagination.HasMore {
+			break
+		}
+		nextOffset = resp.JSON200.Pagination.NextOffset
+	}
+
+	// delete policies
+	var errs *multierror.Error
+	for _, id := range policiesToDelete {
+		resp, err := client.DeletePolicyWithResponse(ctx, id)
+		if err != nil {
+			errs = multierror.Append(errs, fmt.Errorf("delete policy %s: %w", id, err))
+		} else if resp.StatusCode() != http.StatusNoContent {
+			errs = multierror.Append(errs, fmt.Errorf("delete policy %s, status: %s", id, resp.Status()))
+		}
+	}
+	return errs.ErrorOrNil()
+}
+
 // skipOnSchemaMismatch matches the rawURL schema to the current tested storage namespace schema
 func skipOnSchemaMismatch(t *testing.T, rawURL string) {
 	t.Helper()
@@ -66,15 +258,15 @@ func verifyResponse(resp *http.Response, body []byte) error {
 	return nil
 }
 
-// makeRepositoryName changes name to make it an acceptable repository name by replacing all
+// MakeRepositoryName changes name to make it an acceptable repository name by replacing all
 // non-alphanumeric characters with a `-`.
-func makeRepositoryName(name string) string {
+func MakeRepositoryName(name string) string {
 	return nonAlphanumericSequence.ReplaceAllString(name, "-")
 }
 
 func setupTest(t testing.TB) (context.Context, logging.Logger, string) {
 	ctx := context.Background()
-	name := makeRepositoryName(t.Name())
+	name := MakeRepositoryName(t.Name())
 	log := logger.WithField("testName", name)
 	repo := createRepositoryForTest(ctx, t)
 	log.WithField("repo", repo).Info("Created repository")
@@ -93,14 +285,14 @@ func createRepositoryForTest(ctx context.Context, t testing.TB) string {
 
 func createRepositoryByName(ctx context.Context, t testing.TB, name string) string {
 	storageNamespace := generateUniqueStorageNamespace(name)
-	name = makeRepositoryName(name)
+	name = MakeRepositoryName(name)
 	createRepository(ctx, t, name, storageNamespace, false)
 	return name
 }
 
 func createReadOnlyRepositoryByName(ctx context.Context, t testing.TB, name string) string {
 	storageNamespace := generateUniqueStorageNamespace(name)
-	name = makeRepositoryName(name)
+	name = MakeRepositoryName(name)
 	createRepository(ctx, t, name, storageNamespace, true)
 	return name
 }

esti/hooks_failure_test.go

@@ -7,7 +7,6 @@ import (
 	"net/http"
 	"testing"
 	"text/template"
-	"time"
 
 	"github.com/stretchr/testify/require"
 	"github.com/treeverse/lakefs/pkg/api/apigen"
@@ -44,8 +43,6 @@ hooks:
       timeout : {{.Timeout}}
 `))
 
-const hooksTimeout = 2 * time.Second
-
 func TestHooksTimeout(t *testing.T) {
 	hookFailToCommit(t, "timeout")
 }