[RAPTOR-19525] fix(workload): sync upload integrity — hash-while-streaming, Phase 6 write order, zip overwrite form field

cmd/workload/del/cmd_test.go

@@ -18,11 +18,11 @@ import (
 	"bytes"
 	"os"
 	"path/filepath"
-	"runtime"
 	"testing"
 
 	"github.com/datarobot/cli/cmd/workload/internal/idargs"
 	"github.com/datarobot/cli/internal/misc/reader"
+	"github.com/datarobot/cli/internal/testutil"
 	"github.com/datarobot/cli/internal/workload/manifest"
 	"github.com/datarobot/cli/internal/workload/wapi"
 	"github.com/stretchr/testify/assert"
@@ -248,13 +248,9 @@ func TestClearStaleBinding_FindsAManifestUnderDir(t *testing.T) {
 // fails must not fail the command. It does have to say so, because the user is
 // the one who has to finish the job.
 func TestClearStaleBinding_WarnsWhenTheManifestCannotBeWritten(t *testing.T) {
-	if runtime.GOOS == "windows" {
-		t.Skip("directory permission bits do not block rename the same way on Windows")
-	}
+	testutil.SkipIfWindows(t, "directory permission bits do not block rename the same way on Windows")
 
-	if os.Geteuid() == 0 {
-		t.Skip("root ignores directory permission bits")
-	}
+	testutil.SkipIfRoot(t)
 
 	dir := t.TempDir()
 	path := writeManifest(t, dir, boundManifest)

internal/drapi/filesapi/client_test.go

@@ -364,25 +364,38 @@ func TestPollStatus_CompletedRedirect(t *testing.T) {
 	assert.True(t, IsTerminalStatus(resp.Status))
 }
 
+// TestUploadFromZipExisting proves the REPLACE request rides in the
+// multipart form body. The server's /files/<id>/fromFile/ route binds its
+// validator fields from the parsed form only and silently defaults to
+// RENAME when the overwrite field is absent — a query parameter is never
+// read, and RENAME would store re-uploaded paths as "name (2).ext" while
+// the original path keeps its stale bytes. Form fields must also precede
+// the file part: streaming parsers collect fields as they arrive, before
+// committing to a potentially huge file stream.
 func TestUploadFromZipExisting(t *testing.T) {
 	startServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		assert.Equal(t, "/api/v2/files/cid-1/fromFile/", r.URL.Path)
 		assert.Equal(t, "true", r.URL.Query().Get("useArchiveContents"))
-		assert.Equal(t, "REPLACE", r.URL.Query().Get("overwrite"))
+		assert.Empty(t, r.URL.Query().Get("overwrite"),
+			"overwrite in the URL query is silently ignored by the server")
 		assert.Contains(t, r.Header.Get("Content-Type"), "multipart/form-data")
 
-		mr, err := r.MultipartReader()
+		parts, err := readMultipartParts(r)
 		if !assert.NoError(t, err) {
 			return
 		}
 
-		part, err := mr.NextPart()
-		if !assert.NoError(t, err) {
+		if !assert.Len(t, parts, 2) {
 			return
 		}
 
-		assert.Equal(t, "file", part.FormName())
-		assert.Equal(t, "changes.zip", part.FileName())
+		assert.Equal(t, "overwrite", parts[0].Name, "form fields must precede the file part")
+		assert.Equal(t, "REPLACE", string(parts[0].Content))
+		assert.Empty(t, parts[0].FileName)
+
+		assert.Equal(t, "file", parts[1].Name)
+		assert.Equal(t, "changes.zip", parts[1].FileName)
+		assert.Equal(t, "PK\x03\x04fake-zip", string(parts[1].Content))
 
 		w.WriteHeader(http.StatusAccepted)
 		_, _ = w.Write([]byte(`{"catalogId":"cid-1","catalogVersionId":"v9","statusId":"sid-9"}`))

internal/drapi/filesapi/fromfile.go

@@ -33,7 +33,7 @@ func (c *httpClient) UploadFromZipNew(name string, size int64, body io.Reader) (
 		return nil, fmt.Errorf("build files url: %w", err)
 	}
 
-	return uploadZipMultipart(requestURL, name, size, body)
+	return uploadZipMultipart(requestURL, nil, name, size, body)
 }
 
 func (c *httpClient) UploadFromZipExisting(catalogID, name, overwrite string, size int64, body io.Reader) (*FromFileResp, error) {
@@ -43,18 +43,29 @@ func (c *httpClient) UploadFromZipExisting(catalogID, name, overwrite string, si
 
 	q := url.Values{}
 	q.Set("useArchiveContents", "true")
-	q.Set("overwrite", overwrite)
 
 	requestURL, err := drapi.EndpointURL("/files/"+url.PathEscape(catalogID)+"/fromFile/", q)
 	if err != nil {
 		return nil, fmt.Errorf("build fromFile url: %w", err)
 	}
 
-	return uploadZipMultipart(requestURL, name, size, body)
+	// overwrite must ride in the multipart form body: the server's
+	// /files/<id>/fromFile/ route binds its validator fields from the
+	// parsed form only, never from the query string, and silently defaults
+	// to RENAME when the field is absent. RENAME stores a re-uploaded path
+	// as "name (2).ext" while the original path keeps its stale bytes.
+	// useArchiveContents stays in the query: the server also ignores it
+	// there, but its declared form default is 'True' (archive extraction),
+	// so extraction happens either way and the request is unchanged
+	// apart from the overwrite fix.
+	fields := url.Values{}
+	fields.Set("overwrite", overwrite)
+
+	return uploadZipMultipart(requestURL, fields, name, size, body)
 }
 
-func uploadZipMultipart(requestURL, name string, size int64, body io.Reader) (*FromFileResp, error) {
-	req, err := newStreamingMultipartRequest(requestURL, nil, name, size, body)
+func uploadZipMultipart(requestURL string, fields url.Values, name string, size int64, body io.Reader) (*FromFileResp, error) {
+	req, err := newStreamingMultipartRequest(requestURL, nil, fields, name, size, body)
 	if err != nil {
 		return nil, err
 	}

internal/drapi/filesapi/fromfile_test.go

@@ -0,0 +1,203 @@
+// Copyright 2026 DataRobot, Inc. and its affiliates.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package filesapi
+
+import (
+	"bytes"
+	"errors"
+	"io"
+	"mime"
+	"mime/multipart"
+	"net/http"
+	"strings"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// multipartPart is one decoded part of a multipart request body, in wire
+// order. FileName is non-empty for file parts.
+type multipartPart struct {
+	Name     string
+	FileName string
+	Content  []byte
+}
+
+// decodeMultipart walks a multipart reader and returns its parts in wire
+// order. Form fields written before the file part appear before it here,
+// which is how tests pin the fields-first convention.
+func decodeMultipart(mr *multipart.Reader) ([]multipartPart, error) {
+	var parts []multipartPart
+
+	for {
+		part, err := mr.NextPart()
+		if errors.Is(err, io.EOF) {
+			return parts, nil
+		}
+
+		if err != nil {
+			return nil, err
+		}
+
+		content, err := io.ReadAll(part)
+		if err != nil {
+			return nil, err
+		}
+
+		parts = append(parts, multipartPart{
+			Name:     part.FormName(),
+			FileName: part.FileName(),
+			Content:  content,
+		})
+	}
+}
+
+// readMultipartParts parses an UNCONSUMED multipart request body with
+// mime/multipart and returns its parts in wire order. Tests assert on
+// these decoded parts rather than raw bytes so the framing stays free to
+// evolve. Reading r.Body first (e.g. for a Content-Length check) exhausts
+// the stream — parse the buffered copy via multipart.NewReader instead.
+func readMultipartParts(r *http.Request) ([]multipartPart, error) {
+	mr, err := r.MultipartReader()
+	if err != nil {
+		return nil, err
+	}
+
+	return decodeMultipart(mr)
+}
+
+// TestUploadFromZipExisting_UseArchiveContentsStaysInQuery pins the
+// placement of useArchiveContents. The monorepo fromFile validator
+// declares it as a multipart form field whose default is 'True' (archive
+// extraction), and the route binds validator fields from the parsed form
+// only — request.args is never consulted. Extraction therefore happens via
+// the server default regardless of what the CLI sends, so leaving the
+// flag in the query string (where it is ignored) changes nothing
+// observable. This test documents that decision so any future move into
+// the form is a conscious one.
+func TestUploadFromZipExisting_UseArchiveContentsStaysInQuery(t *testing.T) {
+	startServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Equal(t, "true", r.URL.Query().Get("useArchiveContents"))
+
+		parts, err := readMultipartParts(r)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		for _, part := range parts {
+			assert.NotEqual(t, "useArchiveContents", part.Name,
+				"useArchiveContents is intentionally left out of the form body")
+		}
+
+		w.WriteHeader(http.StatusAccepted)
+		_, _ = w.Write([]byte(`{"catalogId":"cid-1","catalogVersionId":"v9","statusId":"sid-9"}`))
+	}))
+
+	c := New()
+
+	body := bytes.NewReader([]byte("PK\x03\x04fake-zip"))
+	_, err := c.UploadFromZipExisting("cid-1", "changes.zip", OverwriteReplace, int64(body.Len()), body)
+	require.NoError(t, err)
+}
+
+// TestUploadFromZipNew_NoOverwriteField guards the shared-framing parity:
+// a first sync has no pre-existing paths, so it must send no overwrite
+// form field and no overwrite query parameter — the server's RENAME
+// default is correct there and the request should stay minimal.
+func TestUploadFromZipNew_NoOverwriteField(t *testing.T) {
+	startServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		assert.Empty(t, r.URL.Query().Get("overwrite"))
+
+		parts, err := readMultipartParts(r)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		for _, part := range parts {
+			assert.NotEqual(t, "overwrite", part.Name)
+		}
+
+		if !assert.Len(t, parts, 1) {
+			return
+		}
+
+		assert.Equal(t, "file", parts[0].Name)
+		assert.Equal(t, "wapi-sync.zip", parts[0].FileName)
+
+		w.WriteHeader(http.StatusAccepted)
+		_, _ = w.Write([]byte(`{"catalogId":"new-cid","catalogVersionId":"new-ver","statusId":"sid-new"}`))
+	}))
+
+	c := New()
+
+	body := bytes.NewReader([]byte("PK\x03\x04fake-zip"))
+	resp, err := c.UploadFromZipNew("wapi-sync.zip", int64(body.Len()), body)
+	require.NoError(t, err)
+	assert.Equal(t, "new-ver", resp.CatalogVersionID)
+}
+
+// TestUploadFromZipExisting_ContentLengthMatchesBodyWithFormFields
+// verifies the Content-Length accounting survives folding form fields
+// into the prologue: the advertised length must equal the received body
+// exactly, and the streamed file must still decode intact. An off-by-N
+// here surfaces as a transport error or a truncated multipart stream,
+// never as a silent pass.
+func TestUploadFromZipExisting_ContentLengthMatchesBodyWithFormFields(t *testing.T) {
+	// Long enough to cross io.Copy's internal buffer more than once.
+	payload := strings.Repeat("0123456789abcdef", 4096)
+
+	startServer(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		raw, err := io.ReadAll(r.Body)
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		assert.Equal(t, r.ContentLength, int64(len(raw)),
+			"advertised Content-Length must match the received body byte-for-byte")
+
+		// The body is consumed by the ReadAll above, so parse the buffered
+		// copy rather than asking the request for a fresh MultipartReader.
+		_, params, err := mime.ParseMediaType(r.Header.Get("Content-Type"))
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		parts, err := decodeMultipart(multipart.NewReader(bytes.NewReader(raw), params["boundary"]))
+		if !assert.NoError(t, err) {
+			return
+		}
+
+		if !assert.Len(t, parts, 2) {
+			return
+		}
+
+		assert.Equal(t, "REPLACE", string(parts[0].Content))
+		assert.Equal(t, payload, string(parts[1].Content))
+
+		w.WriteHeader(http.StatusAccepted)
+		_, _ = w.Write([]byte(`{"catalogId":"cid-1","catalogVersionId":"v9","statusId":"sid-9"}`))
+	}))
+
+	c := New()
+
+	body := bytes.NewReader([]byte(payload))
+	_, err := c.UploadFromZipExisting("cid-1", "changes.zip", OverwriteReplace, int64(body.Len()), body)
+	require.NoError(t, err)
+}
+
+// The package's mime/multipart types stay referenced even if helper
+// implementations drift; mirrors the guard at the bottom of client_test.go.
+var _ = multipart.ErrMessageTooLarge