package pimport

import (
	"context"
	"testing"

	"github.com/pingcap/tidb/br/pkg/storage"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/dataflow-service/idl/pbgen/proto/dataflow"
	"github.com/tidbcloud/dataflow-service/internal/db/entity"
)

const testFile = "test_data.csv"

func mockStorage(t *testing.T, data []byte) storage.ExternalStorage {
	baseDir := t.TempDir()
	s, err := storage.NewLocalStorage(baseDir)
	require.NoError(t, err)

	ctx := context.Background()
	require.NoError(t, s.WriteFile(ctx, testFile, data))
	return s
}

func TestLocalImportPreviewer_doSingleTablePreview(t *testing.T) {
	previewer := &LocalImportPreviewer{
		DataFormat: entity.ImportDataFormatCsv,
		CSVFormat: &dataflow.CustomCSVFormat{
			Separator: ",",
			Delimiter: "\"",
			Header:    true,
		},
	}

	store := mockStorage(t, []byte("column1,column2\nvalue1,value2\n"))

	fileMapping := &PreviewFileMapping{
		DatabaseName: "test_db",
		TableName:    "test_table",
		FileName:     testFile,
	}

	table, err := previewer.doSingleTablePreview(context.Background(), store, fileMapping, 5)
	require.NoError(t, err)

	assert.Equal(t, "test_db", table.SchemaName)
	assert.Equal(t, "test_table", table.TableName)
	require.Len(t, table.RowValues, 1)

	row := table.RowValues[0]
	require.Len(t, row.ColumnValues, 2)
	assert.Equal(t, "value1", row.ColumnValues[0])
	assert.Equal(t, "value2", row.ColumnValues[1])

	require.Len(t, table.RowDataTypes, 2)
	assert.Equal(t, "VARCHAR(512)", table.RowDataTypes[0])
	assert.Equal(t, "VARCHAR(512)", table.RowDataTypes[1])

	require.Len(t, table.RowHeaders, 2)
	assert.Equal(t, "column1", table.RowHeaders[0])
	assert.Equal(t, "column2", table.RowHeaders[1])

	errorCases := []struct {
		name           string
		mockCSVContent []byte
		expectedErr    error
	}{
		{
			name:           "header num does not equal to data type 1",
			mockCSVContent: []byte("column1,column2\nvalue1,value2,value3\n"),
			expectedErr: &ErrHeaderDataColumnMismatch{
				HeaderColumnsCount: 2,
				DataColumnsCount:   3,
			},
		},
		{
			name:           "header num does not equal to data type 2",
			mockCSVContent: []byte("column1,column2,column3\nvalue1,value2\n"),
			expectedErr: &ErrHeaderDataColumnMismatch{
				HeaderColumnsCount: 3,
				DataColumnsCount:   2,
			},
		},
		{
			name:           "row overflow",
			mockCSVContent: []byte("column1,column2\n1,2\n1,2,3\n"),
			expectedErr:    &ErrCSVRowOverflow{1},
		},
		{
			name:           "csv syntax error",
			mockCSVContent: []byte("column1,column2\n\"value1\", \"value2\"\n"),
			expectedErr:    &ErrCSVSyntax{},
		},
	}

	for _, tt := range errorCases {
		t.Run(tt.name, func(t *testing.T) {
			store := mockStorage(t, tt.mockCSVContent)

			_, err := previewer.doSingleTablePreview(context.Background(), store, fileMapping, 5)
			assert.Error(t, err)
			assert.Equal(t, tt.expectedErr, err)
		})
	}
}