test(data): seam the money path — table-test the ingest outcome matrix (#1009 P0)

internal/cli/coverage_test.go

@@ -102,8 +102,15 @@ func TestClassifyPushOutcome(t *testing.T) {
 		{"clean", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 10}}}, nil, "succeeded", 0},
 		{"partial", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 10, InsertedRecords: 7, FailedRecords: 3}}}, nil, "completed_with_failures", 9},
 		{"failed", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeFailed}}, nil, "failed", 9},
+		{"unknown", &submit.Result{Submit: resp, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeUnknown}}, nil, "unknown", 9},
 		{"detached", &submit.Result{Submit: resp}, nil, "detached", 0},
+		{"nil result", nil, nil, "detached", 0},
 		{"watch error", &submit.Result{Submit: resp}, &submit.WatchError{Err: errors.New("stream broke")}, "watch_error", 9},
+		// The submit-side error buckets (exit 5 vs 8) the original matrix missed.
+		{"auth 401", nil, &submit.SubmitError{StatusCode: 401}, "auth_error", 5},
+		{"auth 403", nil, &submit.SubmitError{StatusCode: 403}, "auth_error", 5},
+		{"submit 500", nil, &submit.SubmitError{StatusCode: 500}, "submit_error", 8},
+		{"submit 422", nil, &submit.SubmitError{StatusCode: 422}, "submit_error", 8},
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {

internal/cli/data.go

@@ -758,6 +758,37 @@ other collaborators train against it without ever seeing the raw files.`))
 		return &exitError{code: 7, err: stageErr}
 	}
 
+	// 10–12. The ingestion-run tail: mint token → port-forward → submit →
+	//         classify → emit JSON → reclaim staging. Extracted so its
+	//         outcome matrix (exit 5/8/9, JSON emission, and the
+	//         must-NOT-reclaim-on-partial gate) is table-testable via the
+	//         injected seams without a cluster (#1009). jsonEmitted flows
+	//         back so the --output-json error defer above stays correct.
+	je, runErr := runIngestionRun(ctx, out, a, target, specBytes, spec)
+	jsonEmitted = je
+	return runErr
+}
+
+// runIngestionRun is the money path's outcome tail. It mints the ingestor
+// token, port-forwards to jobs-manager, POSTs the run, classifies the result
+// into a status + process exit code (kept in lockstep by classifyPushOutcome),
+// emits the machine-readable JSON in --output-json mode, and reclaims the
+// staged source copy on a clean success only.
+//
+// Split out of runDataIngest purely for testability: the four cluster-touching
+// steps go through package-level seams (mintIngestorTokenFn /
+// portForwardJobsManagerFn / submitRunFn / cleanStagingFn), so a table test can
+// drive the full classify → exit-code → JSON → reclaim matrix — including the
+// "must NOT reclaim on partial failure" gate — without standing up a cluster
+// (#1009).
+//
+// Returns jsonEmitted so runDataIngest's --output-json error defer knows
+// whether a result object already reached stdout: the mint / port-forward
+// failures return before the emit and rely on that defer; the submit path
+// always emits.
+func runIngestionRun(ctx context.Context, out io.Writer, a runDataIngestArgs, target *clusterTarget, specBytes []byte, spec map[string]any) (jsonEmitted bool, err error) {
+	resolved, cs, release, pvc := target.Resolved, target.Clientset, target.Release, target.PVC
+
 	// 10. Mint the SA token Phase 4 uses to authenticate the POST
 	//     to jobs-manager. Expiry is 1 hour (vs cluster info's 10
 	//     min) because the full Phase 4 lifecycle — submit + watch
@@ -770,10 +801,10 @@ other collaborators train against it without ever seeing the raw files.`))
 		a.Printer.Hintf("Submitting the run, then following along as tracebloc validates your data and loads it into the table — progress streams below.")
 		a.Printer.Hintf("This follows the run for up to an hour; a longer run keeps going on its own (or start it with --detach and check back later).")
 	}
-	tok, err := cluster.MintIngestorToken(ctx, cs, resolved.Namespace,
+	tok, err := mintIngestorTokenFn(ctx, cs, resolved.Namespace,
 		release.IngestorSAName, 3600, nil)
 	if err != nil {
-		return &exitError{code: 5, err: err}
+		return false, &exitError{code: 5, err: err}
 	}
 
 	// 11. Open a port-forward to a Pod backing the jobs-manager
@@ -784,10 +815,10 @@ other collaborators train against it without ever seeing the raw files.`))
 	//     `kubectl port-forward`. Bugbot PR #10 r3 caught the
 	//     original broken-by-design direct-URL POST.
 	a.Printer.Infof("Connecting to your workspace to submit the run…")
-	pf, err := submit.PortForwardJobsManager(ctx, cs, resolved.RestConfig,
+	pf, err := portForwardJobsManagerFn(ctx, cs, resolved.RestConfig,
 		resolved.Namespace, release.JobsManagerServiceName, release.JobsManagerPort)
 	if err != nil {
-		return &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)}
+		return false, &exitError{code: 8, err: fmt.Errorf("setting up jobs-manager port-forward: %w", err)}
 	}
 	defer pf.Close()
 
@@ -810,7 +841,7 @@ other collaborators train against it without ever seeing the raw files.`))
 	//        WatchResult Detached        → 0 (cluster keeps running)
 	//        WatchResult Succeeded clean → 0
 	localEndpoint := fmt.Sprintf("http://localhost:%d", pf.LocalPort)
-	submitRes, err := submit.Run(ctx, submit.Options{
+	submitRes, err := submitRunFn(ctx, submit.Options{
 		Submitter:        submit.NewHTTPSubmitter(localEndpoint, tok.Token),
 		Client:           cs,
 		IngestConfigYAML: string(specBytes),
@@ -845,21 +876,13 @@ other collaborators train against it without ever seeing the raw files.`))
 		jsonEmitted = true
 	}
 
-	// Reclaim the staged source copy on a CLEAN success only. The
-	// ingestor copies (not moves) the staged files into the table, so
-	// leaving .tracebloc-staging/<table> behind doubles PVC use for
-	// file-bearing datasets until the next --overwrite or `data delete`
-	// (the staging-leak found by the ingest UX audit; cli#166 / epic #67).
-	// Gated on status=="succeeded" so we never touch the source on a
-	//   - detached run (status "detached"): the Job is still reading it;
-	//   - partial (status "completed_with_failures") or failed run: the
-	//     user may want the source to inspect/retry.
-	// Best-effort and time-bounded (push.StagingCleanupTimeout): a failed
-	// or slow reclaim must not fail — or noticeably delay — a successful
-	// ingest.
-	if status == "succeeded" {
+	// Reclaim the staged source copy on a CLEAN success only (see
+	// shouldReclaimStaging). Best-effort and time-bounded
+	// (push.StagingCleanupTimeout): a failed or slow reclaim must not
+	// fail — or noticeably delay — a successful ingest.
+	if shouldReclaimStaging(status) {
 		a.Printer.Infof("Reclaiming the temporary staging copy on the cluster…")
-		if cerr := push.CleanStaging(ctx, cs,
+		if cerr := cleanStagingFn(ctx, cs,
 			&push.SPDYExecutor{Config: resolved.RestConfig, Client: cs},
 			resolved.Namespace, a.Spec.Table, push.PodSpecOptions{
 				Namespace:          resolved.Namespace,
@@ -875,9 +898,25 @@ other collaborators train against it without ever seeing the raw files.`))
 	}
 
 	if exitErr != nil {
-		return exitErr
+		return jsonEmitted, exitErr
 	}
-	return nil
+	return jsonEmitted, nil
+}
+
+// shouldReclaimStaging reports whether the staged source copy should be
+// reclaimed after the run. ONLY on a clean success: the ingestor copies (not
+// moves) the staged files into the table, so leaving .tracebloc-staging/<table>
+// behind doubles PVC use for file-bearing datasets until the next --overwrite
+// or `data delete` (the staging-leak found by the ingest UX audit; cli#166 /
+// epic #67). Everything else keeps the source:
+//   - a detached run ("detached") — the Job is still reading it;
+//   - a partial ("completed_with_failures") or failed/errored run — the user
+//     may want the source to inspect or retry.
+//
+// This is the "must NOT reclaim on partial failure" gate (#1009), named so the
+// invariant is table-testable in isolation.
+func shouldReclaimStaging(status string) bool {
+	return status == "succeeded"
 }
 
 // classifyPushOutcome maps the result of submit.Run to a machine-
