Speed up sync integration tests

internal/helpers.go

@@ -5,10 +5,12 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"io"
 	"math/rand"
 	"os"
 	"path/filepath"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -54,25 +56,69 @@ type cobraTestRunner struct {
 	stdout bytes.Buffer
 	stderr bytes.Buffer
 
+	// Line-by-line output.
+	// Background goroutines populate these channels by reading from stdout/stderr pipes.
+	stdoutLines <-chan string
+	stderrLines <-chan string
+
 	errch <-chan error
 }
 
+func consumeLines(ctx context.Context, wg *sync.WaitGroup, r io.Reader) <-chan string {
+	ch := make(chan string, 1000)
+	wg.Add(1)
+	go func() {
+		defer close(ch)
+		defer wg.Done()
+		scanner := bufio.NewScanner(r)
+		for scanner.Scan() {
+			select {
+			case <-ctx.Done():
+				return
+			case ch <- scanner.Text():
+			}
+		}
+	}()
+	return ch
+}
+
 func (t *cobraTestRunner) RunBackground() {
+	var stdoutR, stderrR io.Reader
+	var stdoutW, stderrW io.WriteCloser
+	stdoutR, stdoutW = io.Pipe()
+	stderrR, stderrW = io.Pipe()
 	root := root.RootCmd
-	root.SetOut(&t.stdout)
-	root.SetErr(&t.stderr)
+	root.SetOut(stdoutW)
+	root.SetErr(stderrW)
 	root.SetArgs(t.args)
 
 	errch := make(chan error)
 	ctx, cancel := context.WithCancel(context.Background())
 
+	// Tee stdout/stderr to buffers.
+	stdoutR = io.TeeReader(stdoutR, &t.stdout)
+	stderrR = io.TeeReader(stderrR, &t.stderr)
+
+	// Consume stdout/stderr line-by-line.
+	var wg sync.WaitGroup
+	t.stdoutLines = consumeLines(ctx, &wg, stdoutR)
+	t.stderrLines = consumeLines(ctx, &wg, stderrR)
+
 	// Run command in background.
 	go func() {
 		cmd, err := root.ExecuteContextC(ctx)
 		if err != nil {
 			t.Logf("Error running command: %s", err)
 		}
 
+		// Close pipes to signal EOF.
+		stdoutW.Close()
+		stderrW.Close()
+
+		// Wait for the [consumeLines] routines to finish now that
+		// the pipes they're reading from have closed.
+		wg.Wait()
+
 		if t.stdout.Len() > 0 {
 			// Make a copy of the buffer such that it remains "unread".
 			scanner := bufio.NewScanner(bytes.NewBuffer(t.stdout.Bytes()))
@@ -127,6 +173,9 @@ func (c *cobraTestRunner) Eventually(condition func() bool, waitFor time.Duratio
 	ticker := time.NewTicker(tick)
 	defer ticker.Stop()
 
+	// Kick off condition check immediately.
+	go func() { ch <- condition() }()
+
 	for tick := ticker.C; ; {
 		select {
 		case err := <-c.errch: