package central

import (
	"fmt"
	"testing"

	"github.com/segmentio/kafka-go"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
)

func Test_makeTlsSaslCfg(t *testing.T) {
	m := &connectionModel{EnableTls: true}
	tls, _, err := makeTlsSaslCfg(m)
	require.Nil(t, err)
	require.True(t, tls.InsecureSkipVerify)
}

/*
The following test cases are integration test cases, can not be run without kafka cluster setup.
*/
/*
func TestCheckKafkaAccessV2_Reachability_Happy(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:9092"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error when bootstrap endpoint is reachable and all Kafka brokers are reachable")
}

// Tricks: setup a kafka cluster with two listeners: INTERNAL & EXTERNAL, let one of broker advertise its EXTERNAL listener to an invalid address
func TestCheckKafkaAccessV2_Reachability_NotAllBrokerCanBeConnected(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:39092"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.Error(t, err, "Expected no error when bootstrap endpoint is reachable and one of Kafka brokers can not be reachable")
}

func TestCheckKafkaAccessV2_Reachability_Invalid(t *testing.T) {
	model := &connectionModel{Url: "invalid:9092"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.Error(t, err, "Expected error for invalid bootstrap endpoint")
}

func TestCheckKafkaAccessV2_Reachability_Mixed(t *testing.T) {
	model := &connectionModel{Url: "invalid:9092,127.0.0.1:9092"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error when one of bootstrap endpoints is reachable and all Kafka brokers are reachable")
}

// How Kafka TLS works? Refer to
// 1. https://developer.confluent.io/courses/security/authentication-ssl-and-sasl-ssl
// 2. https://kafka.apache.org/documentation/#security
// TODO: currently, ticdc will verify certificate of kafka, didn't find the way to setup kafka with official CA
// We did trigger connection to a kafka cluster with self-assigned certificate, verification will failed, since this kafka lib enable verification by default
func TestCheckKafkaAccessV2_Reachability_EnableTLS(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:9092", EnableTls: true}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error when bootstrap endpoint is reachable and all Kafka brokers are reachable")
}

func TestCheckKafkaAccessV2_Reachability_SASL_plain(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:9092", AuthType: "sasl-plain", User: "admin", Password: "admin-secret"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error when bootstrap endpoint is reachable and all Kafka brokers are reachable")
}

func TestCheckKafkaAccessV2_Reachability_SASL_sha256(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:9092", AuthType: "sasl-scram-sha-256", User: "admin", Password: "123456"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error when bootstrap endpoint is reachable and all Kafka brokers are reachable")
}

func TestCheckKafkaAccessV2_Reachability_SASL_sha512(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:9092", AuthType: "sasl-scram-sha-512", User: "admin", Password: "123456"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error when bootstrap endpoint is reachable and all Kafka brokers are reachable")
}

func TestCheckKafkaAccessV2_Reachability_SASL_unknown(t *testing.T) {
	model := &connectionModel{Url: "127.0.0.1:9092", AuthType: "unknown", User: "admin", Password: "123456"}
	err := checkKafkaAccessV2(context.Background(), model)
	assert.Error(t, err, "Expected error when it's unknown auth type")
}

func TestCheckKafkaAccessV2_ReplicationFactor(t *testing.T) {
	model := &connectionModel{}
	model.KafkaItem.ReplicationFactor = 2
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error for valid replication factor")

}
func TestCheckKafkaAccessV2_SchemaRegistryEndpoints(t *testing.T) {
	schemaRegistry := "127.0.0.1:8081"
	model := &connectionModel{}

	model.KafkaItem.SchemaRegistryEndpoints = schemaRegistry
	err := checkKafkaAccessV2(context.Background(), model)
	assert.NoError(t, err, "Expected no error for valid schema registry endpoint")
}
*/

