ci: Parallelize change detector

Makefile

@@ -34,6 +34,7 @@ TEST_FLAGS=-race -shuffle=on -timeout 300s
 PLAYGROUND_DIRECTORY=playground
 LENS_TEST_DIRECTORY=tests/integration/schema/migrations
 CLI_TEST_DIRECTORY=tests/integration/cli
+CHANGE_DETECTOR_TEST_DIRECTORY=tests/change_detector
 DEFAULT_TEST_DIRECTORIES=$$(go list ./... | grep -v -e $(LENS_TEST_DIRECTORY) -e $(CLI_TEST_DIRECTORY))
 
 default:
@@ -294,8 +295,7 @@ test\:coverage-html:
 
 .PHONY: test\:changes
 test\:changes:
-	@$(MAKE) deps:lens
-	env DEFRA_DETECT_DATABASE_CHANGES=true DEFRA_CLIENT_GO=true gotestsum -- ./... -shuffle=on -p 1
+	gotestsum --format testname -- ./$(CHANGE_DETECTOR_TEST_DIRECTORY)/... --tags change_detector
 
 .PHONY: validate\:codecov
 validate\:codecov:

tests/change_detector/README.md

@@ -0,0 +1,15 @@
+# Change Detector
+
+The change detector is used to detect data format changes between versions of DefraDB.
+
+## How it works
+
+The tests run using a `source` and `target` branch of DefraDB. Each branch is cloned into a temporary directory and dependencies are installed.
+
+The test runner executes all of the common test packages available in the `source` and `target` tests directory.
+
+For each test package execution the following steps occur:
+
+- Create a temporary data directory. This is used to share data between `source` and `target`.
+- Run the `source` version in setup only mode. This creates test fixtures in the shared data directory.
+- Run the `target` version in change detector mode. This skips the setup and executes the tests using the shared data directory.

tests/change_detector/change_detector_test.go

