package file

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io/fs"
	"math"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"testing/fstest"
	"time"

	files_sdk "github.com/Files-com/files-sdk-go/v3"
	"github.com/Files-com/files-sdk-go/v3/directory"
	uploadManager "github.com/Files-com/files-sdk-go/v3/file/manager"
	"github.com/Files-com/files-sdk-go/v3/file/status"
	"github.com/Files-com/files-sdk-go/v3/ignore"
	"github.com/Files-com/files-sdk-go/v3/lib"
	"github.com/gin-gonic/gin"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

type MockUploader struct {
	files_sdk.File
	findError   files_sdk.ResponseError
	uploadError error
}

func (m *MockUploader) UploadWithResume(...UploadOption) (UploadResumable, error) {
	return UploadResumable{}, m.uploadError
}

type recordingUploader struct {
	MockUploader
	mu       sync.Mutex
	uploadIO uploadIO
}

func (r *recordingUploader) UploadWithResume(opts ...UploadOption) (UploadResumable, error) {
	io := uploadIO{
		Manager:         uploadManager.Default().FilePartsManager,
		passedInContext: context.Background(),
	}
	for _, opt := range opts {
		var err error
		io, err = opt(io)
		if err != nil {
			return UploadResumable{}, err
		}
	}
	r.mu.Lock()
	r.uploadIO = io
	r.mu.Unlock()
	return UploadResumable{}, r.uploadError
}

func (r *recordingUploader) recordedUploadIO() uploadIO {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.uploadIO
}

func (m *MockUploader) Find(files_sdk.FileFindParams, ...files_sdk.RequestResponseOption) (files_sdk.File, error) {
	return m.File, m.findError
}

