package console

import (
	"testing"

	"github.com/stretchr/testify/require"
	"github.com/stretchr/testify/suite"
	"github.com/tidbcloud/pkgv2/import/precheck"
	"github.com/tidbcloud/serverless-service/idl/pbgen/proto/console"
	"github.com/tidbcloud/serverless-service/internal/utils"
	globalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/global/serverless/v1"
	regionalserverlessv1 "github.com/tidbcloud/tidb-management-service/api/spec/regional/serverless/v1"
)

type GetCustomFileRulesTestSuite struct {
	suite.Suite
}

func (suite *GetCustomFileRulesTestSuite) TestGetCustomFileRules_EmptyInfos() {
	assert := require.New(suite.T())
	rules, err := getCustomFileRules([]*console.ImportTargetTableInfo{}, console.ImportFileType_CSV)
	assert.NoError(err)
	assert.Empty(rules)
}

func (suite *GetCustomFileRulesTestSuite) TestGetCustomFileRules_EmptyCustomFile() {
	assert := require.New(suite.T())
	infos := []*console.ImportTargetTableInfo{
		{
			CustomFile: "",
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table1",
			},
		},
	}
	rules, err := getCustomFileRules(infos, console.ImportFileType_CSV)
	assert.Error(err)
	assert.Nil(rules)
}

func (suite *GetCustomFileRulesTestSuite) TestGetCustomFileRules_UnsupportedType() {
	assert := require.New(suite.T())
	infos := []*console.ImportTargetTableInfo{
		{
			CustomFile: "test/*.csv",
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table1",
			},
		},
	}
	rules, err := getCustomFileRules(infos, console.ImportFileType_UNSPECIFIED)
	assert.Error(err)
	assert.Nil(rules)
}

func (suite *GetCustomFileRulesTestSuite) TestGetCustomFileRules_Success() {
	assert := require.New(suite.T())
	infos := []*console.ImportTargetTableInfo{
		{
			CustomFile: "test/*.csv",
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table1",
			},
		},
		{
			CustomFile: "data/*.csv",
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table2",
			},
		},
	}
	rules, err := getCustomFileRules(infos, console.ImportFileType_CSV)
	assert.NoError(err)
	assert.Equal([]*regionalserverlessv1.FileRouteRule{
		{
			Pattern: utils.GenCustomNamingRuleRegexp("test/*.csv"),
			Schema:  "test",
			Table:   "table1",
			Type:    precheck.TypeCSV,
		},
		{
			Pattern: utils.GenCustomNamingRuleRegexp("data/*.csv"),
			Schema:  "test",
			Table:   "table2",
			Type:    precheck.TypeCSV,
		},
	}, rules)
}

func (suite *GetCustomFileRulesTestSuite) TestGetCustomFileRules_EscapeDollarInTableName() {
	assert := require.New(suite.T())
	infos := []*console.ImportTargetTableInfo{
		{
			CustomFile: "product_categories.csv",
			TargetTable: &console.Table{
				Database: "$hyper-prod",
				Table:    "$redirect",
			},
		},
	}

	rules, err := getCustomFileRules(infos, console.ImportFileType_CSV)
	assert.NoError(err)
	assert.Equal([]*regionalserverlessv1.FileRouteRule{
		{
			Pattern: utils.GenCustomNamingRuleRegexp("product_categories.csv"),
			Schema:  "$$hyper-prod",
			Table:   "$$redirect",
			Type:    precheck.TypeCSV,
		},
	}, rules)
}

type ConvertToPrecheckRespTestSuite struct {
	suite.Suite
}

func (suite *ConvertToPrecheckRespTestSuite) TestConvertToPrecheckResp_EmptyResults() {
	assert := require.New(suite.T())
	resp := convertToPrecheckResp(&globalserverlessv1.PrecheckResponse{})
	assert.NotNil(resp)
	assert.Empty(resp.TargetTables)
	assert.Equal(uint64(0), resp.TotalTablesCount)
	assert.Equal(uint64(0), resp.TotalDataFilesCount)
	assert.Empty(resp.ErrorMessage)
}

func (suite *ConvertToPrecheckRespTestSuite) TestConvertToPrecheckResp_NoWarnings() {
	assert := require.New(suite.T())
	result := &globalserverlessv1.PrecheckResponse{
		TableResults: []*regionalserverlessv1.TableResult{
			{
				Table:         &regionalserverlessv1.Table{Schema: "test", Table: "table1"},
				DataFiles:     []string{"test/table1.csv"},
				SchemaFile:    "test/schema1.sql",
				UseSchemaFile: true,
			},
			{
				Table:     &regionalserverlessv1.Table{Schema: "test", Table: "table2"},
				DataFiles: []string{"test/table2.csv", "test/table2_part2.csv"},
			},
		},
	}
	resp := convertToPrecheckResp(result)
	assert.NotNil(resp)
	assert.Equal(uint64(2), resp.TotalTablesCount)
	assert.Equal(uint64(3), resp.TotalDataFilesCount)
	assert.Empty(resp.ErrorMessage)
	assert.Equal([]*console.TableResult{
		{
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table1",
			},
			MatchedDataFiles:  []string{"test/table1.csv"},
			MatchedSchemaFile: "test/schema1.sql",
			UseSchemaFile:     true,
		},
		{
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table2",
			},
			MatchedDataFiles: []string{"test/table2.csv", "test/table2_part2.csv"},
		},
	}, resp.TargetTables)
}

func (suite *ConvertToPrecheckRespTestSuite) TestConvertToPrecheckResp_WithWarnings() {
	assert := require.New(suite.T())
	result := &globalserverlessv1.PrecheckResponse{
		TableResults: []*regionalserverlessv1.TableResult{
			{
				Table:     &regionalserverlessv1.Table{Schema: "test", Table: "table1"},
				DataFiles: []string{"test/table1.csv"},
				ErrorMsg:  "table not empty",
			},
			{
				Table:     &regionalserverlessv1.Table{Schema: "test", Table: "table2"},
				DataFiles: []string{"test/table2.csv"},
				ErrorMsg:  "schema not found",
			},
		},
		Status: regionalserverlessv1.ImportPrecheckStatus_TRUNCATED,
	}
	resp := convertToPrecheckResp(result)
	assert.NotNil(resp)
	assert.Equal(uint64(2), resp.TotalTablesCount)
	assert.Equal(uint64(2), resp.TotalDataFilesCount)
	assert.Equal("Found 2 table(s) with error: test.table1, test.table2", resp.ErrorMessage)
	assert.Equal(regionalserverlessv1.ImportPrecheckStatus_TRUNCATED, result.Status)
	assert.Equal([]*console.TableResult{
		{
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table1",
			},
			MatchedDataFiles: []string{"test/table1.csv"},
			ErrorMessage:     "table not empty",
		},
		{
			TargetTable: &console.Table{
				Database: "test",
				Table:    "table2",
			},
			MatchedDataFiles: []string{"test/table2.csv"},
			ErrorMessage:     "schema not found",
		},
	}, resp.TargetTables)
}

func TestGetCustomFileRulesTestSuite(t *testing.T) {
	suite.Run(t, new(GetCustomFileRulesTestSuite))
}

func TestConvertToPrecheckRespTestSuite(t *testing.T) {
	suite.Run(t, new(ConvertToPrecheckRespTestSuite))
}