package changefeed

import (
	"testing"

	"github.com/stretchr/testify/require"

	. "github.com/tidbcloud/dataflow-service-e2e/internal/model"
	"github.com/tidbcloud/dataflow-service-e2e/internal/tidbcloud/dedicated/consoleapi/cdc/model"
)

func TestMySQLPrivateLinkCaseName(t *testing.T) {
	testCases := []struct {
		name          string
		cloudProvider CloudProviderType
		expected      string
	}{
		{
			name:          "aws",
			cloudProvider: AWS,
			expected:      "awsMySQLPrivateLink",
		},
		{
			name:          "gcp",
			cloudProvider: GCP,
			expected:      "gcpMySQLPrivateLink",
		},
		{
			name:          "azure",
			cloudProvider: AZURE,
			expected:      "azureMySQLPrivateLink",
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			c := &mysqlPrivateLinkCase{cloudProvider: tc.cloudProvider}
			require.Equal(t, tc.expected, c.Name())
		})
	}
}

func TestGenCreateMySQLPrivateLinkSpec_UsesRequestedPayload(t *testing.T) {
	c := &mysqlPrivateLinkCase{cloudProvider: AWS}
	downstream := &MySQLEntity{
		MySQLDimensions: MySQLDimensions{
			NetworkType:   DataflowNetworkTypeNetworkTypePrivateLink,
			CloudProvider: AWS,
		},
		User:     "admin",
		Password: "dataflow.pingcap!!!",
		Ports:    []int64{3306},
	}

	spec := c.genCreateMySQLPrivateLinkSpec("awsMySQLPrivateLink", "2rcu", downstream, "sink-74")

	require.NotNil(t, spec.ConnectorType)
	require.Equal(t, model.DataflowConnectorTypeMYSQL, *spec.ConnectorType)
	require.Equal(t, "awsMySQLPrivateLink", spec.Name)
	require.Equal(t, "2rcu", spec.ChangefeedInstanceType)

	require.NotNil(t, spec.MysqlInfo)
	require.NotNil(t, spec.MysqlInfo.Connection)
	require.Equal(t, "admin", spec.MysqlInfo.Connection.Username)
	require.Equal(t, "dataflow.pingcap!!!", spec.MysqlInfo.Connection.Password)

	require.NotNil(t, spec.MysqlInfo.TableConfig)
	require.Equal(t, []string{"awsMySQLPrivateLink.*"}, spec.MysqlInfo.TableConfig.FilterRule)
	require.Equal(t, []*model.DataflowEventFilterRule{}, spec.MysqlInfo.TableConfig.EventFilterRule)
	require.False(t, spec.MysqlInfo.TableConfig.CaseSensitive)
	require.NotNil(t, spec.MysqlInfo.TableConfig.Mode)
	require.Equal(t, model.DataflowTableModeFORCESYNC, *spec.MysqlInfo.TableConfig.Mode)

	require.NotNil(t, spec.MysqlInfo.StartPosition)
	require.NotNil(t, spec.MysqlInfo.StartPosition.Mode)
	require.Equal(t, model.DataflowStartModeFROMNOW, *spec.MysqlInfo.StartPosition.Mode)

	require.NotNil(t, spec.NetworkInfo)
	require.NotNil(t, spec.NetworkInfo.NetworkType)
	require.Equal(t, model.DataflowNetworkTypeNetworkTypePrivateLink, *spec.NetworkInfo.NetworkType)
	require.Equal(t, []int64{3306}, spec.NetworkInfo.Ports)
	require.Equal(t, "sink-74", spec.NetworkInfo.SinkEndpointID)
}