package pimport

import (
	"context"
	"fmt"
	"strings"
	"testing"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/aws/aws-sdk-go/aws/awserr"
	"github.com/aws/aws-sdk-go/aws/request"
	"github.com/aws/aws-sdk-go/service/s3"
	"github.com/stretchr/testify/require"
	s3util "github.com/tidbcloud/dataflow-service/pkg/s3"
)

type s3clientMock struct {
	listObjectsWithContext func(ctx aws.Context, input *s3.ListObjectsInput, opts ...request.Option) (*s3.ListObjectsOutput, error)
}

func (client *s3clientMock) ListObjectsWithContext(ctx aws.Context, input *s3.ListObjectsInput, opts ...request.Option) (*s3.ListObjectsOutput, error) {
	return client.listObjectsWithContext(ctx, input, opts...)
}

func TestValidaS3Access(t *testing.T) {
	type test struct {
		resp *s3.ListObjectsOutput
		err  error

		// result
		valid          bool
		reasonContains string
		shouldErr      bool
	}

	nonEmptyResp := &s3.ListObjectsOutput{
		// Contents: []s3.Ob
	}

	emptyResp := &s3.ListObjectsOutput{
		Contents:    nil,
		IsTruncated: aws.Bool(false),
	}

	tests := []test{
		// test all handle awserr code case
		{
			resp:           nonEmptyResp,
			err:            awserr.New(s3.ErrCodeNoSuchBucket, "", nil),
			valid:          false,
			reasonContains: "does not exist",
			shouldErr:      false,
		},
		{
			resp:           nonEmptyResp,
			err:            awserr.New("BucketRegionError", "", nil),
			valid:          false,
			reasonContains: "incorrect region, the bucket is not in",
			shouldErr:      false,
		},
		{
			resp:           nonEmptyResp,
			err:            awserr.New("AccessDenied", "", nil),
			valid:          false,
			reasonContains: "when calling the ListObjectsV2 operation: Access Denied",
			shouldErr:      false,
		},
		{
			resp:           nonEmptyResp,
			err:            awserr.New("AuthorizationHeaderMalformed", "the region 'us-east-1' is wrong;", nil),
			valid:          false,
			reasonContains: "the region",
			shouldErr:      false,
		},
		{
			resp:           nonEmptyResp,
			err:            awserr.New("AuthorizationHeaderMalformed", "maybe some other error", nil),
			valid:          false,
			reasonContains: "",
			shouldErr:      true,
		},
		// test not handle awserr error code
		{
			resp:           nonEmptyResp,
			err:            awserr.New("UnknowCode", "", nil),
			valid:          false,
			reasonContains: "the region",
			shouldErr:      true,
		},
		// test empty objects
		{
			resp:           emptyResp,
			err:            nil,
			valid:          false,
			reasonContains: "no any objects under:",
			shouldErr:      false,
		},
	}

	for _, test := range tests {
		mock := &s3clientMock{
			listObjectsWithContext: func(ctx aws.Context, input *s3.ListObjectsInput, opts ...request.Option) (*s3.ListObjectsOutput, error) {
				return test.resp, test.err
			},
		}

		valid, reason, err := validateS3Access(context.Background(), mock, "s3://dumy/dumy", "dumy-region")

		if test.shouldErr {
			require.NotNil(t, err)
		} else {
			require.Nil(t, err)
			require.Equal(t, test.valid, valid)
			if test.reasonContains != "" {
				require.True(t, strings.Contains(reason, test.reasonContains))
			}
		}
	}
}

func TestTrulyRequest(t *testing.T) {
	t.Skip("skip truly request test, manual uncomment this and run the case to see response from aws if need")

	client, err := s3util.CreateClientWithAssumeRole("us-east-1", "", "")
	require.Nil(t, err)

	valid, reason, err := validateS3Access(context.Background(), client, "s3://tidbcloud-samples-data/data-ingestion/", "us-east-1")

	fmt.Println(err)
	fmt.Println(valid)
	fmt.Println(reason)
}