chore(ci): make socket activation test more robust (#4985)

Objective: make the tests more robust and reduce random failures. Preliminary refactoring: - Accept testing.TB in testenv helpers. This is needed to use `require.EventuallyWithT` in socket activation tests. - Rename parameters for clarity Tests refactoring: - use t.Cleanup instead of defer where appropriate - create file handlers in test routine instead of go routines - remove unnecessary var declaration - increased wait time to 30 seconds. - allow running socket activation test on Darwin Ref: - #3283 - #3313 - #3318
2025-12-23 22:57:50 -05:00 · 2025-11-13 19:02:57 -08:00
parent c3c971a785
commit c60ebebb17
6 changed files with 172 additions and 171 deletions
--- a/internal/testutil/tmpdir.go
+++ b/internal/testutil/tmpdir.go
@@ -92,12 +92,12 @@ func TempDirectoryShort(tb testing.TB) string {

 // TempLogDirectory returns a temporary directory used for storing logs.
 // If KOPIA_LOGS_DIR is provided.
-func TempLogDirectory(t *testing.T) string {
-	t.Helper()
+func TempLogDirectory(tb testing.TB) string {
+	tb.Helper()

-	cleanName := strings.NewReplacer("/", "_", "\\", "_", ":", "_").Replace(t.Name())
+	cleanName := strings.NewReplacer("/", "_", "\\", "_", ":", "_").Replace(tb.Name())

-	t.Helper()
+	tb.Helper()

 	logsBaseDir := os.Getenv("KOPIA_LOGS_DIR")
 	if logsBaseDir == "" {
@@ -106,16 +106,16 @@ func TempLogDirectory(t *testing.T) string {

 	logsDir := filepath.Join(logsBaseDir, cleanName+"."+clock.Now().Local().Format("20060102150405"))

-	require.NoError(t, os.MkdirAll(logsDir, logsDirPermissions))
+	require.NoError(tb, os.MkdirAll(logsDir, logsDirPermissions))

-	t.Cleanup(func() {
+	tb.Cleanup(func() {
 		if os.Getenv("KOPIA_KEEP_LOGS") != "" {
-			t.Logf("logs preserved in %v", logsDir)
+			tb.Logf("logs preserved in %v", logsDir)
 			return
 		}

-		if t.Failed() && os.Getenv("KOPIA_DISABLE_LOG_DUMP_ON_FAILURE") == "" {
-			dumpLogs(t, logsDir)
+		if tb.Failed() && os.Getenv("KOPIA_DISABLE_LOG_DUMP_ON_FAILURE") == "" {
+			dumpLogs(tb, logsDir)
 		}

 		os.RemoveAll(logsDir) //nolint:errcheck
@@ -124,36 +124,36 @@ func TempLogDirectory(t *testing.T) string {
 	return logsDir
 }

-func dumpLogs(t *testing.T, dirname string) {
-	t.Helper()
+func dumpLogs(tb testing.TB, dirname string) {
+	tb.Helper()

 	entries, err := os.ReadDir(dirname)
 	if err != nil {
-		t.Errorf("unable to read %v: %v", dirname, err)
+		tb.Errorf("unable to read %v: %v", dirname, err)

 		return
 	}

 	for _, e := range entries {
 		if e.IsDir() {
-			dumpLogs(t, filepath.Join(dirname, e.Name()))
+			dumpLogs(tb, filepath.Join(dirname, e.Name()))
 			continue
 		}

-		dumpLogFile(t, filepath.Join(dirname, e.Name()))
+		dumpLogFile(tb, filepath.Join(dirname, e.Name()))
 	}
 }

-func dumpLogFile(t *testing.T, fname string) {
-	t.Helper()
+func dumpLogFile(tb testing.TB, fname string) {
+	tb.Helper()

 	data, err := os.ReadFile(fname) //nolint:gosec
 	if err != nil {
-		t.Error(err)
+		tb.Error(err)
 		return
 	}

-	t.Logf("LOG FILE: %v %v", fname, trimOutput(string(data)))
+	tb.Logf("LOG FILE: %v %v", fname, trimOutput(string(data)))
 }

 func trimOutput(s string) string {
--- a/tests/clitestutil/clitestutil.go
+++ b/tests/clitestutil/clitestutil.go
@@ -25,8 +25,8 @@ type SnapshotInfo struct {
 }

 // MustParseSnapshots parsers the output of 'snapshot list'.
-func MustParseSnapshots(t *testing.T, lines []string) []SourceInfo {
-	t.Helper()
+func MustParseSnapshots(tb testing.TB, lines []string) []SourceInfo {
+	tb.Helper()

 	var (
 		result        []SourceInfo
@@ -40,16 +40,16 @@ func MustParseSnapshots(t *testing.T, lines []string) []SourceInfo {

 		if strings.HasPrefix(l, "  ") {
 			if currentSource == nil {
-				t.Errorf("snapshot without a source: %q", l)
+				tb.Errorf("snapshot without a source: %q", l)
 				return nil
 			}

-			currentSource.Snapshots = append(currentSource.Snapshots, mustParseSnapshotInfo(t, l[2:]))
+			currentSource.Snapshots = append(currentSource.Snapshots, mustParseSnapshotInfo(tb, l[2:]))

 			continue
 		}

-		s := mustParseSourceInfo(t, l)
+		s := mustParseSourceInfo(tb, l)
 		result = append(result, s)
 		currentSource = &result[len(result)-1]
 	}
@@ -57,8 +57,8 @@ func MustParseSnapshots(t *testing.T, lines []string) []SourceInfo {
 	return result
 }

-func mustParseSnapshotInfo(t *testing.T, l string) SnapshotInfo {
-	t.Helper()
+func mustParseSnapshotInfo(tb testing.TB, l string) SnapshotInfo {
+	tb.Helper()

 	incomplete := strings.Contains(l, "incomplete")

@@ -66,7 +66,7 @@ func mustParseSnapshotInfo(t *testing.T, l string) SnapshotInfo {

 	ts, err := time.Parse("2006-01-02 15:04:05 MST", strings.Join(parts[0:3], " "))
 	if err != nil {
-		t.Fatalf("err: %v", err)
+		tb.Fatalf("err: %v", err)
 	}

 	var manifestField string
@@ -93,18 +93,17 @@ func mustParseSnapshotInfo(t *testing.T, l string) SnapshotInfo {
 	}
 }

-func mustParseSourceInfo(t *testing.T, l string) SourceInfo {
-	t.Helper()
+func mustParseSourceInfo(tb testing.TB, l string) SourceInfo {
+	tb.Helper()

 	p1 := strings.Index(l, "@")
-
 	p2 := strings.Index(l, ":")

 	if p1 >= 0 && p2 > p1 {
 		return SourceInfo{User: l[0:p1], Host: l[p1+1 : p2], Path: l[p2+1:]}
 	}

-	t.Fatalf("can't parse source info: %q", l)
+	tb.Fatalf("can't parse source info: %q", l)

 	return SourceInfo{}
 }
@@ -131,32 +130,32 @@ func mustParseDirectoryEntries(lines []string) []DirEntry {
 }

 type testEnv interface {
-	RunAndExpectSuccess(t *testing.T, args ...string) []string
+	RunAndExpectSuccess(t testing.TB, args ...string) []string
 }

 // ListSnapshotsAndExpectSuccess lists given snapshots and parses the output.
-func ListSnapshotsAndExpectSuccess(t *testing.T, e testEnv, targets ...string) []SourceInfo {
-	t.Helper()
+func ListSnapshotsAndExpectSuccess(tb testing.TB, e testEnv, targets ...string) []SourceInfo {
+	tb.Helper()

-	lines := e.RunAndExpectSuccess(t, append([]string{"snapshot", "list", "-l", "--manifest-id"}, targets...)...)
+	lines := e.RunAndExpectSuccess(tb, append([]string{"snapshot", "list", "-l", "--manifest-id"}, targets...)...)

-	return MustParseSnapshots(t, lines)
+	return MustParseSnapshots(tb, lines)
 }

 // ListDirectory lists a given directory and returns directory entries.
-func ListDirectory(t *testing.T, e testEnv, targets ...string) []DirEntry {
-	t.Helper()
+func ListDirectory(tb testing.TB, e testEnv, targets ...string) []DirEntry {
+	tb.Helper()

-	lines := e.RunAndExpectSuccess(t, append([]string{"ls", "-l"}, targets...)...)
+	lines := e.RunAndExpectSuccess(tb, append([]string{"ls", "-l"}, targets...)...)

 	return mustParseDirectoryEntries(lines)
 }

 // ListDirectoryRecursive lists a given directory recursively and returns directory entries.
-func ListDirectoryRecursive(t *testing.T, e testEnv, targets ...string) []DirEntry {
-	t.Helper()
+func ListDirectoryRecursive(tb testing.TB, e testEnv, targets ...string) []DirEntry {
+	tb.Helper()

-	lines := e.RunAndExpectSuccess(t, append([]string{"ls", "-lr"}, targets...)...)
+	lines := e.RunAndExpectSuccess(tb, append([]string{"ls", "-lr"}, targets...)...)

 	return mustParseDirectoryEntries(lines)
 }
--- a/tests/socketactivation_test/socketactivation_test.go
+++ b/tests/socketactivation_test/socketactivation_test.go
@@ -1,4 +1,4 @@
-//go:build linux
+//go:build linux || darwin

 package socketactivation_test

@@ -7,9 +7,11 @@
 	"os"
 	"strconv"
 	"strings"
+	"sync/atomic"
 	"testing"
 	"time"

+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"

 	"github.com/kopia/kopia/internal/testutil"
@@ -38,28 +40,21 @@ func TestServerControlSocketActivated(t *testing.T) {
 	l1, err := net.Listen("tcp", ":0")
 	require.NoError(t, err, "Failed to open Listener")

-	defer func() {
-		l1.Close()
-	}()
+	t.Cleanup(func() { l1.Close() })

 	port = testutil.EnsureType[*net.TCPAddr](t, l1.Addr()).Port

 	t.Logf("Activating socket on port %v", port)

+	l1File, err := testutil.EnsureType[*net.TCPListener](t, l1).File()
+	require.NoError(t, err, "failed to get filehandle for socket")
+
 	serverStarted := make(chan struct{})
 	serverStopped := make(chan struct{})

 	var sp testutil.ServerParameters

 	go func() {
-		l1File, err := testutil.EnsureType[*net.TCPListener](t, l1).File()
-		if err != nil {
-			t.Log("ERROR: Failed to get filehandle for socket")
-			close(serverStarted)
-
-			return
-		}
-
 		runner.ExtraFiles = append(runner.ExtraFiles, l1File)
 		wait, _ := env.RunAndProcessStderr(t, sp.ProcessOutput,
 			"server", "start", "--insecure", "--random-server-control-password", "--address=127.0.0.1:0")
@@ -77,15 +72,19 @@ func TestServerControlSocketActivated(t *testing.T) {
 		require.NotEmpty(t, sp.BaseURL, "Failed to start server")
 		t.Logf("server started on %v", sp.BaseURL)

-	case <-time.After(5 * time.Second):
+	case <-time.After(15 * time.Second):
 		t.Fatal("server did not start in time")
 	}

 	require.Contains(t, sp.BaseURL, ":"+strconv.Itoa(port))

-	lines := env.RunAndExpectSuccess(t, "server", "status", "--address", "http://127.0.0.1:"+strconv.Itoa(port), "--server-control-password", sp.ServerControlPassword, "--remote")
-	require.Len(t, lines, 1)
-	require.Contains(t, lines, "IDLE: another-user@another-host:"+dir0)
+	checkServerStatusFn := func(collect *assert.CollectT) {
+		lines := env.RunAndExpectSuccess(t, "server", "status", "--address", "http://127.0.0.1:"+strconv.Itoa(port), "--server-control-password", sp.ServerControlPassword, "--remote")
+		require.Len(collect, lines, 1)
+		require.Contains(collect, lines, "IDLE: another-user@another-host:"+dir0)
+	}
+
+	require.EventuallyWithT(t, checkServerStatusFn, 30*time.Second, 2*time.Second, "could not get server status, perhaps it was not listening on the control endpoint yet?")

 	env.RunAndExpectSuccess(t, "server", "shutdown", "--address", sp.BaseURL, "--server-control-password", sp.ServerControlPassword)

@@ -99,8 +98,6 @@ func TestServerControlSocketActivated(t *testing.T) {
 }

 func TestServerControlSocketActivatedTooManyFDs(t *testing.T) {
-	var port int
-
 	serverExe := os.Getenv("KOPIA_SERVER_EXE")
 	if serverExe == "" {
 		t.Skip("skipping socket-activation test")
@@ -110,59 +107,62 @@ func TestServerControlSocketActivatedTooManyFDs(t *testing.T) {
 	env := testenv.NewCLITest(t, testenv.RepoFormatNotImportant, runner)

 	env.RunAndExpectSuccess(t, "repo", "create", "filesystem", "--path", env.RepoDir, "--override-username=another-user", "--override-hostname=another-host")
-	// The KOPIA_EXE wrapper will set the LISTEN_PID variable for us
-	env.Environment["LISTEN_FDS"] = "2"

+	// create 2 file descriptor for a single socket and pass the descriptors to the server
 	l1, err := net.Listen("tcp", ":0")
 	require.NoError(t, err, "Failed to open Listener")

-	defer func() {
-		l1.Close()
-	}()
+	t.Cleanup(func() { l1.Close() })

-	port = testutil.EnsureType[*net.TCPAddr](t, l1.Addr()).Port
+	port := testutil.EnsureType[*net.TCPAddr](t, l1.Addr()).Port

-	t.Logf("Activating socket on port %v", port)
+	t.Logf("activation socket port %v", port)

-	serverStarted := make(chan []string)
+	listener := testutil.EnsureType[*net.TCPListener](t, l1)

+	l1File, err := listener.File()
+	require.NoError(t, err, "failed to get 1st filehandle for socket")
+
+	t.Cleanup(func() { l1File.Close() })
+
+	l2File, err := listener.File()
+	require.NoError(t, err, "failed to get 2nd filehandle for socket")
+
+	t.Cleanup(func() { l2File.Close() })
+
+	runner.ExtraFiles = append(runner.ExtraFiles, l1File, l2File)
+	// The KOPIA_EXE wrapper will set the LISTEN_PID variable for us
+	env.Environment["LISTEN_FDS"] = "2"
+
+	var gotExpectedErrorMessage atomic.Bool
+
+	stderrAsyncCallback := func(line string) {
+		if strings.Contains(line, "Too many activated sockets found.  Expected 1, got 2") {
+			gotExpectedErrorMessage.Store(true)
+		}
+	}
+
+	// although the server is expected to stop quickly with an error, the server's
+	// stderr is processed async to avoid test deadlocks if the server continues
+	// to run and does not exit.
+	wait, kill := env.RunAndProcessStderrAsync(t, func(string) bool { return false }, stderrAsyncCallback, "server", "start", "--insecure", "--random-server-control-password", "--address=127.0.0.1:0")
+
+	t.Cleanup(kill)
+
+	serverStopped := make(chan error)
 	go func() {
-		listener := testutil.EnsureType[*net.TCPListener](t, l1)
+		defer close(serverStopped)

-		l1File, err := listener.File()
-		if err != nil {
-			t.Log("Failed to get filehandle for socket")
-			close(serverStarted)
-
-			return
-		}
-
-		l2File, err := listener.File()
-		if err != nil {
-			t.Log("Failed to get 2nd filehandle for socket")
-			close(serverStarted)
-
-			return
-		}
-
-		runner.ExtraFiles = append(runner.ExtraFiles, l1File, l2File)
-
-		_, stderr := env.RunAndExpectFailure(t, "server", "start", "--insecure", "--random-server-control-password", "--address=127.0.0.1:0")
-
-		l1File.Close()
-		l2File.Close()
-
-		serverStarted <- stderr
-
-		close(serverStarted)
+		serverStopped <- wait()
 	}()

 	select {
-	case stderr := <-serverStarted:
-		require.Contains(t, strings.Join(stderr, ""), "Too many activated sockets found.  Expected 1, got 2")
+	case err := <-serverStopped:
+		require.Error(t, err, "server did not exit with an error")
 		t.Log("Done")
-
-	case <-time.After(5 * time.Second):
+	case <-time.After(30 * time.Second):
 		t.Fatal("server did not exit in time")
 	}
+
+	require.True(t, gotExpectedErrorMessage.Load(), "expected server's stderr to contain a line along the lines of 'Too many activated sockes ...'")
 }
--- a/tests/testenv/cli_exe_runner.go
+++ b/tests/testenv/cli_exe_runner.go
@@ -21,8 +21,8 @@ type CLIExeRunner struct {
 }

 // Start implements CLIRunner.
-func (e *CLIExeRunner) Start(t *testing.T, ctx context.Context, args []string, env map[string]string) (stdout, stderr io.Reader, wait func() error, interrupt func(os.Signal)) {
-	t.Helper()
+func (e *CLIExeRunner) Start(tb testing.TB, ctx context.Context, args []string, env map[string]string) (stdout, stderr io.Reader, wait func() error, interrupt func(os.Signal)) {
+	tb.Helper()

 	c := exec.Command(e.Exe, append([]string{
 		"--log-dir", e.LogsDir,
@@ -36,12 +36,12 @@ func (e *CLIExeRunner) Start(t *testing.T, ctx context.Context, args []string, e

 	stdoutPipe, err := c.StdoutPipe()
 	if err != nil {
-		t.Fatalf("can't set up stdout pipe reader: %v", err)
+		tb.Fatalf("can't set up stdout pipe reader: %v", err)
 	}

 	stderrPipe, err := c.StderrPipe()
 	if err != nil {
-		t.Fatalf("can't set up stderr pipe reader: %v", err)
+		tb.Fatalf("can't set up stderr pipe reader: %v", err)
 	}

 	c.Stdin = e.NextCommandStdin
@@ -49,7 +49,7 @@ func (e *CLIExeRunner) Start(t *testing.T, ctx context.Context, args []string, e
 	c.ExtraFiles = e.ExtraFiles

 	if err := c.Start(); err != nil {
-		t.Fatalf("unable to start: %v", err)
+		tb.Fatalf("unable to start: %v", err)
 	}

 	return stdoutPipe, stderrPipe, c.Wait, func(sig os.Signal) {
@@ -67,8 +67,8 @@ func (e *CLIExeRunner) Start(t *testing.T, ctx context.Context, args []string, e
 // for each. The kopia executable must be passed via KOPIA_EXE environment variable. The test
 // will be skipped if it's not provided (unless running inside an IDE in which case system-wide
 // `kopia` will be used by default).
-func NewExeRunner(t *testing.T) *CLIExeRunner {
-	t.Helper()
+func NewExeRunner(tb testing.TB) *CLIExeRunner {
+	tb.Helper()

 	exe := os.Getenv("KOPIA_EXE")
 	if exe == "" {
@@ -76,22 +76,22 @@ func NewExeRunner(t *testing.T) *CLIExeRunner {
 			// we're launched from VSCode, use system-installed kopia executable.
 			exe = "kopia"
 		} else {
-			t.Skip()
+			tb.Skip()
 		}
 	}

-	return NewExeRunnerWithBinary(t, exe)
+	return NewExeRunnerWithBinary(tb, exe)
 }

 // NewExeRunnerWithBinary returns a CLIRunner that will execute kopia commands by launching subprocesses
 // for each.
-func NewExeRunnerWithBinary(t *testing.T, exe string) *CLIExeRunner {
-	t.Helper()
+func NewExeRunnerWithBinary(tb testing.TB, exe string) *CLIExeRunner {
+	tb.Helper()

 	// unset environment variables that disrupt tests when passed to subprocesses.
 	os.Unsetenv("KOPIA_PASSWORD")

-	logsDir := testutil.TempLogDirectory(t)
+	logsDir := testutil.TempLogDirectory(tb)

 	return &CLIExeRunner{
 		Exe:     filepath.FromSlash(exe),
--- a/tests/testenv/cli_inproc_runner.go
+++ b/tests/testenv/cli_inproc_runner.go
@@ -27,8 +27,8 @@ type CLIInProcRunner struct {
 }

 // Start implements CLIRunner.
-func (e *CLIInProcRunner) Start(t *testing.T, ctx context.Context, args []string, env map[string]string) (stdout, stderr io.Reader, wait func() error, interrupt func(os.Signal)) {
-	t.Helper()
+func (e *CLIInProcRunner) Start(tb testing.TB, ctx context.Context, args []string, env map[string]string) (stdout, stderr io.Reader, wait func() error, interrupt func(os.Signal)) {
+	tb.Helper()

 	a := cli.NewApp()
 	a.DangerousCommands = "enabled"
--- a/tests/testenv/cli_test_env.go
+++ b/tests/testenv/cli_test_env.go
@@ -34,7 +34,7 @@
 // CLIRunner encapsulates running kopia subcommands for testing purposes.
 // It supports implementations that use subprocesses or in-process invocations.
 type CLIRunner interface {
-	Start(t *testing.T, ctx context.Context, args []string, env map[string]string) (stdout, stderr io.Reader, wait func() error, interrupt func(os.Signal))
+	Start(tb testing.TB, ctx context.Context, args []string, env map[string]string) (stdout, stderr io.Reader, wait func() error, interrupt func(os.Signal))
 }

 // CLITest encapsulates state for a CLI-based test.
@@ -66,9 +66,9 @@ type CLITest struct {
 var RepoFormatNotImportant []string

 // NewCLITest creates a new instance of *CLITest.
-func NewCLITest(t *testing.T, repoCreateFlags []string, runner CLIRunner) *CLITest {
-	t.Helper()
-	configDir := testutil.TempDirectory(t)
+func NewCLITest(tb testing.TB, repoCreateFlags []string, runner CLIRunner) *CLITest {
+	tb.Helper()
+	configDir := testutil.TempDirectory(tb)

 	// unset global environment variable that may interfere with the test
 	os.Unsetenv("KOPIA_METRICS_PUSH_ADDR")
@@ -99,9 +99,9 @@ func NewCLITest(t *testing.T, repoCreateFlags []string, runner CLIRunner) *CLITe
 	}

 	return &CLITest{
-		RunContext:                   testsender.CaptureMessages(testlogging.Context(t)),
+		RunContext:                   testsender.CaptureMessages(testlogging.Context(tb)),
 		startTime:                    clock.Now(),
-		RepoDir:                      testutil.TempDirectory(t),
+		RepoDir:                      testutil.TempDirectory(tb),
 		ConfigDir:                    configDir,
 		fixedArgs:                    fixedArgs,
 		DefaultRepositoryCreateFlags: formatFlags,
@@ -113,44 +113,44 @@ func NewCLITest(t *testing.T, repoCreateFlags []string, runner CLIRunner) *CLITe
 }

 // RunAndExpectSuccess runs the given command, expects it to succeed and returns its output lines.
-func (e *CLITest) RunAndExpectSuccess(t *testing.T, args ...string) []string {
-	t.Helper()
+func (e *CLITest) RunAndExpectSuccess(tb testing.TB, args ...string) []string {
+	tb.Helper()

-	stdout, _, err := e.Run(t, false, args...)
-	require.NoError(t, err, "'kopia %v' failed", strings.Join(args, " "))
+	stdout, _, err := e.Run(tb, false, args...)
+	require.NoError(tb, err, "'kopia %v' failed", strings.Join(args, " "))

 	return stdout
 }

 // TweakFile writes a xor-ed byte at a random point in a file.  Used to simulate file corruption.
-func (e *CLITest) TweakFile(t *testing.T, dirn, fglob string) {
-	t.Helper()
+func (e *CLITest) TweakFile(tb testing.TB, dirn, fglob string) {
+	tb.Helper()

 	const RwUserGroupOther = 0o666

 	// find a file within the repository to corrupt
 	mch, err := fs.Glob(os.DirFS(dirn), fglob)
-	require.NoError(t, err)
-	require.NotEmpty(t, mch)
+	require.NoError(tb, err)
+	require.NotEmpty(tb, mch)

 	// grab a random file in the directory dirn
 	fn := mch[rand.Intn(len(mch))]
 	f, err := os.OpenFile(path.Join(dirn, fn), os.O_RDWR, os.FileMode(RwUserGroupOther))
-	require.NoError(t, err)
+	require.NoError(tb, err)

 	// find the length of the file, then seek to a random location
 	l, err := f.Seek(0, io.SeekEnd)
-	require.NoError(t, err)
+	require.NoError(tb, err)

 	i := rand.Int63n(l)
 	bs := [1]byte{}

 	_, err = f.ReadAt(bs[:], i)
-	require.NoError(t, err)
+	require.NoError(tb, err)

 	// write the byte
 	_, err = f.WriteAt([]byte{^bs[0]}, i)
-	require.NoError(t, err)
+	require.NoError(tb, err)
 }

 func (e *CLITest) SetLogOutput(enable bool, prefix string) {
@@ -172,11 +172,11 @@ func (e *CLITest) getLogOutputPrefix() (string, bool) {
 	return e.logOutputPrefix, os.Getenv("KOPIA_TEST_LOG_OUTPUT") != "" || e.logOutputEnabled
 }

-// RunAndProcessStderr runs the given command, and streams its output line-by-line to a given function until it returns false.
-func (e *CLITest) RunAndProcessStderr(t *testing.T, callback func(line string) bool, args ...string) (wait func() error, kill func()) {
-	t.Helper()
+// RunAndProcessStderr runs the given command, and streams its stderr line-by-line to stderrCAllback until it returns false.
+func (e *CLITest) RunAndProcessStderr(tb testing.TB, stderrCallback func(line string) bool, args ...string) (wait func() error, kill func()) {
+	tb.Helper()

-	wait, interrupt := e.RunAndProcessStderrInt(t, callback, nil, args...)
+	wait, interrupt := e.RunAndProcessStderrInt(tb, stderrCallback, nil, args...)
 	kill = func() {
 		interrupt(os.Kill)
 	}
@@ -184,11 +184,11 @@ func (e *CLITest) RunAndProcessStderr(t *testing.T, callback func(line string) b
 	return wait, kill
 }

-// RunAndProcessStderrAsync runs the given command, and streams its output line-by-line to a given function until it returns false.
-func (e *CLITest) RunAndProcessStderrAsync(t *testing.T, callback func(line string) bool, asyncCallback func(line string), args ...string) (wait func() error, kill func()) {
-	t.Helper()
+// RunAndProcessStderrAsync runs the given command, and streams its stderr line-by-line stderrCAllback until it returns false.
+func (e *CLITest) RunAndProcessStderrAsync(tb testing.TB, stderrCallback func(line string) bool, stderrAsyncCallback func(line string), args ...string) (wait func() error, kill func()) {
+	tb.Helper()

-	wait, interrupt := e.RunAndProcessStderrInt(t, callback, asyncCallback, args...)
+	wait, interrupt := e.RunAndProcessStderrInt(tb, stderrCallback, stderrAsyncCallback, args...)
 	kill = func() {
 		interrupt(os.Kill)
 	}
@@ -196,12 +196,14 @@ func (e *CLITest) RunAndProcessStderrAsync(t *testing.T, callback func(line stri
 	return wait, kill
 }

-// RunAndProcessStderrInt runs the given command, and streams its output
-// line-by-line to outputCallback until it returns false.
-func (e *CLITest) RunAndProcessStderrInt(t *testing.T, outputCallback func(line string) bool, asyncCallback func(line string), args ...string) (wait func() error, interrupt func(os.Signal)) {
-	t.Helper()
+// RunAndProcessStderrInt runs the given command, and streams its stderr
+// line-by-line to stderrCallback until it returns false. The remaining lines
+// from stderr, if any, are asynchronously sent line-by-line to
+// stderrAsyncCallback.
+func (e *CLITest) RunAndProcessStderrInt(tb testing.TB, stderrCallback func(line string) bool, stderrAsyncCallback func(line string), args ...string) (wait func() error, interrupt func(os.Signal)) {
+	tb.Helper()

-	stdout, stderr, wait, interrupt := e.Runner.Start(t, e.RunContext, e.cmdArgs(args), e.Environment)
+	stdout, stderr, wait, interrupt := e.Runner.Start(tb, e.RunContext, e.cmdArgs(args), e.Environment)

 	prefix, logOutput := e.getLogOutputPrefix()

@@ -209,18 +211,18 @@ func (e *CLITest) RunAndProcessStderrInt(t *testing.T, outputCallback func(line
 		scanner := bufio.NewScanner(stdout)
 		for scanner.Scan() {
 			if logOutput {
-				t.Logf("[%vstdout] %v", prefix, scanner.Text())
+				tb.Logf("[%vstdout] %v", prefix, scanner.Text())
 			}
 		}

 		if logOutput {
-			t.Logf("[%vstdout] EOF", prefix)
+			tb.Logf("[%vstdout] EOF", prefix)
 		}
 	}()

 	scanner := bufio.NewScanner(stderr)
 	for scanner.Scan() {
-		if !outputCallback(scanner.Text()) {
+		if !stderrCallback(scanner.Text()) {
 			break
 		}
 	}
@@ -228,17 +230,17 @@ func (e *CLITest) RunAndProcessStderrInt(t *testing.T, outputCallback func(line
 	// complete stderr scanning in the background without processing lines.
 	go func() {
 		for scanner.Scan() {
-			if asyncCallback != nil {
-				asyncCallback(scanner.Text())
+			if stderrAsyncCallback != nil {
+				stderrAsyncCallback(scanner.Text())
 			}

 			if logOutput {
-				t.Logf("[%vstderr] %v", prefix, scanner.Text())
+				tb.Logf("[%vstderr] %v", prefix, scanner.Text())
 			}
 		}

 		if logOutput {
-			t.Logf("[%vstderr] EOF", prefix)
+			tb.Logf("[%vstderr] EOF", prefix)
 		}
 	}()

@@ -246,33 +248,33 @@ func (e *CLITest) RunAndProcessStderrInt(t *testing.T, outputCallback func(line
 }

 // RunAndExpectSuccessWithErrOut runs the given command, expects it to succeed and returns its stdout and stderr lines.
-func (e *CLITest) RunAndExpectSuccessWithErrOut(t *testing.T, args ...string) (stdout, stderr []string) {
-	t.Helper()
+func (e *CLITest) RunAndExpectSuccessWithErrOut(tb testing.TB, args ...string) (stdout, stderr []string) {
+	tb.Helper()

-	stdout, stderr, err := e.Run(t, false, args...)
-	require.NoError(t, err, "'kopia %v' failed", strings.Join(args, " "))
+	stdout, stderr, err := e.Run(tb, false, args...)
+	require.NoError(tb, err, "'kopia %v' failed", strings.Join(args, " "))

 	return stdout, stderr
 }

 // RunAndExpectFailure runs the given command, expects it to fail and returns its output lines.
-func (e *CLITest) RunAndExpectFailure(t *testing.T, args ...string) (stdout, stderr []string) {
-	t.Helper()
+func (e *CLITest) RunAndExpectFailure(tb testing.TB, args ...string) (stdout, stderr []string) {
+	tb.Helper()

 	var err error

-	stdout, stderr, err = e.Run(t, true, args...)
-	require.Error(t, err, "'kopia %v' succeeded, but expected failure", strings.Join(args, " "))
+	stdout, stderr, err = e.Run(tb, true, args...)
+	require.Error(tb, err, "'kopia %v' succeeded, but expected failure", strings.Join(args, " "))

 	return stdout, stderr
 }

 // RunAndVerifyOutputLineCount runs the given command and asserts it returns the given number of output lines, then returns them.
-func (e *CLITest) RunAndVerifyOutputLineCount(t *testing.T, wantLines int, args ...string) []string {
-	t.Helper()
+func (e *CLITest) RunAndVerifyOutputLineCount(tb testing.TB, wantLines int, args ...string) []string {
+	tb.Helper()

-	lines := e.RunAndExpectSuccess(t, args...)
-	require.Len(t, lines, wantLines, "unexpected output lines for 'kopia %v', lines:\n %s", strings.Join(args, " "), strings.Join(lines, "\n "))
+	lines := e.RunAndExpectSuccess(tb, args...)
+	require.Len(tb, lines, wantLines, "unexpected output lines for 'kopia %v', lines:\n %s", strings.Join(args, " "), strings.Join(lines, "\n "))

 	return lines
 }
@@ -290,16 +292,16 @@ func (e *CLITest) cmdArgs(args []string) []string {
 }

 // Run executes kopia with given arguments and returns the output lines.
-func (e *CLITest) Run(t *testing.T, expectedError bool, args ...string) (stdout, stderr []string, err error) {
-	t.Helper()
+func (e *CLITest) Run(tb testing.TB, expectedError bool, args ...string) (stdout, stderr []string, err error) {
+	tb.Helper()

 	args = e.cmdArgs(args)
 	outputPrefix, logOutput := e.getLogOutputPrefix()
-	t.Logf("%vrunning 'kopia %v' with %v", outputPrefix, strings.Join(args, " "), e.Environment)
+	tb.Logf("%vrunning 'kopia %v' with %v", outputPrefix, strings.Join(args, " "), e.Environment)

 	timer := timetrack.StartTimer()

-	stdoutReader, stderrReader, wait, _ := e.Runner.Start(t, e.RunContext, args, e.Environment)
+	stdoutReader, stderrReader, wait, _ := e.Runner.Start(tb, e.RunContext, args, e.Environment)

 	var wg sync.WaitGroup

@@ -311,7 +313,7 @@ func (e *CLITest) Run(t *testing.T, expectedError bool, args ...string) (stdout,
 		scanner := bufio.NewScanner(stdoutReader)
 		for scanner.Scan() {
 			if logOutput {
-				t.Logf("[%vstdout] %v", outputPrefix, scanner.Text())
+				tb.Logf("[%vstdout] %v", outputPrefix, scanner.Text())
 			}

 			stdout = append(stdout, scanner.Text())
@@ -326,7 +328,7 @@ func (e *CLITest) Run(t *testing.T, expectedError bool, args ...string) (stdout,
 		scanner := bufio.NewScanner(stderrReader)
 		for scanner.Scan() {
 			if logOutput {
-				t.Logf("[%vstderr] %v", outputPrefix, scanner.Text())
+				tb.Logf("[%vstderr] %v", outputPrefix, scanner.Text())
 			}

 			stderr = append(stderr, scanner.Text())
@@ -338,13 +340,13 @@ func (e *CLITest) Run(t *testing.T, expectedError bool, args ...string) (stdout,
 	gotErr := wait()

 	if expectedError {
-		require.Error(t, gotErr, "unexpected success when running 'kopia %v' (stdout:\n%v\nstderr:\n%v", strings.Join(args, " "), strings.Join(stdout, "\n"), strings.Join(stderr, "\n"))
+		require.Error(tb, gotErr, "unexpected success when running 'kopia %v' (stdout:\n%v\nstderr:\n%v", strings.Join(args, " "), strings.Join(stdout, "\n"), strings.Join(stderr, "\n"))
 	} else {
-		require.NoError(t, gotErr, "unexpected error when running 'kopia %v' (stdout:\n%v\nstderr:\n%v", strings.Join(args, " "), strings.Join(stdout, "\n"), strings.Join(stderr, "\n"))
+		require.NoError(tb, gotErr, "unexpected error when running 'kopia %v' (stdout:\n%v\nstderr:\n%v", strings.Join(args, " "), strings.Join(stdout, "\n"), strings.Join(stderr, "\n"))
 	}

 	//nolint:forbidigo
-	t.Logf("%vfinished in %v: 'kopia %v'", outputPrefix, timer.Elapsed().Milliseconds(), strings.Join(args, " "))
+	tb.Logf("%vfinished in %v: 'kopia %v'", outputPrefix, timer.Elapsed().Milliseconds(), strings.Join(args, " "))

 	return stdout, stderr, gotErr
 }