package s3storage

import (
	"bytes"
	"context"
	"fmt"
	"io/ioutil"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/tidbcloud/metering/pkg/errors"
	"github.com/tidbcloud/metering/test/mocks3storage"
	"go.uber.org/mock/gomock"
)

func TestBuildS3Checkpoint(t *testing.T) {
	r := BuildS3Checkpoint("1111", "pre", nil)
	assert.Equal(t, "1111", r.bucket)
	assert.Equal(t, "pre", r.prefix)
}

var file = fmt.Sprintf("%s/%s/%s", "prefix", "test", checkpointFileName)

func TestS3Checkpoint_GetCheckpoint(t *testing.T) {
	ctrl := gomock.NewController(t)
	mockClient := mocks3storage.NewMockClient(ctrl)
	rc := ioutil.NopCloser(bytes.NewBufferString("checkpoint:\n  1111"))

	mockClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(&rc, nil).Times(1)

	s := BuildS3Checkpoint("bucket", "prefix", mockClient)

	checkpoint, err := s.GetCheckpoint(context.TODO(), "test")
	assert.NoError(t, err)
	assert.Equal(t, int64(1111), checkpoint)
}

func TestMarshal_error(t *testing.T) {
	ctrl := gomock.NewController(t)
	mockClient := mocks3storage.NewMockClient(ctrl)
	rc := ioutil.NopCloser(bytes.NewBufferString("checkpoint  1111"))

	mockClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(&rc, nil).Times(1)

	s := BuildS3Checkpoint("bucket", "prefix", mockClient)

	_, err := s.GetCheckpoint(context.TODO(), "test")
	assert.Error(t, err)
}

func TestS3Checkpoint_GetCheckpoint_error(t *testing.T) {
	ctrl := gomock.NewController(t)
	mockClient := mocks3storage.NewMockClient(ctrl)

	mockClient.EXPECT().Get(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errors.New(errors.OSError, "StatusCode: 404")).Times(1)

	s := BuildS3Checkpoint("bucket", "prefix", mockClient)

	checkpoint, err := s.GetCheckpoint(context.TODO(), "test")
	assert.Equal(t, int64(0), checkpoint)
	assert.NoError(t, err)
}

func TestS3Checkpoint_SaveCheckpoint(t *testing.T) {
	ctrl := gomock.NewController(t)
	mockClient := mocks3storage.NewMockClient(ctrl)
	mockClient.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(nil).Times(1)
	s := BuildS3Checkpoint("bucket", "prefix", mockClient)
	err := s.SaveCheckpoint(context.TODO(), "test", int64(1111))
	assert.NoError(t, err)
}

func TestS3Checkpoint_SaveCheckpoint_error(t *testing.T) {
	ctrl := gomock.NewController(t)
	mockClient := mocks3storage.NewMockClient(ctrl)
	mockClient.EXPECT().Put(gomock.Any(), gomock.Any(), gomock.Any(), gomock.Any()).Return(errors.New(errors.UnrecognizedError, "")).Times(1)
	s := BuildS3Checkpoint("bucket", "prefix", mockClient)
	err := s.SaveCheckpoint(context.TODO(), "test", int64(1111))
	assert.Error(t, err)
}