package pimport

import (
	"context"
	"encoding/json"
	"testing"

	"github.com/aws/aws-sdk-go/aws"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
)

func TestBuildImportSkipChecksum(t *testing.T) {
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	imp, err := buildImport(ctx, &dataflow.ImportCreateReq{
		ClusterId:      123,
		SourceUrl:      "s3://bucket/path/",
		Region:         "us-west-2",
		DataFormat:     uint32(dataflow.ImportDataFormat_SQL),
		DataSourceType: uint32(dataflow.ImportDataSource_S3_SOURCE),
		OrgId:          1,
		ProjectId:      2,
		AwsRoleArn:     "role",
		SkipChecksum:   true,
	})
	require.NoError(t, err)
	require.NotNil(t, imp.OptionalArgs)
	require.True(t, imp.OptionalArgs.SkipChecksum)
}

func TestBuildCSVCfg(t *testing.T) {
	type testCase struct {
		CaseName             string
		CSVFormat            *dataflow.CustomCSVFormat
		ExpectedCSVConfigStr string
	}
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	for _, tc := range []testCase{
		{
			CaseName: "basic",
			CSVFormat: &dataflow.CustomCSVFormat{
				Separator:         ",",
				Delimiter:         "\"",
				Header:            true,
				NotNull:           nil,
				Null:              aws.String("\\N"),
				BackslashEscape:   true,
				TrimLastSeparator: nil,
				HeaderSchemaMatch: aws.Bool(true),
			},
			ExpectedCSVConfigStr: `{"separator":",","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true,"header_schema_match":true}`,
		},
		{
			CaseName: "missing_header_schema_match",
			CSVFormat: &dataflow.CustomCSVFormat{
				Separator:         ",",
				Delimiter:         "\"",
				Header:            true,
				NotNull:           nil,
				Null:              aws.String("\\N"),
				BackslashEscape:   true,
				TrimLastSeparator: nil,
			},
			ExpectedCSVConfigStr: `{"separator":",","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true}`,
		},
		{
			CaseName: "missing_header_schema_match_02",
			CSVFormat: &dataflow.CustomCSVFormat{
				Separator:         ",",
				Delimiter:         "\"",
				Header:            false,
				NotNull:           nil,
				Null:              aws.String("\\N"),
				BackslashEscape:   true,
				TrimLastSeparator: nil,
			},
			ExpectedCSVConfigStr: `{"separator":",","delimiter":"\"","null":"\\N","backslash_escape":true}`,
		},
		{
			CaseName: "tab_separator",
			CSVFormat: &dataflow.CustomCSVFormat{
				Separator:         "\\t",
				Delimiter:         "\"",
				Header:            true,
				NotNull:           nil,
				Null:              aws.String("\\N"),
				BackslashEscape:   true,
				TrimLastSeparator: nil,
			},
			ExpectedCSVConfigStr: `{"separator":"\t","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true}`,
		},
		{
			CaseName: "skip_header_schema_match",
			CSVFormat: &dataflow.CustomCSVFormat{
				Separator:         ",",
				Delimiter:         "\"",
				Header:            true,
				NotNull:           nil,
				Null:              aws.String("\\N"),
				BackslashEscape:   true,
				TrimLastSeparator: nil,
				HeaderSchemaMatch: aws.Bool(false),
			},
			ExpectedCSVConfigStr: `{"separator":",","delimiter":"\"","header":true,"null":"\\N","backslash_escape":true,"header_schema_match":false}`,
		},
	} {
		csvCfgStr, err := buildCsvCfg(ctx, tc.CSVFormat)
		require.NoErrorf(t, err, "case name: %s", tc.CaseName)
		require.Equalf(t, tc.ExpectedCSVConfigStr, csvCfgStr, "case name: %s", tc.CaseName)
		csvConfig := new(dataflow.CustomCSVFormat)
		require.NoErrorf(t, json.Unmarshal([]byte(csvCfgStr), csvConfig), "case name: %s", tc.CaseName)
		require.Equalf(t, tc.CSVFormat, csvConfig, "case name: %s", tc.CaseName)
	}
}