package processor

import (
	"bytes"
	"fmt"
	"os"
	"testing"
	"time"

	"github.com/tidbcloud/observability/data-collector/pkg/common"

	"github.com/aws/aws-sdk-go-v2/aws"
	s3api "github.com/aws/aws-sdk-go-v2/service/s3"
	"github.com/aws/aws-sdk-go-v2/service/s3/types"
	"github.com/golang/mock/gomock"
	"github.com/stretchr/testify/assert"
	mock_data_processor "github.com/tidbcloud/observability/data-collector/test/mockdataprocessor"
)

func TestDataProcessor_CleanData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	mockS3 := newMockS3Client(s3)
	processor := newMockDataProcessor(mockS3)
	t.Run("no checkpoint", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{
			Contents: []types.Object{
				{
					Key: aws.String("20230308-12-0-timestamp_explorer_questions.csv"),
				},
			},
			Prefix: aws.String("test/data"),
		}, nil).Times(1)
		buf := make([]byte, 1)
		tmp := CBuffer{bytes.NewBuffer(buf)}
		s3.EXPECT().GetObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.GetObjectOutput{
			Body:          &tmp,
			ContentLength: 1,
		}, nil).Times(1)

		err := processor.CleanData("test", "test", "ossinsight", "v1")
		defer deleteData("./test")
		assert.NoError(t, err)
	})
	t.Run("checkpoint", func(t *testing.T) {
		ts := time.Now().Unix() - 1800
		os.MkdirAll("test/ossinsight/v1/", 0700)
		SaveCheckPoint("test/ossinsight/v1/checkpoint.json", ts)
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{
			Contents: []types.Object{
				{
					Key: aws.String("20230308-12-0-timestamp_explorer_questions.csv"),
				},
			},
			Prefix: aws.String("test/data"),
		}, nil).Times(1)
		buf := make([]byte, 1)
		tmp := CBuffer{bytes.NewBuffer(buf)}
		s3.EXPECT().GetObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.GetObjectOutput{
			Body:          &tmp,
			ContentLength: 1,
		}, nil).Times(1)

		err := processor.CleanData("test", "test", "ossinsight", "v1")
		defer deleteData("./test")
		assert.NoError(t, err)
	})
	t.Run("download err", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, fmt.Errorf("error")).Times(1)
		err := processor.CleanData("test", "test", "ossinsight", "v1")
		assert.Error(t, err)
	})
	t.Run("get checkpoint err", func(t *testing.T) {
		GetCheckPointFunc = func(fn string) (*common.CheckPoint, error) {
			return nil, fmt.Errorf("error")
		}
		defer func() {
			GetCheckPointFunc = func(fn string) (*common.CheckPoint, error) {
				cp, err := GetCheckPoint(fn)
				return cp, err
			}
		}()

		err := processor.CleanData("test", "test", "ossinsight", "v1")
		assert.Error(t, err)
	})
	t.Run("write data err", func(t *testing.T) {
		writeDataFunc = func(object *s3api.GetObjectOutput, outPath string, name string) error {
			return fmt.Errorf("error")
		}
		defer func() {
			writeDataFunc = func(object *s3api.GetObjectOutput, outPath string, name string) error {
				return writeData(object, outPath, name)
			}
		}()
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{
			Contents: []types.Object{
				{
					Key: aws.String("20230308-12-0-timestamp_explorer_questions.csv"),
				},
			},
			Prefix: aws.String("test/data"),
		}, nil).Times(1)
		buf := make([]byte, 1)
		tmp := CBuffer{bytes.NewBuffer(buf)}
		s3.EXPECT().GetObject(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.GetObjectOutput{
			Body:          &tmp,
			ContentLength: 1,
		}, nil).Times(1)

		err := processor.CleanData("test", "test", "ossinsight", "v1")
		defer deleteData("./test")
		assert.Error(t, err)
	})

}