From 53b3c58285025a84e1268a87e65bf998f62483ea Mon Sep 17 00:00:00 2001 From: Bhukya Date: Tue, 10 Nov 2026 03:45:50 +0000 Subject: [PATCH] Read the test events of a pipeline update readTestEvents reads every page of an update's events and keeps the test fields that the Go SDK event type does not have. It sends the SDK's ListPipelineEventsRequest and uses the filter of buildPipelineEventFilter. fetchAllPipelineEvents, which pipelines logs and run use, is not changed. testRun collects one result per test case and reports whether the results are complete. The command does not use them yet. Co-authored-by: Isaac --- cmd/pipelines/test_events.go | 317 ++++++++++++++++++++++++++++++ cmd/pipelines/test_events_test.go | 254 ++++++++++++++++++++++++ 2 files changed, 571 insertions(+) create mode 100644 cmd/pipelines/test_events.go create mode 100644 cmd/pipelines/test_events_test.go diff --git a/cmd/pipelines/test_events.go b/cmd/pipelines/test_events.go new file mode 100644 index 00000000000..e9dd88aff70 --- /dev/null +++ b/cmd/pipelines/test_events.go @@ -0,0 +1,317 @@ +package pipelines + +import ( + "context" + "fmt" + "net/http" + "strings" + "time" + + "github.com/databricks/cli/libs/auth" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/client" + "github.com/databricks/databricks-sdk-go/service/pipelines" +) + +const ( + testCaseProgressEventType = "test_case_progress" + testRunSummaryEventType = "test_run_summary" +) + +type testCaseStatus string + +const ( + testCasePending testCaseStatus = "PENDING" + testCaseRunning testCaseStatus = "RUNNING" + testCasePassed testCaseStatus = "PASSED" + testCaseFailed testCaseStatus = "FAILED" + testCaseSkipped testCaseStatus = "SKIPPED" + testCaseError testCaseStatus = "ERROR" +) + +// parseTestCaseStatus accepts both the short status and the proto enum value name. +func parseTestCaseStatus(s string) testCaseStatus { + return testCaseStatus(strings.TrimPrefix(s, "PIPELINE_TEST_CASE_STATUS_")) +} + +func (s testCaseStatus) isFinal() bool { + switch s { + case testCasePassed, testCaseFailed, testCaseSkipped, testCaseError: + return true + default: + return false + } +} + +type testRunSummary struct { + Total int `json:"total"` + Passed int `json:"passed"` + Failed int `json:"failed"` + Skipped int `json:"skipped"` + Errored int `json:"errored"` +} + +// testPipelineEvent is an entry of List Pipeline Events. The Go SDK PipelineEvent has no +// details or origin.test_case_id, so test events are decoded from the raw response. +type testPipelineEvent struct { + Id string `json:"id"` + EventType string `json:"event_type"` + Timestamp string `json:"timestamp"` + Origin struct { + TestCaseId string `json:"test_case_id"` + } `json:"origin"` + Error *pipelines.ErrorDetail `json:"error"` + Truncation *pipelines.Truncation `json:"truncation"` + Details struct { + TestCaseProgress *struct { + Status string `json:"status"` + } `json:"test_case_progress"` + TestRunSummary *testRunSummary `json:"test_run_summary"` + } `json:"details"` +} + +func (e testPipelineEvent) isTruncated(field string) bool { + if e.Truncation == nil { + return false + } + for _, f := range e.Truncation.TruncatedFields { + if f.FieldName == field { + return true + } + } + return false +} + +// testEventsPageSize is the largest page List Pipeline Events returns. +const testEventsPageSize = 250 + +// readTestEvents returns the events of one update in timestamp order, following every page. With +// sinceTimestamp it returns only the events at or after that timestamp, which must be a timestamp +// string exactly as the API returned it. +func readTestEvents(ctx context.Context, w *databricks.WorkspaceClient, pipelineID, updateID, sinceTimestamp string) ([]testPipelineEvent, error) { + apiClient, err := client.New(w.Config) + if err != nil { + return nil, fmt.Errorf("failed to create API client: %w", err) + } + path := fmt.Sprintf("/api/2.0/pipelines/%s/events", pipelineID) + request := pipelines.ListPipelineEventsRequest{ + Filter: buildPipelineEventFilter(updateID, nil, nil, sinceTimestamp, ""), + OrderBy: []string{"timestamp asc"}, + MaxResults: testEventsPageSize, + } + var events []testPipelineEvent + for { + var page struct { + Events []testPipelineEvent `json:"events"` + NextPageToken string `json:"next_page_token"` + } + err := apiClient.Do(ctx, http.MethodGet, path, auth.WorkspaceIDHeaders(w.Config), nil, request, &page) + if err != nil { + return nil, fmt.Errorf("failed to fetch pipeline events: %w", err) + } + events = append(events, page.Events...) + if page.NextPageToken == "" || len(page.Events) == 0 { + return events, nil + } + // A request with page_token must not repeat filter or order_by: the token carries them, + // and the service rejects the request otherwise. + request = pipelines.ListPipelineEventsRequest{PageToken: page.NextPageToken, MaxResults: testEventsPageSize} + } +} + +// testCaseResult is what the events report about one test case. +type testCaseResult struct { + // id is origin.test_case_id: the absolute workspace path of the test file followed by the pytest + // test ID, such as /Workspace/Users/someone@example.com/orders/tests/test_orders.py::test_backfill. + id string + status testCaseStatus + // finals counts final outcomes; a complete result has exactly one per case. + finals int + started time.Time + finished time.Time + errorDetail *pipelines.ErrorDetail + detailsTruncated bool +} + +// duration returns the time between the RUNNING and the final event, if both are known. +func (c *testCaseResult) duration() (time.Duration, bool) { + if c.started.IsZero() || c.finished.IsZero() || c.finished.Before(c.started) { + return 0, false + } + return c.finished.Sub(c.started), true +} + +// testRun accumulates the events of a test update, ignoring events it has already seen. +type testRun struct { + seen map[string]bool + cases []*testCaseResult + byID map[string]*testCaseResult + summaries []testRunSummary + // unreadable counts test events whose outcome is missing. + unreadable int + // fatalError is the first fatal error that the update reported, other than the test verdict. + fatalError string + // verdictError is the service's verdict that tests failed, if the update reported it. + verdictError string +} + +// testVerdictErrorPrefix starts the message of the error by which the service reports that tests +// failed, such as "[PIPELINE_TEST_FAILED] 1 of the pipeline's tests failed (1 passed, 0 skipped).". +// The message is matched because ErrorDetail has no error class yet; only the output depends on it. +const testVerdictErrorPrefix = "[PIPELINE_TEST_FAILED]" + +// isTestVerdict reports whether an update error is the service's verdict that tests failed. +func isTestVerdict(e *pipelines.ErrorDetail) bool { + for _, exception := range e.Exceptions { + if strings.HasPrefix(exception.Message, testVerdictErrorPrefix) { + return true + } + } + return false +} + +func newTestRun() *testRun { + return &testRun{seen: map[string]bool{}, byID: map[string]*testCaseResult{}} +} + +// add records an event and returns the case it gave a final outcome to, if any. +func (r *testRun) add(e testPipelineEvent) *testCaseResult { + if r.seen[e.Id] { + return nil + } + r.seen[e.Id] = true + + if e.Error != nil && e.Error.Fatal { + switch { + case isTestVerdict(e.Error): + r.verdictError = errorDetailText(e.Error) + case r.fatalError == "": + r.fatalError = errorDetailText(e.Error) + } + } + + switch e.EventType { + case testRunSummaryEventType: + if e.Details.TestRunSummary == nil { + r.unreadable++ + return nil + } + r.summaries = append(r.summaries, *e.Details.TestRunSummary) + case testCaseProgressEventType: + progress := e.Details.TestCaseProgress + if progress == nil || e.Origin.TestCaseId == "" { + r.unreadable++ + return nil + } + status := parseTestCaseStatus(progress.Status) + if !status.isFinal() && status != testCasePending && status != testCaseRunning { + r.unreadable++ + return nil + } + + c, ok := r.byID[e.Origin.TestCaseId] + if !ok { + c = &testCaseResult{id: e.Origin.TestCaseId, status: testCasePending} + r.byID[c.id] = c + r.cases = append(r.cases, c) + } + timestamp, _ := time.Parse(time.RFC3339Nano, e.Timestamp) + switch { + case status == testCaseRunning: + c.started = timestamp + if !c.status.isFinal() { + c.status = testCaseRunning + } + case status.isFinal(): + c.finals++ + c.status = status + c.finished = timestamp + c.errorDetail = e.Error + c.detailsTruncated = e.isTruncated("error") + if c.finals == 1 { + return c + } + } + } + return nil +} + +// counts returns the number of cases with each final outcome. +func (r *testRun) counts() testRunSummary { + var s testRunSummary + for _, c := range r.cases { + if !c.status.isFinal() { + continue + } + s.Total++ + switch c.status { + case testCasePassed: + s.Passed++ + case testCaseFailed: + s.Failed++ + case testCaseSkipped: + s.Skipped++ + case testCaseError: + s.Errored++ + default: + } + } + return s +} + +// incompleteReason explains why the events of an ended update are not a complete result, +// or returns "" if they are: one summary and one final outcome per case, matching its counts. +func (r *testRun) incompleteReason(state pipelines.UpdateInfoState) string { + if r.unreadable > 0 { + return "some test results could not be read" + } + switch len(r.summaries) { + case 0: + switch state { + case pipelines.UpdateInfoStateFailed: + return "the update failed before all test results were received" + case pipelines.UpdateInfoStateCanceled: + return "the update was canceled before all test results were received" + default: + return "the update ended without a test summary" + } + case 1: + default: + return fmt.Sprintf("received %d test summaries instead of one", len(r.summaries)) + } + + for _, c := range r.cases { + if c.finals > 1 { + return fmt.Sprintf("test %s has more than one result", c.id) + } + } + summary := r.summaries[0] + received := r.counts() + if received.Total != summary.Total { + return fmt.Sprintf("received %d of %d test results", received.Total, summary.Total) + } + if received != summary { + return "the test results do not match the test summary" + } + for _, c := range r.cases { + if c.finals == 0 { + return fmt.Sprintf("test %s has no result", c.id) + } + } + return "" +} + +func errorDetailText(detail *pipelines.ErrorDetail) string { + var parts []string + for _, e := range detail.Exceptions { + switch { + case e.ClassName != "" && e.Message != "": + parts = append(parts, e.ClassName+": "+e.Message) + case e.Message != "": + parts = append(parts, e.Message) + case e.ClassName != "": + parts = append(parts, e.ClassName) + } + } + return strings.Join(parts, "\n") +} diff --git a/cmd/pipelines/test_events_test.go b/cmd/pipelines/test_events_test.go new file mode 100644 index 00000000000..fb0288b26c8 --- /dev/null +++ b/cmd/pipelines/test_events_test.go @@ -0,0 +1,254 @@ +package pipelines + +import ( + "encoding/json" + "fmt" + "net/url" + "strconv" + "strings" + "testing" + "time" + + "github.com/databricks/cli/libs/testserver" + databricks "github.com/databricks/databricks-sdk-go" + "github.com/databricks/databricks-sdk-go/service/pipelines" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// testEvents builds the events of one test, each with its own event ID. +type testEvents struct { + t *testing.T + n int +} + +func newTestEvents(t *testing.T) *testEvents { + return &testEvents{t: t} +} + +func (b *testEvents) testCase(id, status, timestamp string, extra ...string) testPipelineEvent { + b.n++ + return decodeTestEvent(b.t, fmt.Sprintf(`{ + "id": "e%d", "event_type": "test_case_progress", "timestamp": %q, + "origin": {"test_case_id": %q}, + "details": {"test_case_progress": {"status": %q}}%s + }`, b.n, timestamp, id, status, strings.Join(append([]string{""}, extra...), ","))) +} + +func (b *testEvents) summary(total, passed, failed, skipped, errored int) testPipelineEvent { + b.n++ + return decodeTestEvent(b.t, fmt.Sprintf(`{ + "id": "e%d", "event_type": "test_run_summary", + "details": {"test_run_summary": {"total": %d, "passed": %d, "failed": %d, "skipped": %d, "errored": %d}} + }`, b.n, total, passed, failed, skipped, errored)) +} + +func decodeTestEvent(t *testing.T, data string) testPipelineEvent { + var e testPipelineEvent + require.NoError(t, json.Unmarshal([]byte(data), &e)) + return e +} + +func TestTestRunAdd(t *testing.T) { + ev := newTestEvents(t) + run := newTestRun() + events := []testPipelineEvent{ + ev.testCase("tests/test_a.py::test_ok", "PENDING", "2026-01-01T00:00:00.000Z"), + ev.testCase("tests/test_a.py::test_ok", "RUNNING", "2026-01-01T00:00:01.000Z"), + ev.testCase("tests/test_a.py::test_ok", "PASSED", "2026-01-01T00:00:02.840Z"), + ev.testCase("tests/test_a.py::test_bad", "PIPELINE_TEST_CASE_STATUS_FAILED", "2026-01-01T00:00:03.000Z", + `"error": {"exceptions": [{"class_name": "AssertionError", "message": "assert 1 == 2"}]}`, + `"truncation": {"truncated_fields": [{"field_name": "error"}]}`), + // A late RUNNING event does not undo the final outcome. + ev.testCase("tests/test_a.py::test_bad", "RUNNING", "2026-01-01T00:00:02.000Z"), + ev.summary(2, 1, 1, 0, 0), + } + + var finished []string + for _, e := range events { + if c := run.add(e); c != nil { + finished = append(finished, c.id) + } + } + // Reading every page again adds nothing. + for _, e := range events { + assert.Nil(t, run.add(e)) + } + + assert.Equal(t, []string{"tests/test_a.py::test_ok", "tests/test_a.py::test_bad"}, finished) + require.Len(t, run.cases, 2) + + ok := run.cases[0] + assert.Equal(t, testCasePassed, ok.status) + d, known := ok.duration() + assert.True(t, known) + assert.Equal(t, 1840*time.Millisecond, d) + + bad := run.cases[1] + assert.Equal(t, testCaseFailed, bad.status) + assert.True(t, bad.detailsTruncated) + assert.Equal(t, "AssertionError: assert 1 == 2", errorDetailText(bad.errorDetail)) + d, known = bad.duration() + assert.True(t, known) + assert.Equal(t, time.Second, d) + + assert.Empty(t, run.incompleteReason(pipelines.UpdateInfoStateFailed)) +} + +func TestTestRunIncompleteReason(t *testing.T) { + ev := newTestEvents(t) + passed := func() testPipelineEvent { + return ev.testCase("t::a", "PASSED", "") + } + tests := []struct { + name string + events []testPipelineEvent + state pipelines.UpdateInfoState + want string + }{ + { + name: "no summary after completion", + events: []testPipelineEvent{passed()}, + state: pipelines.UpdateInfoStateCompleted, + want: "the update ended without a test summary", + }, + { + name: "two summaries", + events: []testPipelineEvent{passed(), ev.summary(1, 1, 0, 0, 0), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "received 2 test summaries instead of one", + }, + { + name: "missing outcome", + events: []testPipelineEvent{passed(), ev.testCase("t::b", "RUNNING", ""), ev.summary(2, 2, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "received 1 of 2 test results", + }, + { + name: "case without outcome that the summary leaves out", + events: []testPipelineEvent{passed(), ev.testCase("t::b", "RUNNING", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "test t::b has no result", + }, + { + name: "two outcomes for one case", + events: []testPipelineEvent{passed(), ev.testCase("t::a", "FAILED", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "test t::a has more than one result", + }, + { + name: "outcomes differ from summary", + events: []testPipelineEvent{ev.testCase("t::a", "FAILED", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "the test results do not match the test summary", + }, + { + name: "outcome without details", + events: []testPipelineEvent{ + passed(), + decodeTestEvent(t, `{"id": "truncated", "event_type": "test_case_progress", "origin": {"test_case_id": "t::b"}}`), + ev.summary(1, 1, 0, 0, 0), + }, + state: pipelines.UpdateInfoStateCompleted, + want: "some test results could not be read", + }, + { + name: "unknown status", + events: []testPipelineEvent{passed(), ev.testCase("t::b", "PIPELINE_TEST_CASE_STATUS_UNSPECIFIED", ""), ev.summary(1, 1, 0, 0, 0)}, + state: pipelines.UpdateInfoStateCompleted, + want: "some test results could not be read", + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + run := newTestRun() + for _, e := range tc.events { + run.add(e) + } + assert.Equal(t, tc.want, run.incompleteReason(tc.state)) + }) + } +} + +func TestReadTestEventsFollowsPages(t *testing.T) { + server := testserver.New(t) + var queries []url.Values + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any { + queries = append(queries, req.URL.Query()) + if req.URL.Query().Get("page_token") == "" { + return map[string]any{"events": []map[string]string{{"id": "e1"}}, "next_page_token": "page2"} + } + return map[string]any{"events": []map[string]string{{"id": "e2"}}} + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + events, err := readTestEvents(t.Context(), w, "p1", "u1", "2026-01-01T00:00:01.000Z") + require.NoError(t, err) + + var ids []string + for _, e := range events { + ids = append(ids, e.Id) + } + assert.Equal(t, []string{"e1", "e2"}, ids) + assert.Equal(t, []url.Values{ + { + "filter": {"update_id = 'u1' AND timestamp >= '2026-01-01T00:00:01.000Z'"}, + "order_by": {"timestamp asc"}, + "max_results": {"250"}, + }, + { + "page_token": {"page2"}, + "max_results": {"250"}, + }, + }, queries) +} + +// NFR3: a run of 200 cases spans several pages, and each case is counted once. +func TestReadTestEvents200Cases(t *testing.T) { + const cases = 200 + var all []map[string]any + for i := range cases { + id := fmt.Sprintf("/Workspace/p/tests/test_many.py::test_case_%03d", i) + for j, status := range []string{"PENDING", "RUNNING", "PASSED"} { + all = append(all, map[string]any{ + "id": fmt.Sprintf("c%d-%d", i, j), "event_type": "test_case_progress", "timestamp": "2026-01-01T00:00:00.000Z", + "origin": map[string]any{"test_case_id": id}, "details": map[string]any{"test_case_progress": map[string]any{"status": status}}, + }) + } + } + all = append(all, map[string]any{ + "id": "summary", "event_type": "test_run_summary", + "details": map[string]any{"test_run_summary": map[string]any{"total": cases, "passed": cases}}, + }) + + server := testserver.New(t) + pages := 0 + server.Handle("GET", "/api/2.0/pipelines/{pipeline_id}/events", func(req testserver.Request) any { + pages++ + offset := 0 + if token := req.URL.Query().Get("page_token"); token != "" { + _, err := fmt.Sscan(token, &offset) + require.NoError(t, err) + } + end := min(offset+testEventsPageSize, len(all)) + page := map[string]any{"events": all[offset:end]} + if end < len(all) { + page["next_page_token"] = strconv.Itoa(end) + } + return page + }) + + w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: server.URL, Token: "token"}) + require.NoError(t, err) + events, err := readTestEvents(t.Context(), w, "p1", "u1", "") + require.NoError(t, err) + assert.Equal(t, 3, pages) + + run := newTestRun() + for _, e := range append(events, events...) { + run.add(e) + } + assert.Empty(t, run.incompleteReason(pipelines.UpdateInfoStateCompleted)) + assert.Equal(t, testRunSummary{Total: cases, Passed: cases}, run.counts()) +}