package utils

import (
	"fmt"
	"net/url"
	"testing"

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

func TestDesensitiseSinkUri(t *testing.T) {
	encodedPassword := url.QueryEscape("pingcap!@#")
	testcases := []struct {
		sinkUri        string
		expectedResult string
	}{
		{
			sinkUri:        "kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0",
			expectedResult: "kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0",
		},
		{
			sinkUri:        "kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0&sasl-user=test&sasl-password=test1234",
			expectedResult: "kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0&sasl-user=test&sasl-password=********",
		},
		{
			sinkUri:        fmt.Sprintf("kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0&sasl-user=test&sasl-password=%s", encodedPassword),
			expectedResult: "kafka://127.0.0.1:9092/topic-name?protocol=canal-json&kafka-version=2.4.0&sasl-user=test&sasl-password=********",
		},
		{
			sinkUri:        "mysql://root:test1234@127.0.0.1:4000?transaction-atomicity=none",
			expectedResult: "mysql://root:********@127.0.0.1:4000?transaction-atomicity=none",
		},
		{
			sinkUri:        fmt.Sprintf("mysql://root:%s@127.0.0.1:4000?transaction-atomicity=none", encodedPassword),
			expectedResult: "mysql://root:********@127.0.0.1:4000?transaction-atomicity=none",
		},
		{
			sinkUri:        "s3://bucket1?access-key=akakak&secret-access-key=sksksk&force-path-style=true",
			expectedResult: "s3://bucket1?access-key=********&secret-access-key=********&force-path-style=true",
		},
		{
			sinkUri:        fmt.Sprintf("s3://bucket1?access-key=akakak&secret-access-key=%s&force-path-style=true", encodedPassword),
			expectedResult: "s3://bucket1?access-key=********&secret-access-key=********&force-path-style=true",
		},
		{
			sinkUri:        "gs://bucket1?force-path-style=true",
			expectedResult: "gs://bucket1?force-path-style=true",
		},
	}

	for _, tc := range testcases {
		result := DesensitiseSinkUri(tc.sinkUri)
		assert.Equal(t, tc.expectedResult, result)
	}
}