package domain

import (
	"testing"

	"github.com/AlekSi/pointer"
	"github.com/stretchr/testify/assert"
)

func TestTiCDCChangefeedS3URIConfigData_GenerateURI(t *testing.T) {
	t.Run("ak sk without endpoint", func(t *testing.T) {
		cfg := &TiCDCChangefeedS3URIConfigData{
			S3URI:           "s3://howard-test-01/lake-test",
			AccessKey:       pointer.ToString("LTAI5tCMf5zsdLjEvet8DKgH"),
			SecretAccessKey: pointer.ToString("sYHXGNXIFKuhG4slIcuJqjOjPtCbPe"),
			FlushInterval:   "5m",
			FileSize:        1000,
		}
		uri := cfg.GenerateURI()
		assert.Equal(t,
			"s3://howard-test-01/lake-test?flush-interval=5m&file-size=1000"+
				"&access-key=LTAI5tCMf5zsdLjEvet8DKgH&secret-access-key=sYHXGNXIFKuhG4slIcuJqjOjPtCbPe",
			uri)
		assert.NotContains(t, uri, "endpoint=")
	})

	t.Run("ak sk with endpoint", func(t *testing.T) {
		cfg := &TiCDCChangefeedS3URIConfigData{
			S3URI:           "s3://howard-test-01/lake-test",
			AccessKey:       pointer.ToString("LTAI5tCMf5zsdLjEvet8DKgH"),
			SecretAccessKey: pointer.ToString("sYHXGNXIFKuhG4slIcuJqjOjPtCbPe"),
			FlushInterval:   "5m",
			FileSize:        1000,
			Endpoint:        pointer.ToString("https://s3.oss-us-east-1.aliyuncs.com"),
		}
		uri := cfg.GenerateURI()
		assert.Equal(t,
			"s3://howard-test-01/lake-test?flush-interval=5m&file-size=1000"+
				"&access-key=LTAI5tCMf5zsdLjEvet8DKgH&secret-access-key=sYHXGNXIFKuhG4slIcuJqjOjPtCbPe"+
				"&endpoint=https%3A%2F%2Fs3.oss-us-east-1.aliyuncs.com",
			uri)
	})

	t.Run("empty endpoint is not appended", func(t *testing.T) {
		cfg := &TiCDCChangefeedS3URIConfigData{
			S3URI:           "s3://bucket/key",
			AccessKey:       pointer.ToString("ak"),
			SecretAccessKey: pointer.ToString("sk"),
			FlushInterval:   "5m",
			FileSize:        1000,
			Endpoint:        pointer.ToString(""),
		}
		assert.NotContains(t, cfg.GenerateURI(), "endpoint=")
	})

	t.Run("alibaba oss endpoint and provider", func(t *testing.T) {
		cfg := &TiCDCChangefeedS3URIConfigData{
			S3URI:           "s3://howard-test-02/lake-test",
			AccessKey:       pointer.ToString("ak"),
			SecretAccessKey: pointer.ToString("sk"),
			FlushInterval:   "5m",
			FileSize:        1000,
			Endpoint:        pointer.ToString("https://oss-ap-southeast-1.aliyuncs.com"),
			Provider:        pointer.ToString("alibaba"),
		}
		uri := cfg.GenerateURI()
		assert.Equal(t,
			"s3://howard-test-02/lake-test?flush-interval=5m&file-size=1000"+
				"&access-key=ak&secret-access-key=sk"+
				"&endpoint=https%3A%2F%2Foss-ap-southeast-1.aliyuncs.com&provider=alibaba",
			uri)
	})

	t.Run("empty provider is not appended", func(t *testing.T) {
		cfg := &TiCDCChangefeedS3URIConfigData{
			S3URI:           "s3://bucket/key",
			AccessKey:       pointer.ToString("ak"),
			SecretAccessKey: pointer.ToString("sk"),
			FlushInterval:   "5m",
			FileSize:        1000,
			Provider:        pointer.ToString(""),
		}
		assert.NotContains(t, cfg.GenerateURI(), "provider=")
	})

	t.Run("iam role with endpoint", func(t *testing.T) {
		cfg := &TiCDCChangefeedS3URIConfigData{
			S3URI:         "s3://bucket/key",
			RoleARN:       pointer.ToString("arn:aws:iam::123:role/test"),
			ExternalID:    pointer.ToString("ext-1"),
			FlushInterval: "5m",
			FileSize:      1000,
			Endpoint:      pointer.ToString("https://s3.oss-us-east-1.aliyuncs.com"),
		}
		uri := cfg.GenerateURI()
		assert.Contains(t, uri, "&role-arn=arn:aws:iam::123:role/test&external-id=ext-1")
		assert.Contains(t, uri, "&endpoint=https%3A%2F%2Fs3.oss-us-east-1.aliyuncs.com")
	})
}

func TestTiCDCChangefeedKafkaURIConfigData_GenerateURI(t *testing.T) {
	baseCfg := func() *TiCDCChangefeedKafkaURIConfigData {
		return &TiCDCChangefeedKafkaURIConfigData{
			BrokerEndpoints:   "broker1:9092",
			DefaultTopic:      "cdc-events",
			Protocol:          "canal-json",
			EnableTLS:         false,
			ReplicationFactor: 3,
			PartitionNum:      6,
			KafkaVersion:      "3.0.0",
		}
	}

	t.Run("client id not set is not appended", func(t *testing.T) {
		cfg := baseCfg()
		assert.NotContains(t, cfg.GenerateURI(), "kafka-client-id=")
	})

	t.Run("empty client id pointer is not appended", func(t *testing.T) {
		cfg := baseCfg()
		cfg.KafkaClientID = pointer.ToString("")
		assert.NotContains(t, cfg.GenerateURI(), "kafka-client-id=")
	})

	t.Run("client id set is appended and escaped", func(t *testing.T) {
		cfg := baseCfg()
		cfg.KafkaClientID = pointer.ToString("my client/id")
		assert.Contains(t, cfg.GenerateURI(), "&kafka-client-id=my+client%2Fid")
	})
}