func uploadWithS3PartError(t *testing.T, code string, status int, readerOpt UploadOption) (UploadResumable, error, int) {
	t.Helper()

	server := (&MockAPIServer{T: t}).Do()
	t.Cleanup(server.Shutdown)

	attempts := 0
	server.MockRoute("/upload/file.bak", func(c *gin.Context, _ interface{}) bool {
		attempts++
		c.Data(
			status,
			"application/xml",
			[]byte(fmt.Sprintf(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>%s</Code><Message>provider message</Message></Error>`, code)),
		)
		return true
	})

	client := server.Client()
	u, err := client.UploadWithResume(
		UploadWithSize(1024),
		readerOpt,
		UploadWithDestinationPath("file.bak"),
		UploadWithManager(lib.NewConstrainedWorkGroup(1)),
	)

	return u, err, attempts
}

func TestUploadRetryLogError(t *testing.T) {
	rawMessage := "Your socket connection to the server was not read from or written to within the timeout period.\nIdle connections will be closed."
	logError := uploadRetryLogError(lib.S3Error{Code: "RequestTimeout", Message: rawMessage})

	require.Equal(t, "Upload to backend storage timed out.", logError)
	require.NotContains(t, logError, "RequestTimeout")
	require.NotContains(t, logError, "Your socket connection")
	require.NotContains(t, logError, "Idle connections")

	logError = uploadRetryLogError(errors.New("first line\nsecond line"))
	require.Equal(t, "first line second line", logError)
}

func (m *MockUploader) CreateFolder(files_sdk.FolderCreateParams, ...files_sdk.RequestResponseOption) (files_sdk.File, error) {
	return files_sdk.File{}, nil
}

func TestUploaderAdaptiveConcurrencyUseSDKDefaultCapsKeepsJobPartManagerForFallback(t *testing.T) {
	cases := []struct {
		name                       string
		params                     UploaderParams
		wantUploadV2               bool
		wantSDKDefaultV2Caps       bool
		wantSharedAdaptiveProvider bool
	}{
		{
			name: "explicit manager without adaptive concurrency stays v1",
			params: UploaderParams{
				Manager: uploadManager.Build(7, 1),
			},
			wantUploadV2: false,
		},
		{
			name: "nil manager uses sdk default caps",
			params: UploaderParams{
				AdaptiveConcurrency: true,
			},
			wantUploadV2:               true,
			wantSDKDefaultV2Caps:       true,
			wantSharedAdaptiveProvider: true,
		},
		{
			name: "manager used for scheduling only",
			params: UploaderParams{
				AdaptiveConcurrency:                  true,
				AdaptiveConcurrencyUseSDKDefaultCaps: true,
				Manager:                              uploadManager.Build(100, 1),
			},
			wantUploadV2:               true,
			wantSDKDefaultV2Caps:       true,
			wantSharedAdaptiveProvider: true,
		},
		{
			name: "explicit manager cap uses isolated adaptive manager",
			params: UploaderParams{
				AdaptiveConcurrency: true,
				Manager:             uploadManager.Build(7, 1),
			},
			wantUploadV2:         true,
			wantSDKDefaultV2Caps: false,
		},
	}

	for _, tt := range cases {
		t.Run(tt.name, func(t *testing.T) {
			root := t.TempDir()
			path := filepath.Join(root, "upload.bin")
			require.NoError(t, os.WriteFile(path, []byte("payload"), 0600))
			job := (&Job{
				Params: tt.params,
				Config: files_sdk.Config{}.Init(),
				Logger: lib.NullLogger{},
			}).Init()
			job.SetManager(tt.params.Manager)
			job.Ignore, _ = ignore.New()
			job.Type = directory.File
			uploader := &recordingUploader{}
			uploadStatus := &UploadStatus{
				Mutex:      &sync.RWMutex{},
				Uploader:   uploader,
				job:        job,
				localPath:  path,
				remotePath: "upload.bin",
				status:     status.Indexed,
				file: files_sdk.File{
					Path: "upload.bin",
					Size: 7,
				},
			}
			job.Add(uploadStatus)
			onComplete := make(chan *UploadStatus, 1)

			ctx := context.Background()
			require.True(t, job.fileAdmissionManager().WaitWithContext(ctx))
			enqueueUpload(ctx, job, uploadStatus, onComplete)
			<-onComplete

			recorded := uploader.recordedUploadIO()
			assert.Equal(t, tt.wantUploadV2, recorded.uploadV2)
			assert.True(t, recorded.managerSet)
			assert.Equal(t, tt.wantSDKDefaultV2Caps, recorded.uploadV2UseSDKDefaultCaps)
			if tt.wantSharedAdaptiveProvider {
				assert.NotNil(t, recorded.uploadV2ManagerProvider)
			} else {
				assert.Nil(t, recorded.uploadV2ManagerProvider)
			}
			maxer, ok := recorded.Manager.(interface{ Max() int })
			require.True(t, ok)
			assert.Equal(t, job.FilePartsManager.Max(), maxer.Max())
		})
	}
}

func Test_excludeFile(t *testing.T) {
	var progressReportError error

	init := func() (*UploadStatus, fstest.MapFS, *Job) {
		job := (&Job{Logger: files_sdk.Config{}.Init().Logger}).Init()
		job.Ignore, _ = ignore.New()
		job.Params = UploaderParams{}
		uploadStatus := &UploadStatus{job: job, Mutex: &sync.RWMutex{}, file: files_sdk.File{Path: "test"}, remotePath: "test"}
		uploadStatus.Job().Add(uploadStatus)
		uploader := &MockUploader{}
		uploadStatus.Uploader = uploader
		uploadStatus.job.EventsReporter = CreateReporter(func(s JobFile) {
			progressReportError = s.Err
		})

		mockFs := make(fstest.MapFS)
		job.RemoteFs = mockFs
		return uploadStatus, mockFs, job
	}
	assert := assert.New(t)
	uploadStatus, mockFs, job := init()

	t.Run("sync not enabled and sizes do match", func(t *testing.T) {
		mockFs["test"] = &fstest.MapFile{
			Sys: files_sdk.File{Size: 10},
		}
		uploadStatus.file.Size = 10
		uploadStatus.Sync = false
		assert.False(excludeFile(uploadStatus, false))
	})

	t.Run("when sizes don't match", func(t *testing.T) {
		uploadStatus.Sync = true
		mockFs["test"] = &fstest.MapFile{
			Sys: files_sdk.File{Size: 9},
		}
		uploadStatus.file.Size = 10
		assert.False(excludeFile(uploadStatus, false))
		assert.Equal(nil, progressReportError)
	})

	t.Run("when sizes do match", func(t *testing.T) {
		mockFs["test"] = &fstest.MapFile{
			Sys: files_sdk.File{Size: 10},
		}
		uploadStatus.file.Size = 10
		assert.True(excludeFile(uploadStatus, false))
	})

	t.Run("There is no server version", func(t *testing.T) {
		delete(mockFs, "test")
		assert.False(excludeFile(uploadStatus, false))
	})

	t.Run("when sizes do match on a deeply nested path", func(t *testing.T) {
		oldUploadStatus := *uploadStatus
		uploadStatus = &UploadStatus{job: job, Mutex: &sync.RWMutex{}, file: files_sdk.File{Path: "test/path/test"}, remotePath: "test/path/test"}
		uploadStatus.Sync = true
		mockFs["test/path/test"] = &fstest.MapFile{
			Sys: files_sdk.File{Size: 10},
		}
		mockFs["test/path"] = &fstest.MapFile{
			Sys:  files_sdk.File{Size: 10},
			Mode: fs.ModeDir,
		}
		uploadStatus.file.Size = 10
		assert.True(excludeFile(uploadStatus, false))
		uploadStatus = &oldUploadStatus
	})

	t.Run("when transfer is a single file", func(t *testing.T) {
		uploadStatus.job.Type = directory.File
		uploadStatus.Sync = true
		mockFs["test"] = &fstest.MapFile{
			Sys: files_sdk.File{Size: 10},
		}
		uploadStatus.file.Size = 10
		assert.True(excludeFile(uploadStatus, false))
		uploadStatus.job.Type = directory.Dir
		uploadStatus.Sync = false
	})

	t.Run("case insensitive", func(t *testing.T) {
		uploadStatus.job.Type = directory.Dir
		uploadStatus.Sync = true
		mockFs["Test"] = &fstest.MapFile{
			Sys: files_sdk.File{Size: 10},
		}
		uploadStatus.file.Size = 10
		assert.True(excludeFile(uploadStatus, false))
		uploadStatus.Sync = false
	})

	t.Run("Ignore files", func(t *testing.T) {
		job.Ignore, _ = ignore.New([]string{"*.css"}...)
		uploadStatus.localPath = "main.css"
		assert.True(excludeFile(uploadStatus, false))

		uploadStatus.localPath = "main.php"
		assert.False(excludeFile(uploadStatus, false))

		job.Ignore, _ = ignore.New([]string{"*.css", "*.php"}...)
		uploadStatus.localPath = "main.css"
		assert.True(excludeFile(uploadStatus, false))
	})

	t.Run("FeatureFlag incremental-updates", func(t *testing.T) {
		type Args struct {
			destinationMtime *time.Time
			sourceMtime      time.Time
			sync             bool
		}
		t.Run("when sizes do not match", func(t *testing.T) {
			test := func(t *testing.T, args Args) bool {
				uploadStatus, mockFs, _ := init()
				mockFs["test"] = &fstest.MapFile{
					Sys: files_sdk.File{Size: 11, Mtime: args.destinationMtime},
				}
				uploadStatus.Sync = true
				uploadStatus.file.Size = 10
				uploadStatus.file.Mtime = &args.sourceMtime
				skip := excludeFile(uploadStatus, true)
				if args.destinationMtime != nil {
					diff := args.destinationMtime.Sub(args.sourceMtime)
					diff = time.Duration(math.Abs(float64(diff))).Truncate(time.Millisecond)
					t.Logf("%v - destinationMtime: %v, sourceMtime: %v, diff: %v, sync: %v", t.Name(), args.destinationMtime.Format(time.RFC3339), args.sourceMtime.Format(time.RFC3339), diff, !skip)
				}
				assert.Equal(args.sync, !skip, "should it sync")
				return skip
			}

			t.Run("when destination Mtime is older than source time it should sync", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: lib.Time(time.Now().Add(-time.Hour * 48)),
						sourceMtime:      time.Now().Add(-time.Hour * 24),
						sync:             true,
					},
				)
			})

			t.Run("when destination Mtime is the within the same minute as source Mtime it should not sync", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: lib.Time(
							time.Date(2021, 8, 15, 14, 30, 45, 100, time.Local),
						),
						sourceMtime: time.Date(2021, 8, 15, 14, 30, 00, 100, time.Local),
						sync:        false,
					},
				)
			})

			t.Run("when destination Mtime is the just outside the same minute as source Mtime it should not sync", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: lib.Time(
							time.Date(2021, 8, 15, 14, 30, 45, 100, time.Local),
						),
						sourceMtime: time.Date(2021, 8, 15, 14, 31, 45, 100, time.Local),
						sync:        true,
					},
				)
			})

			t.Run("when destination Mtime is the same source Mtime it should not sync", func(t *testing.T) {
				mtime := time.Now().Add(-time.Hour * 24)
				test(
					t,
					Args{
						destinationMtime: &mtime,
						sourceMtime:      mtime,
						sync:             false,
					},
				)
			})

			t.Run("when destination Mtime is the same source Mtime but in different time zones it should not sync", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: lib.Time(time.Date(2021, 8, 15, 14, 31, 45, 100, time.Local).UTC()),
						sourceMtime:      time.Date(2021, 8, 15, 14, 31, 45, 100, time.Local),
						sync:             false,
					},
				)
			})

			t.Run("when destination Mtime is newer than source Mtime it should not sync", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: lib.Time(time.Date(2021, 8, 16, 14, 31, 45, 100, time.Local)),
						sourceMtime:      time.Date(2021, 8, 15, 14, 31, 45, 100, time.Local),
						sync:             false,
					},
				)
			})

			t.Run("when destination Mtime is nil", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: nil,
						sourceMtime:      time.Now().Add(-time.Hour * 24),
						sync:             true,
					},
				)
			})

			t.Run("when destination Mtime is zero", func(t *testing.T) {
				test(
					t,
					Args{
						destinationMtime: &time.Time{},
						sourceMtime:      time.Now().Add(-time.Hour * 24),
						sync:             true,
					},
				)
			})
		})
	})

	t.Run("No overwrite file exists", func(t *testing.T) {
		uploadStatus, mockFs, _ := init()
		mockFs["test"] = &fstest.MapFile{
			Sys: files_sdk.File{},
		}
		uploadStatus.NoOverwrite = true
		assert.True(excludeFile(uploadStatus, false))
		assert.Equal(status.FileExists, uploadStatus.Status())
	})

	t.Run("No overwrite file does not exists", func(t *testing.T) {
		uploadStatus, _, _ := init()
		uploadStatus.NoOverwrite = true
		uploadStatus.status = status.Queued
		assert.False(excludeFile(uploadStatus, false))
		assert.Equal(status.Queued, uploadStatus.Status())
	})
}

func TestUploader(t *testing.T) {
	mutex := &sync.Mutex{}
	t.Run("uploader", func(t *testing.T) {
		sourceFs := lib.ReadWriteFs(lib.LocalFileSystem{})
		destinationFs := &FS{Context: context.Background()}
		for _, tt := range lib.PathSpec(t, sourceFs.PathSeparator(), destinationFs.PathSeparator()) {
			t.Run(tt.Name, func(t *testing.T) {
				client, r, err := CreateClient(t.Name())
				if err != nil {
					t.Fatal(err)
				}
				config := client.Config
				destinationFs = (&FS{Context: context.Background()}).Init(config, true)
				lib.BuildPathSpecTest(t, mutex, tt, sourceFs, destinationFs, func(args lib.PathSpecArgs) lib.Cmd {
					return &CmdRunner{
						run: func() *Job {
							return client.Uploader(UploaderParams{LocalPath: args.Src, RemotePath: args.Dest, config: config, PreserveTimes: args.PreserveTimes})
						},
						args: []string{args.Src, args.Dest, "--times", fmt.Sprintf("%v", args.PreserveTimes)},
					}
				})
				r.Stop()
			})
		}
	})
}

type ReaderWithOutLen struct {
	buffer *bytes.Buffer
}

func (r ReaderWithOutLen) Read(p []byte) (n int, err error) {
	return r.buffer.Read(p)
}

type ReaderAtWithOutLen struct {
	buffer *bytes.Buffer
}

func (r ReaderAtWithOutLen) ReadAt(p []byte, off int64) (n int, err error) {
	return bytes.NewReader(r.buffer.Bytes()).ReadAt(p, off)
}

func TestUploadReader(t *testing.T) {
	t.Run("reader with nil size", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		server.MockFiles["reader-no-size.txt"] = mockFile{File: files_sdk.File{Size: 5}}
		client := server.Client()
		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReader(ReaderWithOutLen{buffer: bytes.NewBufferString("Hello")}),
			UploadWithDestinationPath("reader-no-size.txt"),
			UploadWithManager(lib.NewConstrainedWorkGroup(2)),
		)

		require.NoError(t, err)

		assert.ElementsMatch(t, []string{"/upload/reader-no-size.txt?part_number=1", "/upload/reader-no-size.txt?part_number=2"}, server.TrackRequest["/upload/*path"])
		assert.Equal(t, "reader-no-size.txt", u.File.Path)
		assert.Equal(t, int64(5), u.Size)
		assert.Len(t, u.Parts, 0, "individual parts are not retryable with nil size")
		assert.Equal(t, "reader-no-size.txt", u.FileUploadPart.Path)
	})

	t.Run("reader with size present", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		server.MockFiles["reader-size.txt"] = mockFile{File: files_sdk.File{Size: 10}}
		client := server.Client()
		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReader(bytes.NewBufferString("0123456789")),
			UploadWithDestinationPath("reader-size.txt"),
			UploadWithManager(lib.NewConstrainedWorkGroup(2)),
		)

		require.NoError(t, err)

		assert.ElementsMatch(t, []string{"/upload/reader-size.txt?part_number=1", "/upload/reader-size.txt?part_number=2", "/upload/reader-size.txt?part_number=3"}, server.TrackRequest["/upload/*path"])
		assert.Equal(t, "reader-size.txt", u.File.Path)
		assert.Equal(t, int64(10), u.Size)
		assert.Len(t, u.Parts, 0, "individual parts are not retryable with nil size")
		assert.Equal(t, "reader-size.txt", u.FileUploadPart.Path)
	})

	t.Run("io.ReaderAt and no size", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		server.MockFiles["reader-at_no-size.txt"] = mockFile{File: files_sdk.File{Size: 10}}
		client := server.Client()
		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReaderAt(ReaderAtWithOutLen{buffer: bytes.NewBufferString("0123456789")}),
			UploadWithDestinationPath("reader-at_no-size.txt"),
			UploadWithManager(lib.NewConstrainedWorkGroup(2)),
		)

		require.NoError(t, err)

		assert.ElementsMatch(t, []string{"/upload/reader-at_no-size.txt?part_number=1", "/upload/reader-at_no-size.txt?part_number=2", "/upload/reader-at_no-size.txt?part_number=3"}, server.TrackRequest["/upload/*path"])
		assert.Equal(t, "reader-at_no-size.txt", u.File.Path)
		assert.Equal(t, int64(10), u.Size)
		assert.Len(t, u.Parts, 0, "individual parts are not retryable with nil size")
		assert.Equal(t, "reader-at_no-size.txt", u.FileUploadPart.Path)
	})

	t.Run("io.ReaderAt and size", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		server.MockFiles["reader-at-size.txt"] = mockFile{File: files_sdk.File{Size: 10}}
		client := server.Client()
		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
			UploadWithDestinationPath("reader-at-size.txt"),
			UploadWithSize(10),
			UploadWithManager(lib.NewConstrainedWorkGroup(2)),
		)

		require.NoError(t, err)

		assert.ElementsMatch(t, []string{"/upload/reader-at-size.txt?part_number=1", "/upload/reader-at-size.txt?part_number=2", "/upload/reader-at-size.txt?part_number=3"}, server.TrackRequest["/upload/*path"])
		assert.Equal(t, "reader-at-size.txt", u.File.Path)
		assert.Equal(t, int64(10), u.Size)
		assert.Len(t, u.Parts, 3, "individual parts are not retryable with nil size")
		assert.Equal(t, "reader-at-size.txt", u.FileUploadPart.Path)
	})

	t.Run("configured user agent is sent on part uploads", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		filename := "user-agent.txt"
		server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 10}}
		client := server.Client()
		client.Config.UserAgent = "Files.com custom client test"
		client.Config.APIKey = "secret-api-key"

		var mu sync.Mutex
		var userAgents []string
		var apiKeys []string
		server.MockRoute("/upload/"+filename, func(c *gin.Context, _ interface{}) bool {
			mu.Lock()
			defer mu.Unlock()
			userAgents = append(userAgents, c.Request.UserAgent())
			apiKeys = append(apiKeys, c.Request.Header.Get("X-FilesAPI-Key"))
			return false
		})

		_, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
			UploadWithDestinationPath(filename),
			UploadWithSize(10),
			UploadWithManager(lib.NewConstrainedWorkGroup(2)),
		)

		require.NoError(t, err)
		require.NotEmpty(t, userAgents)
		for _, userAgent := range userAgents {
			assert.Equal(t, "Files.com custom client test", userAgent)
		}
		for _, apiKey := range apiKeys {
			assert.Empty(t, apiKey, "raw upload requests must not receive API auth headers")
		}
	})

	t.Run("zero-size upload part uses empty request body", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		filename := "zero-size.txt"
		server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 0}}
		client := server.Client()

		var contentLength int64
		var transferEncoding []string
		var uploadRequests int
		server.MockRoute("/upload/"+filename, func(c *gin.Context, _ interface{}) bool {
			uploadRequests++
			contentLength = c.Request.ContentLength
			transferEncoding = c.Request.TransferEncoding
			return false
		})

		u, err := client.UploadWithResume(
			UploadWithReaderAt(bytes.NewReader(nil)),
			UploadWithDestinationPath(filename),
			UploadWithSize(0),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
		)

		require.NoError(t, err)
		require.Equal(t, 1, uploadRequests)
		assert.Equal(t, int64(0), contentLength)
		assert.Empty(t, transferEncoding)
		assert.Equal(t, filename, u.FileUploadPart.Path)
	})

	t.Run("zero-size upload part retries request timeout", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		filename := "zero-size-retry.txt"
		server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 0}}
		client := server.Client()

		var contentLength int64
		var transferEncoding []string
		var attempts int
		server.MockRoute("/upload/"+filename, func(c *gin.Context, _ interface{}) bool {
			attempts++
			contentLength = c.Request.ContentLength
			transferEncoding = c.Request.TransferEncoding
			if attempts == 1 {
				c.Data(
					http.StatusBadRequest,
					"application/xml",
					[]byte(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>RequestTimeout</Code><Message>Your socket connection to the server was not read from or written to within the timeout period. Idle connections will be closed.</Message></Error>`),
				)
				return true
			}

			c.Header("Etag", "etag")
			c.Status(http.StatusOK)
			return true
		})

		u, err := client.UploadWithResume(
			UploadWithReaderAt(bytes.NewReader(nil)),
			UploadWithDestinationPath(filename),
			UploadWithSize(0),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
		)

		require.NoError(t, err)
		require.Equal(t, 2, attempts)
		assert.Equal(t, int64(0), contentLength)
		assert.Empty(t, transferEncoding)
		assert.Equal(t, filename, u.FileUploadPart.Path)
	})

	t.Run("io.ReaderAt and size with resume", func(t *testing.T) {
		firstTry := func(t *testing.T, filename string) (context.Context, context.CancelFunc, *Client, *MockAPIServer, UploadResumable) {
			ctx, cancel := context.WithCancel(context.Background())
			server := (&MockAPIServer{T: t}).Do()
			client := server.Client()

			server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 10}}
			progressMutex := sync.Mutex{}
			bytesUploaded := []int64{0, 0}

			firstTry, err := client.UploadWithResume(
				func(io uploadIO) (uploadIO, error) {
					io.PartSizes = []int64{2, 4, 8, 16, 32}
					return io, nil
				},
				UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
				UploadWithDestinationPath(filename),
				UploadWithSize(10),
				UploadWithManager(lib.NewConstrainedWorkGroup(1)),
				UploadWithContext(ctx),
				UploadRewindAllProgressOnFailure(),
				UploadWithProgress(func(i int64) {
					progressMutex.Lock()
					defer progressMutex.Unlock()
					bytesUploaded[0] += i

					if bytesUploaded[0] > 5 {
						cancel()
					}
				}),
			)
			cancel()
			require.ErrorIs(t, err, context.Canceled)

			assert.Len(t, firstTry.Parts, 3)
			assert.Equal(t, int64(2), firstTry.Parts.SuccessfulBytes())
			assert.Equal(t, int64(0), bytesUploaded[0])
			assert.Equal(t, filename, firstTry.FileUploadPart.Path)
			server.CloseClientConnections()
			ctx, cancel = context.WithCancel(context.Background())
			// The last request might still be processing
			time.Sleep(10 * time.Millisecond)
			return ctx, func() {
					server.CloseClientConnections()
					cancel()
				}, client,
				server,
				firstTry
		}
		// Retry
		retry := func(ctx context.Context, u UploadResumable, client *Client) (UploadResumable, error) {
			return client.UploadWithResume(
				func(io uploadIO) (uploadIO, error) {
					io.PartSizes = []int64{2, 4, 8, 16, 32}
					return io, nil
				},
				UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
				UploadWithDestinationPath(u.FileUploadPart.Path),
				UploadWithSize(10),
				UploadWithManager(lib.NewConstrainedWorkGroup(2)),
				UploadWithContext(ctx),
				UploadWithResume(u),
			)
		}

		t.Run("native", func(t *testing.T) {
			ctx, cancel, client, server, resume := firstTry(t, "native-file")
			defer cancel()

			server.traceMutex.Lock()
			server.TrackRequest = make(map[string][]string)
			server.traceMutex.Unlock()
			var beginUploadRequests []files_sdk.FileBeginUploadParams

			server.MockRoute("/api/rest/v1/file_actions/begin_upload/native-file", func(ctx *gin.Context, model interface{}) bool {
				beginUploadRequests = append(beginUploadRequests, model.(files_sdk.FileBeginUploadParams))
				return false
			})

			u, err := retry(ctx, resume, client)
			require.NoError(t, err)

			assert.ElementsMatch(t, []string{"/api/rest/v1/file_actions/begin_upload/native-file", "/api/rest/v1/file_actions/begin_upload/native-file", "/api/rest/v1/file_actions/begin_upload/native-file"}, server.TrackRequest["/api/rest/v1/file_actions/begin_upload/*path"], "only requests part 3")
			assert.ElementsMatch(t, []string{"/upload/native-file?part_number=1", "/upload/native-file?part_number=2", "/upload/native-file?part_number=3"}, server.TrackRequest["/upload/*path"], "1 already succeed rest are uploaded")
			assert.Equal(t, "native-file", u.File.Path)
			assert.Equal(t, int64(10), u.Size)
			assert.Len(t, u.Parts, 3)
			assert.Equal(t, "native-file", u.FileUploadPart.Path)
		})

		t.Run("remote_mount", func(t *testing.T) {
			ctx, cancel, client, server, resume := firstTry(t, "remote_mount-file")
			defer cancel()

			resume.FileUploadPart.ParallelParts = lib.Bool(false)
			server.traceMutex.Lock()
			server.TrackRequest = make(map[string][]string)
			server.traceMutex.Unlock()
			expires := time.Now().Add(time.Hour).Format(time.RFC3339)
			uploadURLPartNumber := func(path string, part int64) string {
				return fmt.Sprintf("%v?part_number=%v&X-Files-Date=%v&X-Files-Expires=%v", lib.UrlJoinNoEscape("upload", path), part, expires, 180)
			}
			server.MockRoute("/api/rest/v1/file_actions/begin_upload/remote_mount-file", func(ctx *gin.Context, model interface{}) bool {
				file := model.(files_sdk.FileBeginUploadParams)
				if file.Part == 0 {
					file.Part = 1
				}
				path := strings.TrimPrefix(ctx.Param("path"), "/")

				uploadURI := lib.UrlJoinNoEscape(server.URL, uploadURLPartNumber(path, file.Part))
				ctx.JSON(http.StatusOK, files_sdk.FileUploadPartCollection{
					files_sdk.FileUploadPart{
						HttpMethod:    "POST",
						Path:          path,
						UploadUri:     uploadURI,
						ParallelParts: lib.Bool(false),
						Expires:       expires,
						PartNumber:    file.Part,
					},
				})
				return true
			})

			u, err := retry(ctx, resume, client)
			require.NoError(t, err)

			assert.ElementsMatch(t, []string{"/api/rest/v1/file_actions/begin_upload/remote_mount-file"}, server.TrackRequest["/api/rest/v1/file_actions/begin_upload/*path"], "upload is invalided because of ParallelParts")
			assert.ElementsMatch(t, []string{"/upload/remote_mount-file?part_number=2", "/" + uploadURLPartNumber("remote_mount-file", 3)}, server.TrackRequest["/upload/*path"], "all parts are uploaded")
			assert.Equal(t, "remote_mount-file", u.File.Path)
			assert.Equal(t, int64(10), u.Size)
			assert.Len(t, u.Parts, 3)
			assert.Equal(t, int64(10), u.Parts.SuccessfulBytes())
			assert.Equal(t, "remote_mount-file", u.FileUploadPart.Path)
		})

		// Regression test: NewPart (used when deserializing checkpoint files) does not set
		// FileUploadPart fields (Path, Ref, PartNumber, etc.). Previously the repair loop
		// that propagates session info to non-successful parts was gated on ParallelParts=true,
		// so non-parallel uploads resumed from a checkpoint would call begin_upload with an
		// empty path and ref, breaking resume after process restart.
		t.Run("non-parallel resume from deserialized checkpoint", func(t *testing.T) {
			server := (&MockAPIServer{T: t}).Do()
			defer server.Shutdown()

			filename := "non-parallel-checkpoint.txt"
			server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 10}}
			client := server.Client()

			// Simulate checkpoint deserialization via NewPart: only offsets/bytes/etag
			// are restored; FileUploadPart (Path, Ref, PartNumber...) is zero-value.
			parts := Parts{
				NewPart(1, 0, 2, 2, "etag1", "1", ""),           // successful
				NewPart(2, 2, 4, 0, "", "", "context canceled"), // failed - must be retried
				NewPart(3, 6, 4, 0, "", "", "context canceled"), // failed - must be retried
			}
			resumable := UploadResumable{
				FileUploadPart: files_sdk.FileUploadPart{
					Ref:           "checkpoint-ref-123",
					Path:          filename,
					HttpMethod:    "POST",
					ParallelParts: lib.Bool(false),
					Expires:       time.Now().Add(time.Hour).Format(time.RFC3339),
				},
				Parts: parts,
			}

			var mu sync.Mutex
			var beginUploadRequests []files_sdk.FileBeginUploadParams
			expires := time.Now().Add(time.Hour).Format(time.RFC3339)

			server.MockRoute("/api/rest/v1/file_actions/begin_upload/"+filename, func(c *gin.Context, model interface{}) bool {
				params := model.(files_sdk.FileBeginUploadParams)
				mu.Lock()
				beginUploadRequests = append(beginUploadRequests, params)
				mu.Unlock()
				part := params.Part
				if part == 0 {
					part = 1
				}
				c.JSON(http.StatusOK, files_sdk.FileUploadPartCollection{
					files_sdk.FileUploadPart{
						HttpMethod:    "POST",
						Path:          filename,
						UploadUri:     fmt.Sprintf("%v?part_number=%v", lib.UrlJoinNoEscape(server.URL, "upload", filename), part),
						ParallelParts: lib.Bool(false),
						Expires:       expires,
						PartNumber:    part,
					},
				})
				return true
			})

			u, err := client.UploadWithResume(
				func(io uploadIO) (uploadIO, error) {
					io.PartSizes = []int64{2, 4, 8, 16, 32}
					return io, nil
				},
				UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
				UploadWithDestinationPath(filename),
				UploadWithSize(10),
				UploadWithResume(resumable),
				UploadWithManager(lib.NewConstrainedWorkGroup(2)),
			)
			require.NoError(t, err)

			require.NotEmpty(t, beginUploadRequests, "failed parts must call begin_upload to get upload URL")
			for _, req := range beginUploadRequests {
				assert.Equal(t, filename, req.Path, "begin_upload path must come from the checkpoint ref, not be empty")
				assert.Equal(t, "checkpoint-ref-123", req.Ref, "begin_upload ref must come from the checkpoint, not be empty")
			}
			assert.Equal(t, filename, u.File.Path)
		})
	})

	t.Run("missing UploadWithDestinationPath", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		client := server.Client()

		err := client.Upload(
			UploadWithReader(ReaderWithOutLen{buffer: bytes.NewBufferString("Hello")}),
		)
		assert.Equal(t, "UploadWithDestinationPath is required", err.Error())
	})

	t.Run("missing reader", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		client := server.Client()

		err := client.Upload(
			UploadWithDestinationPath("reader-no-size.txt"),
		)
		assert.Equal(t, "UploadWithReader or UploadWithReaderAt required", err.Error())
	})

	accessDenied := CustomResponse{
		Status:      http.StatusForbidden,
		ContentType: "application/xml",
		Body:        []byte(`<?xml version="1.0" encoding="UTF-8"?> <Error><Code>AccessDenied</Code><Message>Request has expired</Message><X-Amz-Expires>900</X-Amz-Expires><Expires>2023-09-06T05:27:01Z</Expires><ServerTime>2023-09-06T05:27:21Z</ServerTime><RequestId>DCZ7NV6P08Y6SKY2</RequestId><HostId>a0ww8xPnO34ZC2to9wizy501VJcZicTFKdohzq5P7SArZuXJ7cCo6GpJbUXITjkyFHNPla8Sd1U=</HostId></Error>`),
	}

	t.Run("socket connection error recovers after retry", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		client := server.Client()
		u, err := client.UploadWithResume(
			UploadWithSize(1024),
			UploadWithReaderAt(bytes.NewReader(make([]byte, 1024))),
			UploadWithDestinationPath("file.bak"),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
		)

		assert.NoError(t, err)
		assert.Equal(t, "file.bak", u.FileUploadPart.Path)
	})

	t.Run("s3 request timeout is retried for rewindable upload part", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		attempts := 0
		server.MockRoute("/upload/file.bak", func(c *gin.Context, _ interface{}) bool {
			attempts++
			if attempts == 1 {
				c.Data(
					http.StatusBadRequest,
					"application/xml",
					[]byte(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>RequestTimeout</Code><Message>Your socket connection to the server was not read from or written to within the timeout period. Idle connections will be closed.</Message></Error>`),
				)
				return true
			}

			c.Header("Etag", "etag")
			c.Status(http.StatusOK)
			return true
		})

		client := server.Client()
		u, err := client.UploadWithResume(
			UploadWithSize(1024),
			UploadWithReaderAt(bytes.NewReader(make([]byte, 1024))),
			UploadWithDestinationPath("file.bak"),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
		)

		require.NoError(t, err)
		require.Equal(t, 2, attempts)
		require.Equal(t, "file.bak", u.FileUploadPart.Path)
	})

	t.Run("s3 request timeout exhausts retryablehttp retries", func(t *testing.T) {
		u, err, attempts := uploadWithS3PartError(t, "RequestTimeout", http.StatusBadRequest, UploadWithReaderAt(bytes.NewReader(make([]byte, 1024))))

		require.Error(t, err)
		require.Contains(t, err.Error(), "RequestTimeout")
		require.Equal(t, 4, attempts)
		require.Equal(t, "file.bak", u.FileUploadPart.Path)
	})

	t.Run("s3 request timeout is not retried when upload part cannot rewind", func(t *testing.T) {
		u, err, attempts := uploadWithS3PartError(t, "RequestTimeout", http.StatusBadRequest, UploadWithReader(bytes.NewBuffer(make([]byte, 1024))))

		require.Error(t, err)
		require.Contains(t, err.Error(), "RequestTimeout")
		require.Equal(t, 1, attempts)
		require.Equal(t, "file.bak", u.FileUploadPart.Path)
	})

	t.Run("multipart s3 timeout error can join repeated provider text before fsmount sanitizes", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		server.MockRoute("/upload/file.bak", func(c *gin.Context, _ interface{}) bool {
			c.Data(
				http.StatusBadRequest,
				"application/xml",
				[]byte(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>RequestTimeout</Code><Message>Your socket connection to the server was not read from or written to within the timeout period. Idle connections will be closed.</Message></Error>`),
			)
			return true
		})

		client := server.Client()
		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithSize(10),
			UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
			UploadWithDestinationPath("file.bak"),
			UploadWithManager(lib.NewConstrainedWorkGroup(3)),
		)

		require.Error(t, err)
		require.GreaterOrEqual(t, len(u.Parts), 2)
		require.GreaterOrEqual(t, strings.Count(err.Error(), "RequestTimeout"), 2)
		require.GreaterOrEqual(t, strings.Count(err.Error(), "Your socket connection"), 2)
	})

	t.Run("transient s3 upload part errors are retried for rewindable upload parts", func(t *testing.T) {
		tests := []struct {
			code   string
			status int
		}{
			{"SlowDown", http.StatusServiceUnavailable},
			{"DatabaseTimeout", http.StatusServiceUnavailable},
		}

		for _, test := range tests {
			t.Run(test.code, func(t *testing.T) {
				server := (&MockAPIServer{T: t}).Do()
				defer server.Shutdown()

				attempts := 0
				server.MockRoute("/upload/file.bak", func(c *gin.Context, _ interface{}) bool {
					attempts++
					if attempts == 1 {
						c.Data(
							test.status,
							"application/xml",
							[]byte(fmt.Sprintf(`<?xml version="1.0" encoding="UTF-8"?><Error><Code>%s</Code><Message>provider message</Message></Error>`, test.code)),
						)
						return true
					}

					c.Header("Etag", "etag")
					c.Status(http.StatusOK)
					return true
				})

				client := server.Client()
				u, err := client.UploadWithResume(
					UploadWithSize(1024),
					UploadWithReaderAt(bytes.NewReader(make([]byte, 1024))),
					UploadWithDestinationPath("file.bak"),
					UploadWithManager(lib.NewConstrainedWorkGroup(1)),
				)

				require.NoError(t, err)
				require.Equal(t, 2, attempts)
				require.Equal(t, "file.bak", u.FileUploadPart.Path)
			})
		}
	})

	t.Run("terminal s3 upload part errors are not retried as transient errors", func(t *testing.T) {
		tests := []struct {
			code   string
			status int
		}{
			{"AccessDenied", http.StatusForbidden},
			{"NoSuchKey", http.StatusNotFound},
			{"NoSuchUpload", http.StatusNotFound},
			{"InvalidPart", http.StatusBadRequest},
			{"EntityTooSmall", http.StatusBadRequest},
			{"InvalidRequest", http.StatusBadRequest},
			{"SignatureDoesNotMatch", http.StatusForbidden},
		}

		for _, test := range tests {
			t.Run(test.code, func(t *testing.T) {
				u, err, attempts := uploadWithS3PartError(t, test.code, test.status, UploadWithReaderAt(bytes.NewReader(make([]byte, 1024))))

				require.Error(t, err)
				require.Contains(t, err.Error(), test.code)
				require.Equal(t, 1, attempts)
				require.Equal(t, "file.bak", u.FileUploadPart.Path)
			})
		}
	})

	t.Run("request expired error", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		server.MockRoute("/upload/file.bak", func(c *gin.Context, _ interface{}) bool {
			c.Data(accessDenied.Status, accessDenied.ContentType, accessDenied.Body)
			return true
		})

		client := server.Client()
		u, err := client.UploadWithResume(
			UploadWithReader(bytes.NewBufferString("0123456789")),
			UploadWithDestinationPath("file.bak"),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
		)

		assert.Equal(t, lib.S3Error{Message: "Request has expired", Code: "AccessDenied"}.Error(), err.Error())
		assert.Equal(t, "file.bak", u.FileUploadPart.Path)
	})

	t.Run("File Upload Not Found", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		var partCount int
		server.MockRoute("/api/rest/v1/file_actions/begin_upload/file.bak", func(c *gin.Context, model interface{}) bool {
			partCount += 1
			if beginUpload, ok := model.(files_sdk.FileBeginUploadParams); ok {
				if beginUpload.Part == 0 {
					return false
				}
			}

			c.Data(http.StatusNotFound, "text", []byte("File Upload Not Found"))
			return true
		})

		client := server.Client()
		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReader(bytes.NewBufferString("0123456789")),
			UploadWithDestinationPath("file.bak"),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
		)
		require.Error(t, err)
		assert.Contains(t, err.Error(), "File Upload Not Found", "it invalidates any resuming")
		assert.Len(t, u.Parts, 0)
		assert.Contains(t, []int{1, 2, 3}, partCount)
	})
}

