Revanth14 commented on code in PR #1324:
URL: https://github.com/apache/iceberg-go/pull/1324#discussion_r3573227079
##########
catalog/rest/scan_planning_test.go:
##########
@@ -94,3 +186,385 @@ func TestPlanTableScanResponseRejectsUnknownStatus(t
*testing.T) {
err := json.Unmarshal([]byte(`{"status":"bogus"}`), &resp)
require.ErrorIs(t, err, ErrRESTError)
}
+
+func TestFetchPlanningResultResponseRejectsMalformedResponses(t *testing.T) {
+ t.Parallel()
+
+ t.Run("failed without error", func(t *testing.T) {
+ t.Parallel()
+
+ var resp FetchPlanningResultResponse
+ err := json.Unmarshal([]byte(`{"status":"failed"}`), &resp)
+ require.ErrorIs(t, err, ErrRESTError)
+ })
+
+ t.Run("unknown status", func(t *testing.T) {
+ t.Parallel()
+
+ var resp FetchPlanningResultResponse
+ err := json.Unmarshal([]byte(`{"status":"bogus"}`), &resp)
+ require.ErrorIs(t, err, ErrRESTError)
+ })
+}
+
+func TestPlanTableScanGeneratesIdempotencyKeyAndUsesDefaultAccessDelegation(t
*testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t, []endpoint{endpointPlanTableScan},
func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w
http.ResponseWriter, req *http.Request) {
+ got := req.Header.Get(headerIdempotencyKey)
+ require.NotEmpty(t, got)
+ parsed, err := uuid.Parse(got)
+ require.NoError(t, err)
+ // The spec pins the generated key to UUIDv7.
+ assert.Equal(t, 7, int(parsed.Version()))
+ assert.Equal(t, []string{defaultAccessDelegation},
req.Header.Values(headerIcebergAccessDelegation))
+
+ _, err =
w.Write([]byte(`{"status":"completed","plan-id":"plan-1"}`))
+ require.NoError(t, err)
+ })
+ })
+
+ _, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{})
+ require.NoError(t, err)
+}
+
+func TestPlanTableScanRejectsInvalidIdempotencyKey(t *testing.T) {
+ t.Parallel()
+
+ for _, tc := range []struct {
+ name string
+ key string
+ }{
+ {"not a uuid", "not-a-uuid"},
+ // Valid UUID, but v1 not v7 — the spec pins the header to
UUIDv7.
+ {"valid uuid wrong version",
"11111111-1111-1111-1111-111111111111"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ key := tc.key
+ cat := &Catalog{endpoints:
newEndpointSet([]endpoint{endpointPlanTableScan})}
+
+ _, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{
+ IdempotencyKey: &key,
+ })
+ require.ErrorIs(t, err, iceberg.ErrInvalidArgument)
+ })
+ }
+}
+
+func TestPlanTableScanRequest(t *testing.T) {
+ t.Parallel()
+
+ cases := []struct {
+ name string
+ response string
+ wantStatus PlanStatus
+ wantPlanID *string
+ wantError string
+ }{
+ {
+ name: "completed",
+ response:
`{"status":"completed","plan-id":"plan-1","plan-tasks":["task-1"]}`,
+ wantStatus: PlanStatusCompleted,
+ wantPlanID: stringPtr("plan-1"),
+ },
+ {
+ name: "submitted",
+ response: `{"status":"submitted","plan-id":"plan-2"}`,
+ wantStatus: PlanStatusSubmitted,
+ wantPlanID: stringPtr("plan-2"),
+ },
+ {
+ name: "failed",
+ response:
`{"status":"failed","error":{"message":"boom","type":"ServerError","code":500}}`,
+ wantStatus: PlanStatusFailed,
+ wantError: "boom",
+ },
+ }
+
+ for _, tc := range cases {
+ t.Run(tc.name, func(t *testing.T) {
+ t.Parallel()
+
+ idempotencyKey := "0190b6c5-1c3d-7000-8000-000000000001"
+ accessDelegation := "remote-signing"
+ snapshotID := int64(22)
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointPlanTableScan}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan", func(w http.ResponseWriter,
req *http.Request) {
+ require.Equal(t, http.MethodPost,
req.Method)
+ assert.Equal(t, idempotencyKey,
req.Header.Get(headerIdempotencyKey))
+ assert.Equal(t,
[]string{accessDelegation}, req.Header.Values(headerIcebergAccessDelegation))
+
+ body, err := io.ReadAll(req.Body)
+ require.NoError(t, err)
+ assert.JSONEq(t, `{
+ "snapshot-id": 22,
+ "select": ["id", "data"],
+ "filter": {"type":
"always-true"}
+ }`, string(body))
+
+ _, err = w.Write([]byte(tc.response))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.PlanTableScan(context.Background(),
table.Identifier{"db", "tbl"}, PlanTableScanRequest{
+ IdempotencyKey: &idempotencyKey,
+ AccessDelegation: &accessDelegation,
+ SnapshotID: &snapshotID,
+ Select: []string{"id", "data"},
+ Filter:
json.RawMessage(`{"type":"always-true"}`),
+ })
+ if tc.wantError != "" {
+ // A failed plan returns a *PlanFailedError and
a zero resp; the
+ // detail rides on the error.
+ require.ErrorIs(t, err, ErrPlanFailed)
+ var pfe *PlanFailedError
+ require.ErrorAs(t, err, &pfe)
+ require.NotNil(t, pfe.Detail)
+ assert.Equal(t, tc.wantError,
pfe.Detail.Message)
+ assert.Equal(t, PlanTableScanResponse{}, resp)
+
+ return
+ }
+
+ require.NoError(t, err)
+ assert.Equal(t, tc.wantStatus, resp.Status)
+ if tc.wantPlanID != nil {
+ require.NotNil(t, resp.PlanID)
+ assert.Equal(t, *tc.wantPlanID, *resp.PlanID)
+ }
+ })
+ }
+}
+
+func TestFetchPlanningResultRequest(t *testing.T) {
+ t.Parallel()
+
+ accessDelegation := "remote-signing"
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123",
func(w http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodGet, req.Method)
+ assert.Equal(t, []string{accessDelegation},
req.Header.Values(headerIcebergAccessDelegation))
+
+ _, err :=
w.Write([]byte(`{"status":"completed","plan-tasks":["task-1"]}`))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{
+ AccessDelegation: &accessDelegation,
+ })
+ require.NoError(t, err)
+ assert.Equal(t, PlanStatusCompleted, resp.Status)
+ assert.Equal(t, []string{"task-1"}, resp.PlanTasks)
+}
+
+func TestFetchPlanningResultUsesDefaultAccessDelegation(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+ mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123",
func(w http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodGet, req.Method)
+ assert.Equal(t, []string{defaultAccessDelegation},
req.Header.Values(headerIcebergAccessDelegation))
+
+ _, err := w.Write([]byte(`{"status":"submitted"}`))
+ require.NoError(t, err)
+ })
+ })
+
+ resp, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{})
+ require.NoError(t, err)
+ assert.Equal(t, PlanStatusSubmitted, resp.Status)
+}
+
+func TestFetchPlanningResultMapsNotFoundToPlanExpired(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/expired-plan", func(w
http.ResponseWriter, req *http.Request) {
+ require.Equal(t, http.MethodGet, req.Method)
+ w.WriteHeader(http.StatusNotFound)
+ })
+ })
+
+ _, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "expired-plan", FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanExpired)
+}
+
+func TestFetchPlanningResultResponseAcceptsCancelled(t *testing.T) {
+ t.Parallel()
+
+ // cancelled is a valid poll result (unlike PlanTableScanResponse, which
+ // rejects it). Pin it so a refactor routing it into the default error
case
+ // is caught.
+ var resp FetchPlanningResultResponse
+ require.NoError(t, json.Unmarshal([]byte(`{"status":"cancelled"}`),
&resp))
+ assert.Equal(t, PlanStatusCancelled, resp.Status)
+}
+
+func TestFetchPlanningResultStatusArms(t *testing.T) {
+ t.Parallel()
+
+ t.Run("cancelled returns ErrPlanCancelled", func(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123", func(w
http.ResponseWriter, req *http.Request) {
+ _, err :=
w.Write([]byte(`{"status":"cancelled"}`))
+ require.NoError(t, err)
+ })
+ })
+
+ _, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanCancelled)
+ })
+
+ t.Run("failed returns PlanFailedError", func(t *testing.T) {
+ t.Parallel()
+
+ cat := newScanPlanningTestCatalog(t,
[]endpoint{endpointFetchPlanResult}, func(mux *http.ServeMux) {
+
mux.HandleFunc("/v1/namespaces/db/tables/tbl/plan/plan-123", func(w
http.ResponseWriter, req *http.Request) {
+ _, err :=
w.Write([]byte(`{"status":"failed","error":{"message":"boom","type":"ServerError","code":500}}`))
+ require.NoError(t, err)
+ })
+ })
+
+ _, err := cat.FetchPlanningResult(context.Background(),
table.Identifier{"db", "tbl"}, "plan-123", FetchPlanningResultOptions{})
+ require.ErrorIs(t, err, ErrPlanFailed)
+ var pfe *PlanFailedError
+ require.ErrorAs(t, err, &pfe)
+ assert.Equal(t, "boom", pfe.Detail.Message)
Review Comment:
Addressed. Added `require.NotNil(t, pfe.Detail)` before dereferencing
`pfe.Detail.Message`, matching the sibling test and ensuring a missing detail
fails cleanly instead of panicking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
