package service

import (
	"testing"

	"github.com/stretchr/testify/assert"
	regionalpb "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/ticdc-service/v1"
)

func TestMSKAuthTypeRoundTrip(t *testing.T) {
	cases := []apiv1.TiCDCSinkEndpoint_MskAuthType{
		apiv1.TiCDCSinkEndpoint_SASL_SCRAM,
		apiv1.TiCDCSinkEndpoint_SASL_IAM,
		apiv1.TiCDCSinkEndpoint_MSK_AUTH_TYPE_UNSPECIFIED,
	}
	for _, in := range cases {
		got := mskAuthTypeFromRegional(mskAuthTypeToRegional(in))
		assert.Equal(t, in, got, "round trip for %s", in)
	}
	assert.Equal(t, regionalpb.PrivateLinkEndpoint_SASL_SCRAM,
		mskAuthTypeToRegional(apiv1.TiCDCSinkEndpoint_SASL_SCRAM))
}

func TestAssembleApiEndpoint_AmazonMSK(t *testing.T) {
	endpoint := &regionalpb.PrivateLinkEndpoint{
		Id:     "ep-msk-1",
		Status: regionalpb.PrivateLinkEndpoint_AVAILABLE,
		Info: &regionalpb.PrivateLinkEndpoint_AwsMsk_{
			AwsMsk: &regionalpb.PrivateLinkEndpoint_AwsMsk{
				MskArn:          "arn:msk",
				AuthType:        regionalpb.PrivateLinkEndpoint_SASL_SCRAM,
				BootstrapBroker: "b-1.demo:9196,b-2.demo:9196",
			},
		},
	}

	got := assembleApiEndpoint("acc-1", "sp-1", endpoint)

	assert.Equal(t, apiv1.TiCDCSinkEndpoint_AMAZON_MSK, got.EndpointType)
	assert.Equal(t, "arn:msk", got.MskArn)
	assert.Equal(t, apiv1.TiCDCSinkEndpoint_SASL_SCRAM, got.AuthType)
	assert.Equal(t, "b-1.demo:9196,b-2.demo:9196", got.BootstrapBroker)
	assert.Equal(t, apiv1.TiCDCSinkEndpoint_AVAILABLE, got.Phase)
	assert.Empty(t, got.ServiceName, "msk endpoint must not surface a service name")
}