Improve integration test reliability and diagnostics

internal/termpty/conpty_windows_test.go

@@ -41,8 +41,7 @@ func TestStartProcessWithTerminal_StdoutFromChild(t *testing.T) {
 
 	sp := startTermchildWithPTY(t, ctx, "--print", "hello-from-child")
 
-	out, err := readUntil(ctx, sp.PTY, "hello-from-child")
-	require.NoError(t, err, "expected to see 'hello-from-child'; got: %q", out)
+	_ = requireReadUntil(t, ctx, sp, "hello-from-child")
 
 	ei := awaitExit(t, ctx, sp.ExitHandler)
 	require.NoError(t, ei.Err)
@@ -88,8 +87,7 @@ func TestStartProcessWithTerminal_StderrFromChild(t *testing.T) {
 
 	sp := startTermchildWithPTY(t, ctx, "--print-stderr", "boom-from-stderr")
 
-	out, err := readUntil(ctx, sp.PTY, "boom-from-stderr")
-	require.NoError(t, err, "expected to see 'boom-from-stderr'; got: %q", out)
+	_ = requireReadUntil(t, ctx, sp, "boom-from-stderr")
 
 	ei := awaitExit(t, ctx, sp.ExitHandler)
 	require.NoError(t, ei.Err)

internal/termpty/pty_test.go

@@ -10,6 +10,7 @@ package termpty_test
 import (
 	"bytes"
 	"context"
+	"fmt"
 	"io"
 	"os/exec"
 	"strings"
@@ -88,13 +89,17 @@ func readUntil(ctx context.Context, r io.Reader, target string) (string, error)
 	}
 
 	var accumulated bytes.Buffer
+	readAttempts := 0
+	bytesRead := 0
+	startedAt := time.Now()
 	for {
 		if strings.Contains(accumulated.String(), target) {
 			return accumulated.String(), nil
 		}
 
 		buf := make([]byte, 4096)
 		resCh := make(chan readResult, 1)
+		readAttempts++
 		go func() {
 			n, err := r.Read(buf)
 			if n > 0 {
@@ -106,21 +111,98 @@ func readUntil(ctx context.Context, r io.Reader, target string) (string, error)
 
 		select {
 		case <-ctx.Done():
-			return accumulated.String(), ctx.Err()
+			return accumulated.String(), &readUntilError{
+				target:       target,
+				readAttempts: readAttempts,
+				bytesRead:    bytesRead,
+				elapsed:      time.Since(startedAt),
+				err:          ctx.Err(),
+			}
 		case res := <-resCh:
 			if len(res.data) > 0 {
 				accumulated.Write(res.data)
+				bytesRead += len(res.data)
 			}
 			if res.err != nil {
 				if strings.Contains(accumulated.String(), target) {
 					return accumulated.String(), nil
 				}
-				return accumulated.String(), res.err
+				return accumulated.String(), &readUntilError{
+					target:       target,
+					readAttempts: readAttempts,
+					bytesRead:    bytesRead,
+					elapsed:      time.Since(startedAt),
+					err:          res.err,
+				}
 			}
 		}
 	}
 }
 
+type readUntilError struct {
+	target       string
+	readAttempts int
+	bytesRead    int
+	elapsed      time.Duration
+	err          error
+}
+
+func (e *readUntilError) Error() string {
+	return fmt.Sprintf(
+		"pty read failed before observing %q after %d attempts, %d bytes, and %s: %v",
+		e.target,
+		e.readAttempts,
+		e.bytesRead,
+		e.elapsed,
+		e.err,
+	)
+}
+
+func (e *readUntilError) Unwrap() error {
+	return e.err
+}
+
+func requireReadUntil(
+	t *testing.T,
+	ctx context.Context,
+	sp *termpty.PseudoTerminalProcess,
+	target string,
+) string {
+	t.Helper()
+
+	out, readErr := readUntil(ctx, sp.PTY, target)
+	if readErr == nil {
+		return out
+	}
+
+	exitSummary := "process exit was not observed"
+	exitTimer := time.NewTimer(drainExitTimeout)
+	defer exitTimer.Stop()
+
+	select {
+	case <-sp.ExitHandler.Exited():
+		exitInfo := sp.ExitHandler.ExitInfo()
+		exitSummary = fmt.Sprintf(
+			"process exit: PID=%d, exit code=%d, error=%v",
+			exitInfo.PID,
+			exitInfo.ExitCode,
+			exitInfo.Err,
+		)
+	case <-exitTimer.C:
+		exitSummary = fmt.Sprintf("process exit was not observed within %s", drainExitTimeout)
+	}
+
+	t.Fatalf(
+		"expected PTY output %q; child PID=%d; %s; read error=%v; output=%q",
+		target,
+		sp.Handle.Pid,
+		exitSummary,
+		readErr,
+		out,
+	)
+	return out
+}
+
 // awaitExit blocks until the exit notification arrives or ctx is done.
 func awaitExit(t *testing.T, ctx context.Context, exitHandler *process.ConcurrentProcessExitHandler) process.ProcessExitInfo {
 	t.Helper()

internal/termpty/pty_unix_test.go

@@ -39,8 +39,7 @@ func TestStartProcessWithTerminal_StdoutFromChild(t *testing.T) {
 
 	sp := startTermchildWithPTY(t, ctx, "--print", "hello-from-child")
 
-	out, err := readUntil(ctx, sp.PTY, "hello-from-child")
-	require.NoError(t, err, "expected to see 'hello-from-child'; got: %q", out)
+	_ = requireReadUntil(t, ctx, sp, "hello-from-child")
 
 	ei := awaitExit(t, ctx, sp.ExitHandler)
 	require.NoError(t, ei.Err)
@@ -56,8 +55,7 @@ func TestStartProcessWithTerminal_StderrFromChild(t *testing.T) {
 
 	sp := startTermchildWithPTY(t, ctx, "--print-stderr", "boom-from-stderr")
 
-	out, err := readUntil(ctx, sp.PTY, "boom-from-stderr")
-	require.NoError(t, err, "expected to see 'boom-from-stderr'; got: %q", out)
+	_ = requireReadUntil(t, ctx, sp, "boom-from-stderr")
 
 	ei := awaitExit(t, ctx, sp.ExitHandler)
 	require.NoError(t, ei.Err)

test/integration/container_controller_test.go

@@ -118,7 +118,7 @@ func TestPersistentContainerRecordsWorkloadID(t *testing.T) {
 	ctx, cancel := testutil.GetTestContext(t, defaultIntegrationTestTimeout)
 	defer cancel()
 
-	serverInfo, teInfo, envStartErr := StartTestEnvironmentWithOptions(ctx, ContainerController, "PersistentContainerWorkloadID", t.TempDir(), TestEnvironmentOptions{
+	serverInfo, teInfo, envStartErr := StartTestEnvironmentWithOptions(t, ctx, ContainerController, "PersistentContainerWorkloadID", t.TempDir(), TestEnvironmentOptions{
 		WorkloadID: "workload-a",
 	})
 	require.NoError(t, envStartErr)
@@ -324,26 +324,17 @@ func TestContainerInstanceStarts(t *testing.T) {
 func TestContainerRuntimeUnhealthy(t *testing.T) {
 	t.Parallel()
 	ctx, cancel := testutil.GetTestContext(t, defaultIntegrationTestTimeout)
+	defer cancel()
 
 	const testName = "container-runtime-unhealthy"
 	const imageName = testName + "-image"
 
 	// We are going to use a separate instance of the API server because we need to simulate container runtime being unhealthy,
 	// and that might interfere with other tests if we used the shared container orchestrator.
 
-	serverInfo, _, startupErr := StartTestEnvironment(ctx, ContainerController, t.Name(), NoSeparateWorkingDir)
+	serverInfo, _, startupErr := StartTestEnvironment(t, ctx, ContainerController, t.Name(), NoSeparateWorkingDir)
 	require.NoError(t, startupErr, "Failed to start the API server")
 
-	defer func() {
-		cancel()
-
-		// Wait for the API server cleanup to complete.
-		select {
-		case <-serverInfo.ApiServerDisposalComplete.Wait():
-		case <-time.After(5 * time.Second):
-		}
-	}()
-
 	ctr := apiv1.Container{
 		ObjectMeta: metav1.ObjectMeta{
 			Name:      testName,
@@ -2014,19 +2005,13 @@ func TestContainerCleanupModeRemovesExistingContainerOnDelete(t *testing.T) {
 func TestContainerCleanupModeDeletedBeforeAdoptionRemovesExistingContainer(t *testing.T) {
 	t.Parallel()
 	ctx, cancel := testutil.GetTestContext(t, defaultIntegrationTestTimeout)
+	defer cancel()
 
 	const testName = "container-cleanup-mode-delete-before-adopt"
 	const imageName = testName + "-image"
 
-	serverInfo, _, startupErr := StartTestEnvironment(ctx, ContainerController, t.Name(), NoSeparateWorkingDir)
+	serverInfo, _, startupErr := StartTestEnvironment(t, ctx, ContainerController, t.Name(), NoSeparateWorkingDir)
 	require.NoError(t, startupErr, "failed to start the API server")
-	defer func() {
-		cancel()
-		select {
-		case <-serverInfo.ApiServerDisposalComplete.Wait():
-		case <-time.After(5 * time.Second):
-		}
-	}()
 
 	ctr := apiv1.Container{
 		ObjectMeta: metav1.ObjectMeta{