func Test_doValidateReplicationFactor(t *testing.T) {
	// Test case 1: Replication factor is valid.
	t.Run("Valid replication factor", func(t *testing.T) {
		// Mock DescribeConfigsResponse
		resp := &kafka.DescribeConfigsResponse{
			Resources: []kafka.DescribeConfigResponseResource{
				{
					ResourceName: "test-resource",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "some.other.config",
							ConfigValue: "value",
						},
						{
							ConfigName:  "min.insync.replicas",
							ConfigValue: "2",
						},
					},
				},
				{
					ResourceName: "test-resource",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "min.insync.replicas",
							ConfigValue: "2",
						},
					},
				},
			},
		}

		// Mock the connectionModel
		mockModel := &connectionModel{}
		mockModel.KafkaItem.ReplicationFactor = 3

		// Call the function
		err := doValidateReplicationFactor(resp, mockModel, 5)

		// Assert no error
		assert.NoError(t, err)
	})

	// Test case 2: Replication factor is too small.
	t.Run("Replication factor too small", func(t *testing.T) {
		resp := &kafka.DescribeConfigsResponse{
			Resources: []kafka.DescribeConfigResponseResource{
				{
					ResourceName: "test-resource1",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "min.insync.replicas",
							ConfigValue: "1", // valid
						},
					},
				},
				{
					ResourceName: "test-resource2",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "min.insync.replicas",
							ConfigValue: "2", // invalid
						},
					},
				},
			},
		}

		mockModel := &connectionModel{}
		mockModel.KafkaItem.ReplicationFactor = 1

		// Call the function
		err := doValidateReplicationFactor(resp, mockModel, 5)

		// Assert the expected error message
		assert.Error(t, err)
		assert.Equal(t, fmt.Sprintf("ReplicationFactor(%d) should >= %s(%d), and <= broker number(%d)", 1, "min.insync.replicas", 2, 5), err.Error())
	})

	// Test case 3: Replication factor is too large.
	t.Run("Replication factor too large", func(t *testing.T) {
		resp := &kafka.DescribeConfigsResponse{
			Resources: []kafka.DescribeConfigResponseResource{
				{
					ResourceName: "test-resource",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "min.insync.replicas",
							ConfigValue: "2",
						},
					},
				},
			},
		}

		mockModel := &connectionModel{}
		mockModel.KafkaItem.ReplicationFactor = 6

		// Call the function
		err := doValidateReplicationFactor(resp, mockModel, 5)

		// Assert the expected error message
		assert.Error(t, err)
		assert.Equal(t, fmt.Sprintf("ReplicationFactor(%d) should >= %s(%d), and <= broker number(%d)", 6, "min.insync.replicas", 2, 5), err.Error())
	})

	// Test case 4: No min.insync.replicas config found.
	t.Run("No min.insync.replicas config", func(t *testing.T) {
		resp := &kafka.DescribeConfigsResponse{
			Resources: []kafka.DescribeConfigResponseResource{
				{
					ResourceName: "test-resource",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "some.other.config",
							ConfigValue: "value",
						},
					},
				},
			},
		}

		mockModel := &connectionModel{}
		mockModel.KafkaItem.ReplicationFactor = 3

		// Call the function
		err := doValidateReplicationFactor(resp, mockModel, 5)

		// Assert no error, as there's no min.insync.replicas config
		assert.NoError(t, err)
	})

	// Test case 5: Invalid min.insync.replicas value (non-integer).
	t.Run("Invalid min.insync.replicas value", func(t *testing.T) {
		resp := &kafka.DescribeConfigsResponse{
			Resources: []kafka.DescribeConfigResponseResource{
				{
					ResourceName: "test-resource",
					ConfigEntries: []kafka.DescribeConfigResponseConfigEntry{
						{
							ConfigName:  "min.insync.replicas",
							ConfigValue: "invalid", // Non-integer value
						},
					},
				},
			},
		}

		mockModel := &connectionModel{}
		mockModel.KafkaItem.ReplicationFactor = 3
		// Call the function
		err := doValidateReplicationFactor(resp, mockModel, 5)

		// Assert the error related to invalid integer conversion
		assert.Error(t, err)
		assert.Contains(t, err.Error(), "atoi min.insync.replicas config failed")
	})
}