fix(data list): route through the listDatasetsFn seam + cover the query path

internal/cli/data_list.go

@@ -115,7 +115,7 @@ func runDataList(ctx context.Context, a runDataListArgs) (err error) {
 	}
 	resolved, cs, release := target.Resolved, target.Clientset, target.Release
 
-	tables, err := push.ListDatasets(ctx, cs, resolved.RestConfig, resolved.Namespace)
+	tables, err := listDatasetsFn(ctx, cs, resolved.RestConfig, resolved.Namespace)
 	if err != nil {
 		return &exitError{code: 7, err: err}
 	}

internal/push/list.go

@@ -18,13 +18,18 @@ import (
 // cluster where nothing has been pushed yet — the database doesn't even
 // exist — returns an empty list rather than an error.
 func ListDatasets(ctx context.Context, cs kubernetes.Interface, cfg *rest.Config, namespace string) ([]string, error) {
-	exec := &SPDYExecutor{Config: cfg, Client: cs}
-
 	mysqlPod, mysqlContainer, err := findRunningPod(ctx, cs, namespace, "mysql")
 	if err != nil {
 		return nil, fmt.Errorf("locating mysql pod: %w", err)
 	}
+	return listDatasetsWith(ctx, &SPDYExecutor{Config: cfg, Client: cs}, namespace, mysqlPod, mysqlContainer)
+}
 
+// listDatasetsWith runs the information_schema table query in the given mysql
+// pod through exec and parses the result. It is split out from ListDatasets
+// (which locates the pod and builds the production SPDYExecutor) so the
+// exec + error-wrap + parse path is unit-testable with a fake Executor.
+func listDatasetsWith(ctx context.Context, exec Executor, namespace, pod, container string) ([]string, error) {
 	// -N drops the column header so stdout is one bare table name per
 	// line. IngestionDatabase is a compile-time constant, so the
 	// interpolation carries no injection risk.
@@ -34,7 +39,7 @@ func ListDatasets(ctx context.Context, cs kubernetes.Interface, cfg *rest.Config
 	script := fmt.Sprintf(`mysql -uroot -p"$MYSQL_ROOT_PASSWORD" -N -e "%s"`, query)
 
 	var stdout, stderr bytes.Buffer
-	if err := exec.Exec(ctx, namespace, mysqlPod, mysqlContainer,
+	if err := exec.Exec(ctx, namespace, pod, container,
 		[]string{"sh", "-c", script}, nil, &stdout, &stderr); err != nil {
 		return nil, fmt.Errorf("querying datasets: %w%s", err, stderrSuffix(&stderr))
 	}

internal/push/list_test.go

@@ -1,7 +1,10 @@
 package push
 
 import (
+	"context"
+	"errors"
 	"reflect"
+	"strings"
 	"testing"
 )
 
@@ -28,3 +31,46 @@ func TestParseDatasetList(t *testing.T) {
 		})
 	}
 }
+
+// TestListDatasetsWith covers the exec + error-wrap path of ListDatasets, which
+// was 0% because the production ListDatasets builds a real SPDYExecutor. The
+// split-out core takes an Executor, so a fake drives it: a happy query result is
+// parsed into table names, an empty database yields no datasets, and a failing
+// exec surfaces "querying datasets" plus the remote stderr (via stderrSuffix).
+func TestListDatasetsWith(t *testing.T) {
+	t.Run("query result parsed into table names", func(t *testing.T) {
+		fe := &fakeExecutor{stdoutToReturn: []byte("cats_dogs\nchurn\n")}
+		got, err := listDatasetsWith(context.Background(), fe, "tracebloc", "mysql-0", "mysql")
+		if err != nil {
+			t.Fatalf("unexpected error: %v", err)
+		}
+		if !reflect.DeepEqual(got, []string{"cats_dogs", "churn"}) {
+			t.Fatalf("got %#v, want [cats_dogs churn]", got)
+		}
+		if len(fe.gotCmd) != 3 || fe.gotCmd[0] != "sh" || !strings.Contains(fe.gotCmd[2], "information_schema") {
+			t.Errorf("unexpected exec cmd: %#v", fe.gotCmd)
+		}
+	})
+	t.Run("empty database yields no datasets", func(t *testing.T) {
+		fe := &fakeExecutor{stdoutToReturn: []byte("")}
+		got, err := listDatasetsWith(context.Background(), fe, "tracebloc", "mysql-0", "mysql")
+		if err != nil || got != nil {
+			t.Fatalf("got (%#v, %v), want (nil, nil)", got, err)
+		}
+	})
+	t.Run("exec failure surfaces the query + remote stderr", func(t *testing.T) {
+		fe := &fakeExecutor{
+			errToReturn:    errors.New("command terminated with exit code 1"),
+			stderrToReturn: []byte("ERROR 1045: Access denied"),
+		}
+		_, err := listDatasetsWith(context.Background(), fe, "tracebloc", "mysql-0", "mysql")
+		if err == nil {
+			t.Fatal("want an error when the exec fails")
+		}
+		for _, want := range []string{"querying datasets", "Access denied"} {
+			if !strings.Contains(err.Error(), want) {
+				t.Errorf("error %q must contain %q", err.Error(), want)
+			}
+		}
+	})
+}

internal/push/stream_test.go

@@ -28,6 +28,10 @@ type fakeExecutor struct {
 	// "tar: no space left on device" style remote-side failure).
 	stderrToReturn []byte
 
+	// What to write back to the caller as stdout (e.g. the mysql query
+	// result ListDatasets parses). Nil for the tar-stream callers.
+	stdoutToReturn []byte
+
 	// What to return as the exec error (simulates a 4xx/5xx from
 	// the apiserver, a network drop, etc.).
 	errToReturn error
@@ -43,7 +47,7 @@ func (f *fakeExecutor) Exec(
 	_ context.Context,
 	ns, pod, container string,
 	cmd []string,
-	stdin io.Reader, _ io.Writer, stderr io.Writer,
+	stdin io.Reader, stdout, stderr io.Writer,
 ) error {
 	f.gotNS, f.gotPod, f.gotContainer = ns, pod, container
 	f.gotCmd = cmd
@@ -52,6 +56,9 @@ func (f *fakeExecutor) Exec(
 		b, _ := io.ReadAll(stdin)
 		f.gotStdin = b
 	}
+	if len(f.stdoutToReturn) > 0 && stdout != nil {
+		_, _ = stdout.Write(f.stdoutToReturn)
+	}
 	if len(f.stderrToReturn) > 0 && stderr != nil {
 		_, _ = stderr.Write(f.stderrToReturn)
 	}