@@ -0,0 +1,132 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+//go:build change_detector
+
+package change_detector
+
+import (
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestChanges(t *testing.T) {
+	sourceRepoDir := t.TempDir()
+	execClone(t, sourceRepoDir, repository, sourceBranch)
+
+	if checkIfDatabaseFormatChangesAreDocumented(t, sourceRepoDir) {
+		t.Skip("skipping test with documented database format changes")
+	}
+
+	var targetRepoDir string
+	if targetBranch == "" {
+		// default to the local branch
+		out, err := exec.Command("git", "rev-parse", "--show-toplevel").Output()
+		require.NoError(t, err, string(out))
+		targetRepoDir = strings.TrimSpace(string(out))
+	} else {
+		// check out the target branch
+		targetRepoDir = t.TempDir()
+		execClone(t, targetRepoDir, repository, targetBranch)
+		execMakeDeps(t, targetRepoDir)
+	}
+
+	execMakeDeps(t, sourceRepoDir)
+
+	targetRepoTestDir := filepath.Join(targetRepoDir, "tests", "integration")
+	targetRepoPkgList := execList(t, targetRepoTestDir)
+
+	sourceRepoTestDir := filepath.Join(sourceRepoDir, "tests", "integration")
+	sourceRepoPkgList := execList(t, sourceRepoTestDir)
+
+	sourceRepoPkgMap := make(map[string]bool)
+	for _, pkg := range sourceRepoPkgList {
+		sourceRepoPkgMap[pkg] = true
+	}
+
+	for _, pkg := range targetRepoPkgList {
+		pkgName := strings.TrimPrefix(pkg, "github.com/sourcenetwork/defradb/")
+		t.Run(pkgName, func(t *testing.T) {
+			if pkg == "" || !sourceRepoPkgMap[pkg] {
+				t.Skip("skipping unknown or new test package")
+			}
+
+			t.Parallel()
+			dataDir := t.TempDir()
+
+			sourceTestPkg := filepath.Join(sourceRepoDir, pkgName)
+			execTest(t, sourceTestPkg, dataDir, true)
+
+			targetTestPkg := filepath.Join(targetRepoDir, pkgName)
+			execTest(t, targetTestPkg, dataDir, false)
+		})
+	}
+}
+
+// execList returns a list of all packages in the given directory.
+func execList(t *testing.T, dir string) []string {
+	cmd := exec.Command("go", "list", "./...")
+	cmd.Dir = dir
+
+	out, err := cmd.Output()
+	require.NoError(t, err, string(out))
+
+	return strings.Split(string(out), "\n")
+}
+
+// execTest runs the tests in the given directory and sets the data
+// directory and setup only environment variables.
+func execTest(t *testing.T, dir, dataDir string, setupOnly bool) {
+	cmd := exec.Command("go", "test", ".", "-count", "1", "-v")
+	cmd.Dir = dir
+	cmd.Env = append(
+		os.Environ(),
+		fmt.Sprintf("%s=%s", enableEnvName, "true"),
+		fmt.Sprintf("%s=%s", rootDataDirEnvName, dataDir),
+	)
+
+	if setupOnly {
+		cmd.Env = append(cmd.Env, fmt.Sprintf("%s=%s", setupOnlyEnvName, "true"))
+	}
+
+	out, err := cmd.Output()
+	require.NoError(t, err, string(out))
+}
+
+// execClone clones the repo from the given url and branch into the directory.
+func execClone(t *testing.T, dir, url, branch string) {
+	cmd := exec.Command(
+		"git",
+		"clone",
+		"--single-branch",
+		"--branch", branch,
+		"--depth", "1",
+		url,
+		dir,
+	)
+
+	out, err := cmd.Output()
+	require.NoError(t, err, string(out))
+}
+
+// execMakeDeps runs make:deps in the given directory.
+func execMakeDeps(t *testing.T, dir string) {
+	cmd := exec.Command("make", "deps:lens")
+	cmd.Dir = dir
+
+	out, err := cmd.Output()
+	require.NoError(t, err, string(out))
+}

tests/change_detector/utils.go

@@ -0,0 +1,189 @@
+// Copyright 2023 Democratized Data Foundation
+//
+// Use of this software is governed by the Business Source License
+// included in the file licenses/BSL.txt.
+//
+// As of the Change Date specified in that file, in accordance with
+// the Business Source License, use of this software will be governed
+// by the Apache License, Version 2.0, included in the file
+// licenses/APL.txt.
+
+package change_detector
+
+import (
+	"io/fs"
+	"os"
+	"path"
+	"runtime"
+	"strconv"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+var (
+	// Enabled is true when the change detector is running.
+	Enabled bool
+	// SetupOnly is true when the change detector is running in setup mode.
+	SetupOnly bool
+	// rootDatabaseDir is the shared database directory for running tests.
+	rootDatabaseDir string
+	// repository is the url of the repository to run change detector on.
+	repository string
+	// sourceBranch is the name of the source branch to run change detector on.
+	sourceBranch string
+	// targetBranch is the name of the target branch to run change detector on.
+	targetBranch string
+	// previousTestCaseTestName is the name of the previous test.
+	previousTestCaseTestName string
+)
+
+const (
+	repositoryEnvName   = "DEFRA_CHANGE_DETECTOR_REPOSITORY"
+	sourceBranchEnvName = "DEFRA_CHANGE_DETECTOR_SOURCE_BRANCH"
+	targetBranchEnvName = "DEFRA_CHANGE_DETECTOR_TARGET_BRANCH"
+	setupOnlyEnvName    = "DEFRA_CHANGE_DETECTOR_SETUP_ONLY"
+	rootDataDirEnvName  = "DEFRA_CHANGE_DETECTOR_ROOT_DATA_DIR"
+	enableEnvName       = "DEFRA_CHANGE_DETECTOR_ENABLE"
+)
+
+const (
+	defaultRepository          = "https://github.com/nasdf/defradb.git"
+	defaultSourceBranch        = "nasdf/test/parallel-change-detector"
+	documentationDirectoryName = "data_format_changes"
+)
+
+func init() {
+	Enabled, _ = strconv.ParseBool(os.Getenv(enableEnvName))
+	SetupOnly, _ = strconv.ParseBool(os.Getenv(setupOnlyEnvName))
+	rootDatabaseDir = os.Getenv(rootDataDirEnvName)
+	targetBranch = os.Getenv(targetBranchEnvName)
+
+	if value, ok := os.LookupEnv(repositoryEnvName); ok {
+		repository = value
+	} else {
+		repository = defaultRepository
+	}
+
+	if value, ok := os.LookupEnv(sourceBranchEnvName); ok {
+		sourceBranch = value
+	} else {
+		sourceBranch = defaultSourceBranch
+	}
+}
+
+// DatabaseDir returns the database directory for change detector test.
+func DatabaseDir(t testing.TB) string {
+	return path.Join(rootDatabaseDir, t.Name())
+}
+
+// PreTestChecks skips any test that can't be run by the change detector.
+func PreTestChecks(t *testing.T, collectionNames []string) {
+	if !Enabled {
+		return
+	}
+
+	if previousTestCaseTestName == t.Name() {
+		t.Skip("skipping duplicate test")
+	}
+	previousTestCaseTestName = t.Name()
+
+	if len(collectionNames) == 0 {
+		t.Skip("skipping test with no collections")
+	}
+
+	if SetupOnly {
+		return
+	}
+
+	_, err := os.Stat(DatabaseDir(t))
+	if os.IsNotExist(err) {
+		t.Skip("skipping new test package")
+	}
+	require.NoError(t, err)
+}
+
+func checkIfDatabaseFormatChangesAreDocumented(t *testing.T, codeDir string) bool {
+	previousDbChangeFiles, targetDirFound := getDatabaseFormatDocumentation(t, codeDir, false)
+	if !targetDirFound {
+		t.Fatalf("Documentation directory not found")
+	}
+
+	previousDbChanges := make(map[string]struct{}, len(previousDbChangeFiles))
+	for _, f := range previousDbChangeFiles {
+		// Note: we assume flat directory for now - sub directories are not expanded
+		previousDbChanges[f.Name()] = struct{}{}
+	}
+
+	_, thisFilePath, _, _ := runtime.Caller(0)
+	currentDbChanges, currentDirFound := getDatabaseFormatDocumentation(t, thisFilePath, true)
+	if !currentDirFound {
+		t.Fatalf("Documentation directory not found")
+	}
+
+	for _, f := range currentDbChanges {
+		if _, isChangeOld := previousDbChanges[f.Name()]; !isChangeOld {
+			// If there is a new file in the directory then the change
+			// has been documented and the test should pass
+			return true
+		}
+	}
+
+	return false
+}
+
+func getDatabaseFormatDocumentation(t *testing.T, startPath string, allowDescend bool) ([]fs.DirEntry, bool) {
+	startInfo, err := os.Stat(startPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	var currentDirectory string
+	if startInfo.IsDir() {
+		currentDirectory = startPath
+	} else {
+		currentDirectory = path.Dir(startPath)
+	}
+
+	for {
+		directoryContents, err := os.ReadDir(currentDirectory)
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		for _, directoryItem := range directoryContents {
+			directoryItemPath := path.Join(currentDirectory, directoryItem.Name())
+			if directoryItem.Name() == documentationDirectoryName {
+				probableFormatChangeDirectoryContents, err := os.ReadDir(directoryItemPath)
+				if err != nil {
+					t.Fatal(err)
+				}
+				for _, possibleDocumentationItem := range probableFormatChangeDirectoryContents {
+					if path.Ext(possibleDocumentationItem.Name()) == ".md" {
+						// If the directory's name matches the expected, and contains .md files
+						// we assume it is the documentation directory
+						return probableFormatChangeDirectoryContents, true
+					}
+				}
+			} else {
+				if directoryItem.IsDir() {
+					childContents, directoryFound := getDatabaseFormatDocumentation(t, directoryItemPath, false)
+					if directoryFound {
+						return childContents, true
+					}
+				}
+			}
+		}
+
+		if allowDescend {
+			// If not found in this directory, continue down the path
+			currentDirectory = path.Dir(currentDirectory)
+
+			if currentDirectory == "." || currentDirectory == "/" {
+				t.Fatal(err)
+			}
+		} else {
+			return []fs.DirEntry{}, false
+		}
+	}
+}