func TestUploadPauseResume(t *testing.T) {
	t.Run("pause preserves upload session", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		client := server.Client()

		ctx, cancel := context.WithCancelCause(context.Background())
		defer cancel(nil)

		var progressMu sync.Mutex
		var uploaded int64
		resumable, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
			UploadWithDestinationPath("pause-file.txt"),
			UploadWithSize(10),
			UploadWithContext(ctx),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
			UploadWithProgress(func(i int64) {
				progressMu.Lock()
				uploaded += i
				trigger := uploaded > 2 // cancel after part 1 (2 bytes) completes and part 2 starts
				progressMu.Unlock()
				if trigger {
					cancel(ErrJobPaused)
				}
			}),
		)
		_ = err // context.Canceled expected
		assert.Greater(t, resumable.Parts.SuccessfulBytes(), int64(0), "upload session parts should be preserved on pause")
	})

	t.Run("cancel removes upload session", func(t *testing.T) {
		root := t.TempDir()
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		client := server.Client()

		localPath := filepath.Join(root, "cancel-test.txt")
		require.NoError(t, os.WriteFile(localPath, make([]byte, 100), 0644))

		ctx, cancel := context.WithCancel(context.Background())
		cancel() // cancel before start

		job := client.Uploader(
			UploaderParams{LocalPath: localPath, RemotePath: "cancel-test.txt"},
			files_sdk.WithContext(ctx),
		)
		job.Start()
		job.Wait()

		require.Len(t, job.Statuses, 1)
		assert.Equal(t, status.Canceled, job.Statuses[0].Status())
	})

	t.Run("does not duplicate caller events reporter when reusing job", func(t *testing.T) {
		root := t.TempDir()
		localPath := filepath.Join(root, "missing.txt")
		client := &Client{Config: files_sdk.Config{}.Init()}

		var mu sync.Mutex
		reporterCalls := 0
		reporter := CreateFileEvents(func(JobFile) {
			mu.Lock()
			reporterCalls++
			mu.Unlock()
		}, status.Errored)

		job := client.Uploader(
			UploaderParams{
				LocalPath:      localPath,
				RemotePath:     "missing.txt",
				EventsReporter: reporter,
			},
		)
		job.Start()
		job.Wait()

		require.Len(t, job.Statuses, 1)
		assert.Equal(t, status.Errored, job.Statuses[0].Status())

		mu.Lock()
		assert.Equal(t, 1, reporterCalls)
		mu.Unlock()
	})

	t.Run("resume skips completed paths", func(t *testing.T) {
		root := t.TempDir()
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		client := server.Client()

		uploadDir := filepath.Join(root, "upload_dir")
		require.NoError(t, os.MkdirAll(uploadDir, 0755))
		require.NoError(t, os.WriteFile(filepath.Join(uploadDir, "a.txt"), make([]byte, 100), 0644))
		require.NoError(t, os.WriteFile(filepath.Join(uploadDir, "b.txt"), make([]byte, 100), 0644))

		alreadyDone := filepath.Join(uploadDir, "a.txt")

		var mu sync.Mutex
		var reporterCalls []JobFile
		reporter := CreateFileEvents(func(f JobFile) {
			mu.Lock()
			reporterCalls = append(reporterCalls, f)
			mu.Unlock()
		}, append(status.Excluded, status.Included...)...)

		job := client.Uploader(
			UploaderParams{
				LocalPath:          uploadDir + string(os.PathSeparator),
				RemotePath:         "uploaded",
				PriorJobCheckpoint: &JobUploadCheckpoint{CompletedPaths: []string{alreadyDone}},
				EventsReporter:     reporter,
			},
		)
		job.Start()
		job.Wait()

		var aStatuses []status.Status
		for _, c := range reporterCalls {
			if c.LocalPath == alreadyDone {
				aStatuses = append(aStatuses, c.Status)
			}
		}
		assert.Contains(t, aStatuses, status.Skipped)
		assert.NotContains(t, aStatuses, status.Complete)
	})

	// When all bytes are already on the server and the job is paused mid-commit,
	// the SDK must still complete the commit rather than abandoning it.
	// The fix: completeUpload uses context.Background() instead of the job context,
	// so a job cancellation cannot abort the final action=end call.
	t.Run("context canceled during commit still completes upload", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		filename := "pause-cancel-commit.txt"
		server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 10}}
		client := server.Client()

		ctx, cancel := context.WithCancelCause(context.Background())
		defer cancel(nil)

		// Track whether the commit was called with all parts already on the server.
		// The MockRoute fires only after all part uploads complete (partsFinished),
		// so by the time we cancel here, all bytes are on the server.
		commitCalled := false
		server.MockRoute("/api/rest/v1/files/"+filename, func(c *gin.Context, model interface{}) bool {
			commitCalled = true
			cancel(ErrJobPaused) // pause arrives during commit
			return false         // let server respond 200 — bytes are already there
		})

		u, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
			UploadWithDestinationPath(filename),
			UploadWithSize(10),
			UploadWithContext(ctx),
			UploadWithManager(lib.NewConstrainedWorkGroup(2)),
		)

		require.True(t, commitCalled, "commit endpoint must be reached (all bytes on server)")
		require.NoError(t, err, "upload must complete even when job context is canceled during commit")
		assert.Equal(t, filename, u.File.Path)
		assert.Equal(t, int64(10), u.Size)
	})

	// When the job context is canceled before all bytes land on the server,
	// the upload must fail — commit is never called because partsFinished never fires cleanly.
	t.Run("context canceled mid-upload does not commit", func(t *testing.T) {
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()

		filename := "mid-upload-cancel.txt"
		server.MockFiles[filename] = mockFile{File: files_sdk.File{Size: 10}}
		client := server.Client()

		ctx, cancel := context.WithCancelCause(context.Background())
		defer cancel(nil)

		// Cancel while part bytes are still being sent — before all parts succeed.
		var progressMu sync.Mutex
		var uploaded int64
		commitCalled := false

		server.MockRoute("/api/rest/v1/files/"+filename, func(c *gin.Context, model interface{}) bool {
			commitCalled = true
			return false
		})

		_, err := client.UploadWithResume(
			func(io uploadIO) (uploadIO, error) {
				io.PartSizes = []int64{2, 4, 8, 16, 32}
				return io, nil
			},
			UploadWithReaderAt(bytes.NewReader(bytes.NewBufferString("0123456789").Bytes())),
			UploadWithDestinationPath(filename),
			UploadWithSize(10),
			UploadWithContext(ctx),
			UploadWithManager(lib.NewConstrainedWorkGroup(1)),
			UploadWithProgress(func(i int64) {
				progressMu.Lock()
				uploaded += i
				trigger := uploaded >= 2 // cancel as soon as part 1 bytes arrive
				progressMu.Unlock()
				if trigger {
					cancel(ErrJobPaused)
				}
			}),
		)

		require.Error(t, err, "upload must fail when context is canceled before all bytes are on server")
		assert.False(t, commitCalled, "commit must not be called when bytes are incomplete")
	})

	t.Run("resume uses pending parts", func(t *testing.T) {
		root := t.TempDir()
		server := (&MockAPIServer{T: t}).Do()
		defer server.Shutdown()
		client := server.Client()

		// Use a folder upload so PriorJobCheckpoint.PendingParts is applied via buildUploadStatus
		uploadDir := filepath.Join(root, "upload_dir")
		require.NoError(t, os.MkdirAll(uploadDir, 0755))
		filename := "pending-parts.txt"
		localPath := filepath.Join(uploadDir, filename)
		require.NoError(t, os.WriteFile(localPath, []byte("0123456789"), 0644))
		server.MockFiles[filename] = mockFile{File: files_sdk.File{Path: filename, DisplayName: filename, Size: 10}}

		expires := time.Now().Add(time.Hour).Format(time.RFC3339)
		parts := Parts{
			NewPart(1, 0, 2, 2, "etag1", "ref-123", ""),     // successful - should not re-upload
			NewPart(2, 2, 4, 0, "", "", "context canceled"), // failed - must be retried
			NewPart(3, 6, 4, 0, "", "", "context canceled"), // failed - must be retried
		}
		resumable := UploadResumable{
			FileUploadPart: files_sdk.FileUploadPart{
				Ref:           "ref-123",
				Path:          filename,
				HttpMethod:    "POST",
				ParallelParts: lib.Bool(false),
				Expires:       expires,
			},
			Parts: parts,
		}

		server.MockRoute("/api/rest/v1/file_actions/begin_upload/"+filename, func(c *gin.Context, model interface{}) bool {
			params := model.(files_sdk.FileBeginUploadParams)
			part := params.Part
			if part == 0 {
				part = 1
			}
			c.JSON(http.StatusOK, files_sdk.FileUploadPartCollection{
				files_sdk.FileUploadPart{
					HttpMethod:    "POST",
					Path:          filename,
					UploadUri:     fmt.Sprintf("%v?part_number=%v", lib.UrlJoinNoEscape(server.URL, "upload", filename), part),
					ParallelParts: lib.Bool(false),
					Expires:       expires,
					PartNumber:    part,
					Ref:           "ref-123",
				},
			})
			return true
		})

		job := client.Uploader(
			UploaderParams{
				LocalPath:          uploadDir + string(os.PathSeparator),
				RemotePath:         "",
				PriorJobCheckpoint: &JobUploadCheckpoint{PendingParts: map[string]UploadResumable{localPath: resumable}},
			},
		)
		job.Start()
		job.Wait()

		require.NoError(t, job.Statuses[0].Err())

		server.traceMutex.Lock()
		uploadRequests := server.TrackRequest["/upload/*path"]
		server.traceMutex.Unlock()

		assert.NotContains(t, uploadRequests, "/upload/"+filename+"?part_number=1", "part 1 was already uploaded")
		assert.Contains(t, uploadRequests, "/upload/"+filename+"?part_number=2", "part 2 should be uploaded")
	})
}
