package exportaction

import (
	"os"
	"path/filepath"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/pkg/template/render"
)

func TestParseExportDumplingConfig(t *testing.T) {
	cfg, err := ParseExportDumplingConfig(`
node = "4c8g"
cpuRequest = "2"
cpuLimit = "4"
memoryRequest = "4Gi"
memoryLimit = "7Gi"
tidbGroupNodeCount = 3
tidbGroupNode = "8c16g"
thread = 4
`)
	require.NoError(t, err)
	assert.Equal(t, "4C8G", cfg.Node)
	assert.Equal(t, "2", cfg.CPURequest)
	assert.Equal(t, "4", cfg.CPULimit)
	assert.Equal(t, "4Gi", cfg.MemoryRequest)
	assert.Equal(t, "7Gi", cfg.MemoryLimit)
	assert.Equal(t, uint32(3), cfg.TiDBGroupNodeCount)
	assert.Equal(t, "8C16G", cfg.TiDBGroupNodeSpec)
	assert.Equal(t, int(4), cfg.Thread)
	assert.True(t, cfg.HasTiDBGroupScaleConfig())
}
func TestRenderExportDumplingJobTemplate(t *testing.T) {
	templatePath := filepath.Join("..", "..", "..", "..", "deployments", "templates", "export_dumpling_job.templ")
	content, err := os.ReadFile(templatePath)
	require.NoError(t, err)

	rendered := render.RenderContent(string(content), map[string]string{
		"JobName":       "job-1",
		"Namespace":     "ns-1",
		"SharedPoolId":  "sp-1",
		"NodeSpec":      "4C8G",
		"NodeVersion":   "v1.30.1",
		"DumplingImage": "repo/dumpling:v1",
		"SetResource":   "true",
		"CPURequest":    "2",
		"CPULimit":      "4",
		"MemoryLimit":   "7Gi",
		"MemoryRequest": "4Gi",
	})

	assert.Contains(t, rendered, "cpu: 2")
	assert.Contains(t, rendered, "cpu: 4")
	assert.Contains(t, rendered, "memory: 4Gi")
	assert.Contains(t, rendered, "memory: 7Gi")
}

func TestBuildDumplingCSVDialectArgs(t *testing.T) {
	tests := []struct {
		name    string
		dialect *apiv1.ExportOptions_ExportCSVDialect_Enum
		want    []string
		wantErr string
	}{
		{name: "omitted", want: nil},
		{
			name:    "default",
			dialect: dialectPtr(apiv1.ExportOptions_ExportCSVDialect_DEFAULT),
			want:    []string{"--csv-output-dialect="},
		},
		{
			name:    "snowflake",
			dialect: dialectPtr(apiv1.ExportOptions_ExportCSVDialect_SNOWFLAKE),
			want:    []string{"--csv-output-dialect=snowflake"},
		},
		{
			name:    "redshift",
			dialect: dialectPtr(apiv1.ExportOptions_ExportCSVDialect_REDSHIFT),
			want:    []string{"--csv-output-dialect=redshift"},
		},
		{
			name:    "bigquery",
			dialect: dialectPtr(apiv1.ExportOptions_ExportCSVDialect_BIGQUERY),
			want:    []string{"--csv-output-dialect=bigquery"},
		},
		{
			name:    "unknown",
			dialect: dialectPtr(apiv1.ExportOptions_ExportCSVDialect_Enum(99)),
			wantErr: "unsupported csv dialect",
		},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			got, err := buildDumplingCSVDialectArgs(test.dialect)
			if test.wantErr != "" {
				require.ErrorContains(t, err, test.wantErr)
				return
			}
			require.NoError(t, err)
			require.Equal(t, test.want, got)
		})
	}
}

func dialectPtr(dialect apiv1.ExportOptions_ExportCSVDialect_Enum) *apiv1.ExportOptions_ExportCSVDialect_Enum {
	return &dialect
}

// TestDumplingFileType pins the case Dumpling's --csv-output-dialect guard
// compares against. The guard runs during flag parsing and matches the literal
// "csv", while Dumpling only lower-cases --filetype in a later adjust step, so
// passing the enum name verbatim fails every export that also asks for a
// dialect - and only those.
func TestDumplingFileType(t *testing.T) {
	tests := []struct {
		name     string
		fileType apiv1.ExportOptions_FileType
		want     string
	}{
		{name: "csv", fileType: apiv1.ExportOptions_CSV, want: "csv"},
		{name: "sql", fileType: apiv1.ExportOptions_SQL, want: "sql"},
		{name: "parquet", fileType: apiv1.ExportOptions_PARQUET, want: "parquet"},
	}

	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			assert.Equal(t, test.want, dumplingFileType(test.fileType))
		})
	}
}

func TestBuildDumplingFilterArgsWithPartitions(t *testing.T) {
	args, err := buildDumplingFilterArgs(&apiv1.ExportOptions{
		Filter: &apiv1.ExportOptions_Filter{
			Type: &apiv1.ExportOptions_Filter_Table_{
				Table: &apiv1.ExportOptions_Filter_Table{
					Patterns:   []string{"sales.orders"},
					Partitions: []string{"p0", "p1"},
				},
			},
		},
	})
	require.NoError(t, err)
	assert.Equal(t, []string{
		"--filter=sales.orders",
		"--filter=" + defaultTableFilter,
		"--partitions=p0",
		"--partitions=p1",
	}, args)
}

func TestBuildDumplingEscapeBackslashArgs(t *testing.T) {
	assert.Nil(t, buildDumplingEscapeBackslashArgs(nil))
	assert.Equal(t, []string{"--escape-backslash=true"}, buildDumplingEscapeBackslashArgs(boolPtr(true)))
	assert.Equal(t, []string{"--escape-backslash=false"}, buildDumplingEscapeBackslashArgs(boolPtr(false)))
}

func TestBuildDumplingCompressionArgs(t *testing.T) {
	assert.Nil(t, buildDumplingCompressionArgs(nil))
	assert.Equal(t, []string{"--compress=gzip"}, buildDumplingCompressionArgs(&apiv1.ExportOptions{
		FileType:    apiv1.ExportOptions_CSV,
		Compression: apiv1.ExportOptions_GZIP,
	}))
	assert.Nil(t, buildDumplingCompressionArgs(&apiv1.ExportOptions{
		FileType:    apiv1.ExportOptions_PARQUET,
		Compression: apiv1.ExportOptions_GZIP,
	}))
}

func boolPtr(value bool) *bool {
	return &value
}

func TestToDumplingParquetCompression(t *testing.T) {
	tests := map[string]struct {
		compression apiv1.ExportParquetCompressionType_Enum
		want        string
	}{
		"unspecified": {compression: apiv1.ExportParquetCompressionType_PARQUET_COMPRESSION_TYPE_UNSPECIFIED, want: ""},
		"gzip":        {compression: apiv1.ExportParquetCompressionType_GZIP, want: "gzip"},
		"snappy":      {compression: apiv1.ExportParquetCompressionType_SNAPPY, want: "snappy"},
		"zstd":        {compression: apiv1.ExportParquetCompressionType_ZSTD, want: "zstd"},
		"none":        {compression: apiv1.ExportParquetCompressionType_NONE, want: "no-compression"},
	}

	for name, test := range tests {
		t.Run(name, func(t *testing.T) {
			assert.Equal(t, test.want, toDumplingParquetCompression(test.compression))
		})
	}
}