@@ -1072,6 +1111,18 @@ func writePushErrorJSON(w io.Writer, sp push.SpecArgs, e error, code int) {
 // listDatasetsFn is a test seam over push.ListDatasets.
 var listDatasetsFn = push.ListDatasets
 
+// Test seams over the cluster-touching steps of runIngestionRun (#1009).
+// Production wires them to the real functions; a table test overrides them to
+// drive the classify → exit-code → JSON → reclaim matrix without a cluster
+// (mirrors the listDatasetsFn seam). cleanStagingFn is here too so a test can
+// observe whether the staging reclaim ran (the must-NOT-reclaim gate).
+var (
+	mintIngestorTokenFn      = cluster.MintIngestorToken
+	portForwardJobsManagerFn = submit.PortForwardJobsManager
+	submitRunFn              = submit.Run
+	cleanStagingFn           = push.CleanStaging
+)
+
 // destTableExists reports whether the destination table already holds an
 // ingested dataset, via the same query `data list` uses. It fails OPEN: a
 // broken check returns (false, note) so the ingest proceeds — the in-cluster

internal/cli/ingestion_run_test.go

@@ -0,0 +1,166 @@
+package cli
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"testing"
+
+	"github.com/tracebloc/cli/internal/cluster"
+	"github.com/tracebloc/cli/internal/push"
+	"github.com/tracebloc/cli/internal/submit"
+	"github.com/tracebloc/cli/internal/ui"
+	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
+)
+
+// The money path (#1009): submit → classify → exit-code → JSON → reclaim.
+// These tests pin the outcome matrix — including the "must NOT reclaim on
+// partial failure" gate — without standing up a cluster, via the seams
+// (mintIngestorTokenFn / portForwardJobsManagerFn / submitRunFn /
+// cleanStagingFn) that runIngestionRun goes through.
+
+func succeededResult() *submit.Result {
+	return &submit.Result{
+		Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "ingestor-x"},
+		Watch:  &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 2, InsertedRecords: 2}},
+	}
+}
+
+func partialResult() *submit.Result {
+	return &submit.Result{
+		Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "ingestor-x"},
+		Watch:  &submit.WatchResult{Outcome: submit.JobOutcomeSucceeded, Summary: &submit.Summary{TotalRecords: 2, InsertedRecords: 1, FailedRecords: 1}},
+	}
+}
+
+// TestShouldReclaimStaging pins the must-NOT-reclaim-on-partial gate: the
+// staged source is reclaimed ONLY on a clean success.
+func TestShouldReclaimStaging(t *testing.T) {
+	if !shouldReclaimStaging("succeeded") {
+		t.Error(`shouldReclaimStaging("succeeded") = false, want true`)
+	}
+	for _, st := range []string{
+		"completed_with_failures", "failed", "unknown", "detached",
+		"auth_error", "submit_error", "watch_error", "dry-run", "error", "",
+	} {
+		if shouldReclaimStaging(st) {
+			t.Errorf("shouldReclaimStaging(%q) = true, want false — a non-clean run must keep the source", st)
+		}
+	}
+}
+
+// TestRunIngestionRun_Matrix drives the whole outcome tail through the seams:
+// per row it asserts the exit code, whether the staging reclaim ran, and the
+// emitted --output-json status — all in lockstep.
+func TestRunIngestionRun_Matrix(t *testing.T) {
+	origMint, origPF, origRun, origClean := mintIngestorTokenFn, portForwardJobsManagerFn, submitRunFn, cleanStagingFn
+	defer func() {
+		mintIngestorTokenFn, portForwardJobsManagerFn, submitRunFn, cleanStagingFn = origMint, origPF, origRun, origClean
+	}()
+
+	target := &clusterTarget{
+		Resolved:  &cluster.ResolvedConfig{Namespace: "tracebloc"},
+		Clientset: nil, // the seams ignore it; the reclaim SPDYExecutor literal doesn't deref
+		Release:   &cluster.ParentRelease{IngestorSAName: "ingestor", JobsManagerServiceName: "jm", JobsManagerPort: 8080},
+		PVC:       &cluster.SharedPVC{ClaimName: "pvc", MountPath: "/data/shared"},
+	}
+	spec := map[string]any{"table": "t", "category": "image_classification", "intent": "train", "label": "label"}
+
+	cases := []struct {
+		name        string
+		mintErr     error
+		pfErr       error
+		submitRes   *submit.Result
+		submitErr   error
+		wantCode    int // 0 == success (nil err)
+		wantStatus  string
+		wantReclaim bool
+		wantJSON    bool
+	}{
+		{"succeeded", nil, nil, succeededResult(), nil, 0, "succeeded", true, true},
+		{"partial", nil, nil, partialResult(), nil, 9, "completed_with_failures", false, true},
+		{"failed", nil, nil, &submit.Result{Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "j"}, Watch: &submit.WatchResult{Outcome: submit.JobOutcomeFailed}}, nil, 9, "failed", false, true},
+		{"detached", nil, nil, &submit.Result{Submit: &submit.SubmitResponse{Namespace: "tracebloc", JobName: "j"}, Watch: nil}, nil, 0, "detached", false, true},
+		{"submit-auth", nil, nil, nil, &submit.SubmitError{StatusCode: 401}, 5, "auth_error", false, true},
+		{"submit-5xx", nil, nil, nil, &submit.SubmitError{StatusCode: 500}, 8, "submit_error", false, true},
+		{"watch-err", nil, nil, nil, &submit.WatchError{Err: errors.New("x")}, 9, "watch_error", false, true},
+		// mint / port-forward failures return BEFORE the JSON emit and the
+		// reclaim; jsonEmitted is false (runDataIngest's error defer covers it).
+		{"mint-fail", errors.New("mint boom"), nil, nil, nil, 5, "", false, false},
+		{"pf-fail", nil, errors.New("pf boom"), nil, nil, 8, "", false, false},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			reclaimCalled := false
+			mintIngestorTokenFn = func(_ context.Context, _ kubernetes.Interface, _, _ string, _ int64, _ []string) (*cluster.IngestorToken, error) {
+				if c.mintErr != nil {
+					return nil, c.mintErr
+				}
+				return &cluster.IngestorToken{Token: "tok"}, nil
+			}
+			portForwardJobsManagerFn = func(_ context.Context, _ kubernetes.Interface, _ *rest.Config, _, _ string, _ int) (*submit.ForwardedConnection, error) {
+				if c.pfErr != nil {
+					return nil, c.pfErr
+				}
+				return &submit.ForwardedConnection{LocalPort: 12345}, nil
+			}
+			submitRunFn = func(_ context.Context, _ submit.Options) (*submit.Result, error) {
+				return c.submitRes, c.submitErr
+			}
+			cleanStagingFn = func(_ context.Context, _ kubernetes.Interface, _ push.Executor, _, _ string, _ push.PodSpecOptions) error {
+				reclaimCalled = true
+				return nil
+			}
+
+			var jsonBuf bytes.Buffer
+			a := runDataIngestArgs{
+				Spec:       push.SpecArgs{Table: "t"},
+				Printer:    ui.New(io.Discard, ui.WithColor(false)),
+				OutputJSON: true,
+				JSONOut:    &jsonBuf,
+			}
+			je, err := runIngestionRun(context.Background(), io.Discard, a, target, []byte("yaml"), spec)
+
+			code := 0
+			if err != nil {
+				var ee *exitError
+				if !errors.As(err, &ee) {
+					t.Fatalf("err is not *exitError: %v", err)
+				}
+				code = ee.Code()
+			}
+			if code != c.wantCode {
+				t.Errorf("exit code = %d, want %d", code, c.wantCode)
+			}
+			if reclaimCalled != c.wantReclaim {
+				t.Errorf("reclaim called = %v, want %v (only a clean success reclaims)", reclaimCalled, c.wantReclaim)
+			}
+			if je != c.wantJSON {
+				t.Errorf("jsonEmitted = %v, want %v", je, c.wantJSON)
+			}
+			if c.wantJSON {
+				var got pushJSONResult
+				if err := json.Unmarshal(jsonBuf.Bytes(), &got); err != nil {
+					t.Fatalf("emitted JSON invalid: %v (%q)", err, jsonBuf.String())
+				}
+				if got.Status != c.wantStatus {
+					t.Errorf("emitted JSON status = %q, want %q", got.Status, c.wantStatus)
+				}
+			} else if jsonBuf.Len() != 0 {
+				t.Errorf("expected no JSON on the pre-submit failure path, got %q", jsonBuf.String())
+			}
+		})
+	}
+}
+
+// TestSeamsWiredToRealFns guards that the indirection didn't accidentally
+// leave a seam nil (a nil seam would panic the money path in production).
+func TestSeamsWiredToRealFns(t *testing.T) {
+	if mintIngestorTokenFn == nil || portForwardJobsManagerFn == nil ||
+		submitRunFn == nil || cleanStagingFn == nil {
+		t.Fatal("a money-path seam is nil — production would panic")
+	}
+}

internal/submit/portforward.go

@@ -30,8 +30,13 @@ type ForwardedConnection struct {
 	done   chan struct{}
 }
 
-// Close tears down the port-forward. Safe to call multiple times.
+// Close tears down the port-forward. Safe to call multiple times, and
+// safe on a zero-value connection that was never started (stopCh nil) —
+// e.g. a test fake handed back by an injected PortForwardJobsManager.
 func (f *ForwardedConnection) Close() {
+	if f.stopCh == nil {
+		return // never started; nothing to tear down
+	}
 	select {
 	case <-f.stopCh:
 		return // already closed