rclone/fs/operations/reopen_test.go

package operations

import (
	"context"
	"errors"
	"io"
	"testing"

	"github.com/rclone/rclone/fs"
	"github.com/rclone/rclone/fs/hash"
	"github.com/rclone/rclone/fstest/mockobject"
	"github.com/rclone/rclone/lib/pool"
	"github.com/rclone/rclone/lib/readers"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

// check interfaces
var (
	_ io.ReadSeekCloser      = (*ReOpen)(nil)
	_ pool.DelayAccountinger = (*ReOpen)(nil)
)

var errorTestError = errors.New("test error")

// this is a wrapper for a mockobject with a custom Open function
//
// breaks indicate the number of bytes to read before returning an
// error
type reOpenTestObject struct {
	fs.Object
	t           *testing.T
	breaks      []int64
	unknownSize bool
}

// Open opens the file for read.  Call Close() on the returned io.ReadCloser
//
// This will break after reading the number of bytes in breaks
func (o *reOpenTestObject) Open(ctx context.Context, options ...fs.OpenOption) (io.ReadCloser, error) {
	gotHash := false
	gotRange := false
	startPos := int64(0)
	for _, option := range options {
		switch x := option.(type) {
		case *fs.HashesOption:
			gotHash = true
		case *fs.RangeOption:
			gotRange = true
			startPos = x.Start
			if o.unknownSize {
				assert.Equal(o.t, int64(-1), x.End)
			}
		case *fs.SeekOption:
			startPos = x.Offset
		}
	}
	// Check if ranging, mustn't have hash if offset != 0
	if gotHash && gotRange {
		assert.Equal(o.t, int64(0), startPos)
	}
	rc, err := o.Object.Open(ctx, options...)
	if err != nil {
		return nil, err
	}
	if len(o.breaks) > 0 {
		// Pop a breakpoint off
		N := o.breaks[0]
		o.breaks = o.breaks[1:]
		// If 0 then return an error immediately
		if N == 0 {
			return nil, errorTestError
		}
		// Read N bytes then an error
		r := io.MultiReader(&io.LimitedReader{R: rc, N: N}, readers.ErrorReader{Err: errorTestError})
		// Wrap with Close in a new readCloser
		rc = readCloser{Reader: r, Closer: rc}
	}
	return rc, nil
}

func TestReOpen(t *testing.T) {
	for _, testName := range []string{"Normal", "WithRangeOption", "WithSeekOption", "UnknownSize"} {
		t.Run(testName, func(t *testing.T) {
			// Contents for the mock object
			var (
				reOpenTestcontents = []byte("0123456789")
				expectedRead       = reOpenTestcontents
				rangeOption        *fs.RangeOption
				seekOption         *fs.SeekOption
				unknownSize        = false
			)
			switch testName {
			case "Normal":
			case "WithRangeOption":
				rangeOption = &fs.RangeOption{Start: 1, End: 7} // range is inclusive
				expectedRead = reOpenTestcontents[1:8]
			case "WithSeekOption":
				seekOption = &fs.SeekOption{Offset: 2}
				expectedRead = reOpenTestcontents[2:]
			case "UnknownSize":
				rangeOption = &fs.RangeOption{Start: 1, End: -1}
				expectedRead = reOpenTestcontents[1:]
				unknownSize = true
			default:
				panic("bad test name")
			}

			// Start the test with the given breaks
			testReOpen := func(breaks []int64, maxRetries int) (*ReOpen, error) {
				srcOrig := mockobject.New("potato").WithContent(reOpenTestcontents, mockobject.SeekModeNone)
				srcOrig.SetUnknownSize(unknownSize)
				src := &reOpenTestObject{
					Object:      srcOrig,
					t:           t,
					breaks:      breaks,
					unknownSize: unknownSize,
				}
				opts := []fs.OpenOption{}
				if rangeOption == nil && seekOption == nil {
					opts = append(opts, &fs.HashesOption{Hashes: hash.NewHashSet(hash.MD5)})
				}
				if rangeOption != nil {
					opts = append(opts, rangeOption)
				}
				if seekOption != nil {
					opts = append(opts, seekOption)
				}
				return NewReOpen(context.Background(), src, maxRetries, opts...)
			}

			t.Run("Basics", func(t *testing.T) {
				// open
				h, err := testReOpen(nil, 10)
				assert.NoError(t, err)

				// Check contents read correctly
				got, err := io.ReadAll(h)
				assert.NoError(t, err)
				assert.Equal(t, expectedRead, got)

				// Check read after end
				var buf = make([]byte, 1)
				n, err := h.Read(buf)
				assert.Equal(t, 0, n)
				assert.Equal(t, io.EOF, err)

				// Rewind the stream
				_, err = h.Seek(0, io.SeekStart)
				require.NoError(t, err)

				// Check contents read correctly
				got, err = io.ReadAll(h)
				assert.NoError(t, err)
				assert.Equal(t, expectedRead, got)

				// Check close
				assert.NoError(t, h.Close())

				// Check double close
				assert.Equal(t, errFileClosed, h.Close())

				// Check read after close
				n, err = h.Read(buf)
				assert.Equal(t, 0, n)
				assert.Equal(t, errFileClosed, err)
			})

			t.Run("ErrorAtStart", func(t *testing.T) {
				// open with immediate breaking
				h, err := testReOpen([]int64{0}, 10)
				assert.Equal(t, errorTestError, err)
				assert.Nil(t, h)
			})

			t.Run("WithErrors", func(t *testing.T) {
				// open with a few break points but less than the max
				h, err := testReOpen([]int64{2, 1, 3}, 10)
				assert.NoError(t, err)

				// check contents
				got, err := io.ReadAll(h)
				assert.NoError(t, err)
				assert.Equal(t, expectedRead, got)

				// check close
				assert.NoError(t, h.Close())
			})

			t.Run("TooManyErrors", func(t *testing.T) {
				// open with a few break points but >= the max
				h, err := testReOpen([]int64{2, 1, 3}, 3)
				assert.NoError(t, err)

				// check contents
				got, err := io.ReadAll(h)
				assert.Equal(t, errorTestError, err)
				assert.Equal(t, expectedRead[:6], got)

				// check old error is returned
				var buf = make([]byte, 1)
				n, err := h.Read(buf)
				assert.Equal(t, 0, n)
				assert.Equal(t, errTooManyTries, err)

				// Check close
				assert.Equal(t, errFileClosed, h.Close())
			})

			t.Run("Seek", func(t *testing.T) {
				// open
				h, err := testReOpen([]int64{2, 1, 3}, 10)
				assert.NoError(t, err)

				// Seek to end
				pos, err := h.Seek(int64(len(expectedRead)), io.SeekStart)
				assert.NoError(t, err)
				assert.Equal(t, int64(len(expectedRead)), pos)

				// Seek to start
				pos, err = h.Seek(0, io.SeekStart)
				assert.NoError(t, err)
				assert.Equal(t, int64(0), pos)

				// Should not allow seek past end
				pos, err = h.Seek(int64(len(expectedRead))+1, io.SeekCurrent)
				if !unknownSize {
					assert.Equal(t, errSeekPastEnd, err)
					assert.Equal(t, len(expectedRead), int(pos))
				} else {
					assert.Equal(t, nil, err)
					assert.Equal(t, len(expectedRead)+1, int(pos))

					// Seek back to start to get tests in sync
					pos, err = h.Seek(0, io.SeekStart)
					assert.NoError(t, err)
					assert.Equal(t, int64(0), pos)
				}

				// Should not allow seek to negative position start
				pos, err = h.Seek(-1, io.SeekCurrent)
				assert.Equal(t, errNegativeSeek, err)
				assert.Equal(t, 0, int(pos))

				// Should not allow seek with invalid whence
				pos, err = h.Seek(0, 3)
				assert.Equal(t, errInvalidWhence, err)
				assert.Equal(t, 0, int(pos))

				// check read
				dst := make([]byte, 5)
				n, err := h.Read(dst)
				assert.Nil(t, err)
				assert.Equal(t, 5, n)
				assert.Equal(t, expectedRead[:5], dst)

				// Test io.SeekCurrent
				pos, err = h.Seek(-3, io.SeekCurrent)
				assert.Nil(t, err)
				assert.Equal(t, 2, int(pos))

				// check read
				n, err = h.Read(dst)
				assert.Nil(t, err)
				assert.Equal(t, 5, n)
				assert.Equal(t, expectedRead[2:7], dst)

				pos, err = h.Seek(-2, io.SeekCurrent)
				assert.Nil(t, err)
				assert.Equal(t, 5, int(pos))

				// Test io.SeekEnd
				pos, err = h.Seek(-3, io.SeekEnd)
				if !unknownSize {
					assert.Nil(t, err)
					assert.Equal(t, len(expectedRead)-3, int(pos))
				} else {
					assert.Equal(t, errBadEndSeek, err)
					assert.Equal(t, 0, int(pos))

					// sync
					pos, err = h.Seek(1, io.SeekCurrent)
					assert.Nil(t, err)
					assert.Equal(t, 6, int(pos))
				}

				// check read
				dst = make([]byte, 3)
				n, err = h.Read(dst)
				assert.Nil(t, err)
				assert.Equal(t, 3, n)
				assert.Equal(t, expectedRead[len(expectedRead)-3:], dst)

				// check close
				assert.NoError(t, h.Close())
				_, err = h.Seek(0, io.SeekCurrent)
				assert.Equal(t, errFileClosed, err)
			})

			t.Run("AccountRead", func(t *testing.T) {
				h, err := testReOpen(nil, 10)
				assert.NoError(t, err)

				var total int
				h.SetAccounting(func(n int) error {
					total += n
					return nil
				})

				dst := make([]byte, 3)
				n, err := h.Read(dst)
				assert.Equal(t, 3, n)
				assert.NoError(t, err)
				assert.Equal(t, 3, total)
			})

			t.Run("AccountReadDelay", func(t *testing.T) {
				h, err := testReOpen(nil, 10)
				assert.NoError(t, err)

				var total int
				h.SetAccounting(func(n int) error {
					total += n
					return nil
				})

				rewind := func() {
					_, err := h.Seek(0, io.SeekStart)
					require.NoError(t, err)
				}

				h.DelayAccounting(3)

				dst := make([]byte, 16)

				n, err := h.Read(dst)
				assert.Equal(t, len(expectedRead), n)
				assert.Equal(t, io.EOF, err)
				assert.Equal(t, 0, total)
				rewind()

				n, err = h.Read(dst)
				assert.Equal(t, len(expectedRead), n)
				assert.Equal(t, io.EOF, err)
				assert.Equal(t, 0, total)
				rewind()

				n, err = h.Read(dst)
				assert.Equal(t, len(expectedRead), n)
				assert.Equal(t, io.EOF, err)
				assert.Equal(t, len(expectedRead), total)
				rewind()

				n, err = h.Read(dst)
				assert.Equal(t, len(expectedRead), n)
				assert.Equal(t, io.EOF, err)
				assert.Equal(t, 2*len(expectedRead), total)
				rewind()
			})

			t.Run("AccountReadError", func(t *testing.T) {
				// Test accounting errors
				h, err := testReOpen(nil, 10)
				assert.NoError(t, err)

				h.SetAccounting(func(n int) error {
					return errorTestError
				})

				dst := make([]byte, 3)
				n, err := h.Read(dst)
				assert.Equal(t, 3, n)
				assert.Equal(t, errorTestError, err)
			})
		})
	}
}