package processor

import (
	"fmt"
	"testing"

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

func TestDataTask_CleanAndProcessData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	s3Cli := newMockS3Client(s3)
	processor := newMockDataProcessor(s3Cli)
	task := DataTask{
		p: processor,
	}
	t.Run("normal", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{}, nil).Times(1)
		task.CleanAndProcessData("test", []string{"ossinsight"}, "v1")
	})
	t.Run("list error", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{}, fmt.Errorf("error")).Times(1)
		task.CleanAndProcessData("test", []string{"ossinsight"}, "v1")
	})
}

func TestDataTask_MergeData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	s3Client := newMockS3Client(s3)
	processor := newMockDataProcessor(s3Client)
	task := DataTask{
		p: processor,
	}
	t.Run("normal", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{}, nil).Times(1)
		task.MergeData("test", "test", []string{"ossinsight"}, "v1")
	})
	t.Run("list error", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{}, fmt.Errorf("error")).Times(1)
		task.MergeData("test", "test", []string{"ossinsight"}, "v1")
	})
}

func TestDataTask_JoinData(t *testing.T) {
	ctl := gomock.NewController(t)
	s3 := mock_data_processor.NewMocks3Client(ctl)
	s3Client := newMockS3Client(s3)
	processor := newMockDataProcessor(s3Client)
	task := DataTask{
		p: processor,
	}
	t.Run("normal", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{}, nil).Times(1)
		task.JoinData("test", "test", []string{"ossinsight"}, "v1")
	})
	t.Run("list error", func(t *testing.T) {
		s3.EXPECT().ListObjects(gomock.Any(), gomock.Any(), gomock.Any()).Return(&s3api.ListObjectsOutput{}, fmt.Errorf("error")).Times(1)
		task.JoinData("test", "test", []string{"ossinsight"}, "v1")
	})
}