package tioperator

import (
	"context"
	"fmt"
	"testing"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func TestListTiDBGroups(t *testing.T) {
	clusterId := "alpha"
	namespace := dataplaneNamespace(clusterId)

	tidbId := testComponentId("tidb-alpha")
	workerId := testComponentId("worker-alpha")

	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
	workerGroup := newTiDBGroup(t, namespace, workerId, "worker", "worker-tidb", 1)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup, workerGroup).Build()

	ctx := context.Background()
	components, err := ListTiDBGroups(ctx, cli, clusterId)
	require.NoError(t, err)
	require.Len(t, components, 1)
	assert.Equal(t, apiv1.ComponentType_TIDB, components[0].GetComponentType())
	assert.Equal(t, tidbId, components[0].GetId())
}

func TestGetTiDBGroup(t *testing.T) {
	clusterId := "beta"
	namespace := dataplaneNamespace(clusterId)

	tidbId := testComponentId("tidb-beta")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

	ctx := context.Background()
	groupId := groupIdWithComponent(tidbId, "tidb")
	component, err := GetTiDBGroup(ctx, cli, clusterId, groupId)
	require.NoError(t, err)
	require.NotNil(t, component)
	assert.Equal(t, apiv1.ComponentType_TIDB, component.GetComponentType())
	assert.Equal(t, tidbId, component.GetId())
	assert.Equal(t, uint32(3), component.GetNodeQuota().GetNodeCount())
}

func TestUpdateTiDBGroup(t *testing.T) {
	clusterId := "gamma"
	namespace := dataplaneNamespace(clusterId)

	tidbId := testComponentId("tidb-gamma")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

	ctx := context.Background()
	options := []UpdateOption{{Option: OpsUpdateNodeCount, Value: uint32(3)}}
	groupId := groupIdWithComponent(tidbId, "tidb")
	updated, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)
	require.NoError(t, err)
	require.NotNil(t, updated)
	assert.Equal(t, uint32(3), updated.GetNodeQuota().GetNodeCount())

	stored := &v1alpha1.TiDBGroup{}
	err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
	require.NoError(t, err)
	require.NotNil(t, stored.Spec.Replicas)
	assert.Equal(t, int32(3), *stored.Spec.Replicas)
}

func TestGetUpdateExpectedTiDBGroup(t *testing.T) {
	clusterId := "tidb-update"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-update")
	base := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

	// Add GOMEMLIMIT env to the base group
	base.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Env = []corev1.EnvVar{
		{
			Name:  "GOMEMLIMIT",
			Value: "13107MiB", // 80% of 16Gi (16384MiB)
		},
	}

	// Add NodeSelector for idempotent test
	base.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector = map[string]string{
		schemav1.K8sVersionLabelKey: "1.27",
	}

	nodeSpec := &entity.NodeSpec{Cpu: 8, MemoryGi: 32, AvailableCpu: 8, AvailableMemoryMi: 32768}

	cases := []struct {
		name            string
		mutate          func(*v1alpha1.TiDBGroup)
		options         []UpdateOption
		wantUpdate      bool
		wantErrContains string
		wantErrCode     codes.Code
		verify          func(*testing.T, *v1alpha1.TiDBGroup)
	}{
		{
			name: "multi-option-update",
			options: []UpdateOption{
				{Option: OpsUpdateNodeCount, Value: uint32(4)},
				{Option: OpsUpdateNodeSpec, Value: nodeSpec},
				{Option: OpsUpdateMaxCount, Value: uint32(8)},
				{Option: OpsUpdateMinCount, Value: uint32(2)},
				{Option: OpsUpdateVersion, Value: "v2.0.0"},
				{Option: OpsUpdateImage, Value: "repo/tidb:new"},
				{Option: OpsUpdateConfig, Value: "tidb-config"},
				{Option: OpsUpdateUpdateStrategy, Value: v1alpha1.ConfigUpdateStrategyHotReload},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				require.NotNil(t, group.Spec.Replicas)
				assert.Equal(t, int32(4), *group.Spec.Replicas)

				var container *corev1.Container
				for i := range group.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
						container = &group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
						break
					}
				}
				require.NotNil(t, container)

				assertMainContainerResources(t, container.Resources, nodeSpec, apiv1.ComponentType_TIDB)

				// Verify GOMEMLIMIT env is updated to 80% of new memory (32768MiB * 0.8 = 26214MiB)
				var goMemLimit string
				for _, env := range container.Env {
					if env.Name == "GOMEMLIMIT" {
						goMemLimit = env.Value
						break
					}
				}
				expectedGoMemLimit := fmt.Sprintf("%dMiB", uint64(float64(nodeSpec.AvailableMemoryMi)*GoMemlimitRate))
				assert.Equal(t, expectedGoMemLimit, goMemLimit, "GOMEMLIMIT should be 80%% of available memory")

				assert.Equal(t, "v2.0.0", group.Spec.Template.Spec.Version)
				require.NotNil(t, group.Spec.Template.Spec.Image)
				assert.Equal(t, "repo/tidb:new", *group.Spec.Template.Spec.Image)
				assert.Equal(t, v1alpha1.ConfigFile("tidb-config"), group.Spec.Template.Spec.Config)
				assert.Equal(t, v1alpha1.ConfigUpdateStrategyHotReload, group.Spec.Template.Spec.UpdateStrategy.Config)

				assert.Equal(t, "8", group.Annotations[schemav1.AnnotationsComponentMaxCount])
				assert.Equal(t, "2", group.Annotations[schemav1.AnnotationsComponentMinCount])
				assert.Equal(t, "8C32G", group.Labels[schemav1.LabelNodeSpecTypeKey])
			},
		},
		{
			name:       "update-node-spec-only-updates-gomemlimit",
			options:    []UpdateOption{{Option: OpsUpdateNodeSpec, Value: nodeSpec}},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				var container *corev1.Container
				for i := range group.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
						container = &group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
						break
					}
				}
				require.NotNil(t, container)

				// Verify GOMEMLIMIT env is updated
				var goMemLimit string
				for _, env := range container.Env {
					if env.Name == "GOMEMLIMIT" {
						goMemLimit = env.Value
						break
					}
				}
				expectedGoMemLimit := fmt.Sprintf("%dMiB", uint64(float64(nodeSpec.AvailableMemoryMi)*GoMemlimitRate))
				assert.Equal(t, expectedGoMemLimit, goMemLimit, "GOMEMLIMIT should be updated when node spec changes")
			},
		},
		{
			name: "update-node-spec-with-small-memory-keeps-nonzero-request",
			options: []UpdateOption{
				{
					Option: OpsUpdateNodeSpec,
					Value: &entity.NodeSpec{
						Cpu:               2,
						MemoryGi:          4,
						AvailableCpu:      1,
						AvailableMemoryMi: 2039,
					},
				},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				var container *corev1.Container
				for i := range group.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
						container = &group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
						break
					}
				}
				require.NotNil(t, container)

				assertMainContainerResources(t, container.Resources, &entity.NodeSpec{
					Cpu:               2,
					MemoryGi:          4,
					AvailableCpu:      1,
					AvailableMemoryMi: 2039,
				}, apiv1.ComponentType_TIDB)
			},
		},
		{
			name: "update-node-spec-with-different-memory-sizes",
			options: []UpdateOption{
				{Option: OpsUpdateNodeSpec, Value: &entity.NodeSpec{Cpu: 4, MemoryGi: 16, AvailableCpu: 4, AvailableMemoryMi: 16384}},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				var container *corev1.Container
				for i := range group.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
						container = &group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
						break
					}
				}
				require.NotNil(t, container)

				// Verify GOMEMLIMIT for 16Gi memory (16384MiB * 0.8 = 13107MiB)
				var goMemLimit string
				for _, env := range container.Env {
					if env.Name == "GOMEMLIMIT" {
						goMemLimit = env.Value
						break
					}
				}
				expectedGoMemLimit := "13107MiB"
				assert.Equal(t, expectedGoMemLimit, goMemLimit)
			},
		},
		{
			name: "update-node-spec-with-large-memory",
			options: []UpdateOption{
				{Option: OpsUpdateNodeSpec, Value: &entity.NodeSpec{Cpu: 16, MemoryGi: 64, AvailableCpu: 16, AvailableMemoryMi: 65536}},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				var container *corev1.Container
				for i := range group.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
						container = &group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
						break
					}
				}
				require.NotNil(t, container)

				// Verify GOMEMLIMIT for 64Gi memory (65536MiB * 0.8 = 52428MiB)
				var goMemLimit string
				for _, env := range container.Env {
					if env.Name == "GOMEMLIMIT" {
						goMemLimit = env.Value
						break
					}
				}
				expectedGoMemLimit := "52428MiB"
				assert.Equal(t, expectedGoMemLimit, goMemLimit)
			},
		},
		{
			name: "update-node-spec-gomemlimit-follows-max-memory",
			options: []UpdateOption{
				{
					Option: OpsUpdateNodeSpec,
					Value: &entity.NodeSpec{
						Cpu:               8,
						MemoryGi:          32,
						AvailableCpu:      8,
						AvailableMemoryMi: 32768,
						MaxMemoryMi:       40000,
					},
				},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				var container *corev1.Container
				for i := range group.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
						container = &group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
						break
					}
				}
				require.NotNil(t, container)
				expectedLimitMemory := resource.NewQuantity(int64(40000)*1024*1024, resource.BinarySI)
				assert.Equal(t, expectedLimitMemory.String(), container.Resources.Limits.Memory().String())
				var goMemLimit string
				for _, env := range container.Env {
					if env.Name == "GOMEMLIMIT" {
						goMemLimit = env.Value
						break
					}
				}
				assert.Equal(t, fmt.Sprintf("%dMiB", uint64(float64(40000)*GoMemlimitRate)), goMemLimit)
			},
		},
		{
			name:            "invalid-node-count-type",
			options:         []UpdateOption{{Option: OpsUpdateNodeCount, Value: "bad"}},
			wantErrContains: "invalid node count",
		},
		{
			name:        "node-count-exceeds-max",
			options:     []UpdateOption{{Option: OpsUpdateNodeCount, Value: uint32(10)}},
			wantErrCode: codes.InvalidArgument,
		},
		{
			name:            "invalid-node-spec-type",
			options:         []UpdateOption{{Option: OpsUpdateNodeSpec, Value: "bad"}},
			wantErrContains: "invalid node spec",
		},
		{
			name:            "invalid-config-type",
			options:         []UpdateOption{{Option: OpsUpdateConfig, Value: 123}},
			wantErrContains: "invalid config",
		},
		{
			name:            "invalid-update-strategy-type",
			options:         []UpdateOption{{Option: OpsUpdateUpdateStrategy, Value: map[string]string{"bad": "value"}}},
			wantErrContains: "invalid update strategy",
		},
		{
			name: "update-audit-log-config",
			options: []UpdateOption{
				{Option: OpsUpdateTiDBAuditLog, Value: `{"enabled": true, "log_path": "/var/log/tidb/audit.log"}`},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				require.NotNil(t, group.Spec.Template.Spec.Overlay)
				require.NotNil(t, group.Spec.Template.Spec.Overlay.Pod)
				require.NotNil(t, group.Spec.Template.Spec.Overlay.Pod.Annotations)

				auditLogConfig, exists := group.Spec.Template.Spec.Overlay.Pod.Annotations["tidb.auditlog.config"]
				assert.True(t, exists, "audit log config annotation should exist")
				assert.JSONEq(t, `{"enabled": true, "log_path": "/var/log/tidb/audit.log"}`, auditLogConfig)
			},
		},
		{
			name: "update-audit-log-config-with-multiple-options",
			options: []UpdateOption{
				{Option: OpsUpdateNodeCount, Value: uint32(5)},
				{Option: OpsUpdateTiDBAuditLog, Value: `{"enabled": true, "retention_days": 30}`},
				{Option: OpsUpdateVersion, Value: "v2.1.0"},
			},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				require.NotNil(t, group.Spec.Replicas)
				assert.Equal(t, int32(5), *group.Spec.Replicas)
				assert.Equal(t, "v2.1.0", group.Spec.Template.Spec.Version)

				require.NotNil(t, group.Spec.Template.Spec.Overlay.Pod.Annotations)
				auditLogConfig := group.Spec.Template.Spec.Overlay.Pod.Annotations["tidb.auditlog.config"]
				assert.Equal(t, `{"enabled": true, "retention_days": 30}`, auditLogConfig)
			},
		},
		{
			name:            "invalid-audit-log-config-type",
			options:         []UpdateOption{{Option: OpsUpdateTiDBAuditLog, Value: 123}},
			wantErrContains: "invalid audit log config",
		},
		{
			name:       "empty-audit-log-config-no-update",
			options:    []UpdateOption{{Option: OpsUpdateTiDBAuditLog, Value: ""}},
			wantUpdate: false,
		},
		{
			name:       "no-op",
			options:    []UpdateOption{{Option: OpsUpdateNodeCount, Value: uint32(0)}},
			wantUpdate: false,
		},
		// OpsUpdateNodeVersion tests
		{
			name:       "update-node-version",
			options:    []UpdateOption{{Option: OpsUpdateNodeVersion, Value: "1.28"}},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				require.NotNil(t, group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector)
				assert.Equal(t, "1.28", group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector[schemav1.K8sVersionLabelKey])
			},
		},
		{
			name:       "update-node-version-idempotent",
			options:    []UpdateOption{{Option: OpsUpdateNodeVersion, Value: "1.27"}},
			wantUpdate: false,
		},
		{
			name: "update-node-version-nil-nodeSelector",
			mutate: func(group *v1alpha1.TiDBGroup) {
				group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector = nil
			},
			options:    []UpdateOption{{Option: OpsUpdateNodeVersion, Value: "1.29"}},
			wantUpdate: true,
			verify: func(t *testing.T, group *v1alpha1.TiDBGroup) {
				t.Helper()
				require.NotNil(t, group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector)
				assert.Equal(t, "1.29", group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector[schemav1.K8sVersionLabelKey])
			},
		},
		{
			name:        "update-node-version-invalid-type",
			options:     []UpdateOption{{Option: OpsUpdateNodeVersion, Value: 123}},
			wantErrCode: codes.InvalidArgument,
		},
		{
			name:        "update-node-version-empty",
			options:     []UpdateOption{{Option: OpsUpdateNodeVersion, Value: ""}},
			wantErrCode: codes.InvalidArgument,
		},
	}

	ctx := context.Background()

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			tidb := base.DeepCopy()
			if tc.mutate != nil {
				tc.mutate(tidb)
			}

			needUpdate, err := getUpdateExpectedTiDBGroup(ctx, tidb, apiv1.ComponentType_TIDB, tc.options...)

			if tc.wantErrContains != "" {
				require.Error(t, err)
				assert.Contains(t, err.Error(), tc.wantErrContains)
				return
			}

			if tc.wantErrCode != codes.OK {
				require.Error(t, err)
				st, ok := status.FromError(err)
				require.True(t, ok)
				assert.Equal(t, tc.wantErrCode, st.Code())
				return
			}

			require.NoError(t, err)
			assert.Equal(t, tc.wantUpdate, needUpdate)
			if tc.verify != nil {
				tc.verify(t, tidb)
			}
		})
	}
}

func TestGetUpdateExpectedTiDBWorkerRequestMemory(t *testing.T) {
	clusterId := "tidb-worker-request"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-worker-request")
	tidb := newTiDBGroup(t, namespace, tidbId, "worker", "worker-tidb", 1)
	nodeSpec := &entity.NodeSpec{Cpu: 2, MemoryGi: 4, AvailableCpu: 1, AvailableMemoryMi: 2039}

	needUpdate, err := getUpdateExpectedTiDBGroup(
		context.Background(),
		tidb,
		apiv1.ComponentType_TIDB_WORKER,
		UpdateOption{Option: OpsUpdateNodeSpec, Value: nodeSpec},
	)
	require.NoError(t, err)
	assert.True(t, needUpdate)

	var container *corev1.Container
	for i := range tidb.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
		if tidb.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
			container = &tidb.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
			break
		}
	}
	require.NotNil(t, container)
	assertMainContainerResources(t, container.Resources, nodeSpec, apiv1.ComponentType_TIDB_WORKER)
}

func TestListTiDBWorkers(t *testing.T) {
	clusterId := "delta"
	namespace := dataplaneNamespace(clusterId)

	tidbId := testComponentId("tidb-delta")
	workerId := testComponentId("worker-delta")

	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
	workerGroup := newTiDBGroup(t, namespace, workerId, "worker", "worker-tidb", 1)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup, workerGroup).Build()

	ctx := context.Background()
	workers, err := ListTiDBWorkers(ctx, cli, clusterId)
	require.NoError(t, err)
	require.Len(t, workers, 1)
	assert.Equal(t, apiv1.ComponentType_TIDB_WORKER, workers[0].GetComponentType())
	assert.Equal(t, workerId, workers[0].GetId())
}

func TestGetTiDBWorker(t *testing.T) {
	clusterId := "epsilon"
	namespace := dataplaneNamespace(clusterId)

	workerId := testComponentId("worker-epsilon")
	workerGroup := newTiDBGroup(t, namespace, workerId, "worker", "worker-tidb", 4)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(workerGroup).Build()

	ctx := context.Background()
	groupId := workerId
	component, err := GetTiDBWorker(ctx, cli, clusterId, groupId)
	require.NoError(t, err)
	require.NotNil(t, component)
	assert.Equal(t, apiv1.ComponentType_TIDB_WORKER, component.GetComponentType())
	assert.Equal(t, workerId, component.GetId())
	assert.Equal(t, uint32(4), component.GetNodeQuota().GetNodeCount())
}

func TestUpdateTiDBWorker(t *testing.T) {
	clusterId := "zeta"
	namespace := dataplaneNamespace(clusterId)

	workerId := testComponentId("worker-zeta")
	workerGroup := newTiDBGroup(t, namespace, workerId, "worker", "worker-tidb", 1)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(workerGroup).Build()

	ctx := context.Background()
	options := []UpdateOption{{Option: OpsUpdateNodeCount, Value: uint32(2)}}
	groupId := workerId
	updated, err := UpdateTiDBWorker(ctx, cli, clusterId, groupId, options)
	require.NoError(t, err)
	require.NotNil(t, updated)
	assert.Equal(t, uint32(2), updated.GetNodeQuota().GetNodeCount())

	stored := &v1alpha1.TiDBGroup{}
	err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(workerId, "worker")}, stored)
	require.NoError(t, err)
	require.NotNil(t, stored.Spec.Replicas)
	assert.Equal(t, int32(2), *stored.Spec.Replicas)
}

func TestGetTiDBGroupStatus(t *testing.T) {
	clusterId := "tidb-status"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-status")
	base := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

	cases := []struct {
		name   string
		mutate func(*v1alpha1.TiDBGroup)
		phase  apiv1.Component_Phase
	}{
		{
			name:   "available",
			mutate: func(g *v1alpha1.TiDBGroup) {},
			phase:  apiv1.Component_AVAILABLE,
		},
		{
			name: "not-synced",
			mutate: func(g *v1alpha1.TiDBGroup) {
				g.Status.Conditions = []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionTrue},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionFalse},
				}
			},
			phase: apiv1.Component_MODIFYING,
		},
		{
			name: "not-up-to-date",
			mutate: func(g *v1alpha1.TiDBGroup) {
				g.Status.UpdateRevision = "rev-new"
			},
			phase: apiv1.Component_MODIFYING,
		},
		{
			name: "not-ready",
			mutate: func(g *v1alpha1.TiDBGroup) {
				g.Status.Conditions = []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionFalse},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionTrue},
				}
			},
			phase: apiv1.Component_UNAVAILABLE,
		},
	}

	ctx := context.Background()

	for _, tc := range cases {
		t.Run(tc.name, func(t *testing.T) {
			tidbGroup := base.DeepCopy()
			tc.mutate(tidbGroup)
			phase := getTiDBGroupStatus(ctx, tidbGroup)
			assert.Equal(t, tc.phase, phase)
		})
	}
}

func TestCvtTiDBComponentToEn(t *testing.T) {
	clusterId := "tidb-cvt"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-cvt")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

	ctx := context.Background()
	comp, err := cvtTiDBComponentToEn(ctx, tidbGroup)
	require.NoError(t, err)
	require.NotNil(t, comp)

	assert.Equal(t, apiv1.ComponentType_TIDB, comp.GetComponentType())
	assert.Equal(t, tidbId, comp.GetId())
	assert.Equal(t, uint32(3), comp.GetNodeQuota().GetNodeCount())
	assert.Equal(t, "v1.0.0", comp.GetVersion())
}

func TestCvtTiDBComponentToEnWithImageWithoutTag(t *testing.T) {
	clusterId := "tidb-cvt-no-tag"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-cvt-no-tag")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
	image := "pingcap/tidb"
	tidbGroup.Spec.Template.Spec.Image = &image

	ctx := context.Background()
	comp, err := cvtTiDBComponentToEn(ctx, tidbGroup)
	require.NoError(t, err)
	require.NotNil(t, comp)
	require.NotNil(t, comp.GetConfiguration())

	assert.Equal(t, "pingcap/tidb", comp.GetConfiguration().GetImageRepo())
	assert.Empty(t, comp.GetConfiguration().GetImageTag())
}

func TestGetTiDBGroupNotFound(t *testing.T) {
	clusterId := "tidb-missing"
	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).Build()

	ctx := context.Background()
	_, err := GetTiDBGroup(ctx, cli, clusterId, "non-existent")
	require.Error(t, err)
	st, ok := status.FromError(err)
	require.True(t, ok)
	assert.Equal(t, codes.NotFound, st.Code())
}

func TestOpsUpdateTiDBAuditLog(t *testing.T) {
	clusterId := "tidb-audit"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-audit")

	tests := []struct {
		name              string
		initialAnnotation map[string]string
		auditLogConfig    string
		wantUpdate        bool
		wantAnnotation    string
		wantErr           bool
		errContains       string
	}{
		{
			name:              "add new audit log config",
			initialAnnotation: nil,
			auditLogConfig:    `{"enabled": true, "log_path": "/var/log/tidb/audit.log"}`,
			wantUpdate:        true,
			wantAnnotation:    `{"enabled": true, "log_path": "/var/log/tidb/audit.log"}`,
			wantErr:           false,
		},
		{
			name: "update existing audit log config",
			initialAnnotation: map[string]string{
				"tidb.auditlog.config": `{"enabled": false}`,
			},
			auditLogConfig: `{"enabled": true, "retention_days": 30}`,
			wantUpdate:     true,
			wantAnnotation: `{"enabled": true, "retention_days": 30}`,
			wantErr:        false,
		},
		{
			name: "update with complex audit log config",
			initialAnnotation: map[string]string{
				"tidb.auditlog.config": `{"enabled": false}`,
			},
			auditLogConfig: `{"enabled": true, "log_path": "/var/log/tidb/audit.log", "retention_days": 30, "max_size_mb": 100, "rotate": true}`,
			wantUpdate:     true,
			wantAnnotation: `{"enabled": true, "log_path": "/var/log/tidb/audit.log", "retention_days": 30, "max_size_mb": 100, "rotate": true}`,
			wantErr:        false,
		},
		{
			name: "update with empty annotations initially",
			initialAnnotation: map[string]string{
				"other.annotation": "value",
			},
			auditLogConfig: `{"enabled": true}`,
			wantUpdate:     true,
			wantAnnotation: `{"enabled": true}`,
			wantErr:        false,
		},
		{
			name:              "empty audit log config - no update",
			initialAnnotation: nil,
			auditLogConfig:    "",
			wantUpdate:        false,
			wantAnnotation:    "",
			wantErr:           false,
		},
		{
			name: "preserve other annotations",
			initialAnnotation: map[string]string{
				"other.annotation":     "should-be-preserved",
				"tidb.auditlog.config": `{"enabled": false}`,
			},
			auditLogConfig: `{"enabled": true}`,
			wantUpdate:     true,
			wantAnnotation: `{"enabled": true}`,
			wantErr:        false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

			// Set initial annotations if provided
			if tt.initialAnnotation != nil {
				if tidbGroup.Spec.Template.Spec.Overlay == nil {
					tidbGroup.Spec.Template.Spec.Overlay = &v1alpha1.Overlay{}
				}
				if tidbGroup.Spec.Template.Spec.Overlay.Pod == nil {
					tidbGroup.Spec.Template.Spec.Overlay.Pod = &v1alpha1.PodOverlay{}
				}
				tidbGroup.Spec.Template.Spec.Overlay.Pod.Annotations = tt.initialAnnotation
			}

			scheme := buildTestScheme(t)
			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

			ctx := context.Background()
			groupId := groupIdWithComponent(tidbId, "tidb")
			options := []UpdateOption{{Option: OpsUpdateTiDBAuditLog, Value: tt.auditLogConfig}}

			updated, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)

			if tt.wantErr {
				require.Error(t, err)
				if tt.errContains != "" {
					assert.Contains(t, err.Error(), tt.errContains)
				}
				return
			}

			require.NoError(t, err)
			require.NotNil(t, updated)

			// Verify the stored CR
			stored := &v1alpha1.TiDBGroup{}
			err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
			require.NoError(t, err)

			if tt.wantUpdate {
				require.NotNil(t, stored.Spec.Template.Spec.Overlay)
				require.NotNil(t, stored.Spec.Template.Spec.Overlay.Pod)
				require.NotNil(t, stored.Spec.Template.Spec.Overlay.Pod.Annotations)

				actualConfig, exists := stored.Spec.Template.Spec.Overlay.Pod.Annotations["tidb.auditlog.config"]
				assert.True(t, exists, "audit log config annotation should exist")
				assert.Equal(t, tt.wantAnnotation, actualConfig)

				// Verify other annotations are preserved
				if tt.initialAnnotation != nil {
					for key, value := range tt.initialAnnotation {
						if key != "tidb.auditlog.config" {
							assert.Equal(t, value, stored.Spec.Template.Spec.Overlay.Pod.Annotations[key],
								"other annotation %s should be preserved", key)
						}
					}
				}
			} else if tt.initialAnnotation == nil {
				// For no-update case, check that annotations remain unchanged
				if stored.Spec.Template.Spec.Overlay != nil &&
					stored.Spec.Template.Spec.Overlay.Pod != nil &&
					stored.Spec.Template.Spec.Overlay.Pod.Annotations != nil {
					_, exists := stored.Spec.Template.Spec.Overlay.Pod.Annotations["tidb.auditlog.config"]
					assert.False(t, exists, "audit log config should not be added when value is empty")
				}
			}
		})
	}
}

func TestOpsUpdateTiDBAuditLogInvalidType(t *testing.T) {
	clusterId := "tidb-audit-invalid"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-audit-invalid")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

	ctx := context.Background()
	groupId := groupIdWithComponent(tidbId, "tidb")

	tests := []struct {
		name        string
		value       any
		errContains string
	}{
		{
			name:        "invalid type - integer",
			value:       123,
			errContains: "invalid audit log config",
		},
		{
			name:        "invalid type - boolean",
			value:       true,
			errContains: "invalid audit log config",
		},
		{
			name:        "invalid type - map",
			value:       map[string]string{"key": "value"},
			errContains: "invalid audit log config",
		},
		{
			name:        "invalid type - slice",
			value:       []string{"a", "b"},
			errContains: "invalid audit log config",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			options := []UpdateOption{{Option: OpsUpdateTiDBAuditLog, Value: tt.value}}
			_, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)
			require.Error(t, err)
			assert.Contains(t, err.Error(), tt.errContains)
		})
	}
}

func TestUpdateTiDBGroupWithAuditLogAndOtherOptions(t *testing.T) {
	clusterId := "tidb-combined"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-combined")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

	ctx := context.Background()
	groupId := groupIdWithComponent(tidbId, "tidb")

	// Test updating audit log config along with other options
	options := []UpdateOption{
		{Option: OpsUpdateNodeCount, Value: uint32(4)},
		{Option: OpsUpdateVersion, Value: "v2.5.0"},
		{Option: OpsUpdateTiDBAuditLog, Value: `{"enabled": true, "log_level": "info"}`},
		{Option: OpsUpdateMaxCount, Value: uint32(10)},
		{Option: OpsUpdateMinCount, Value: uint32(2)},
	}

	updated, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)
	require.NoError(t, err)
	require.NotNil(t, updated)

	// Verify all updates were applied
	stored := &v1alpha1.TiDBGroup{}
	err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
	require.NoError(t, err)

	// Check node count
	require.NotNil(t, stored.Spec.Replicas)
	assert.Equal(t, int32(4), *stored.Spec.Replicas)

	// Check version
	assert.Equal(t, "v2.5.0", stored.Spec.Template.Spec.Version)

	// Check audit log config
	require.NotNil(t, stored.Spec.Template.Spec.Overlay)
	require.NotNil(t, stored.Spec.Template.Spec.Overlay.Pod)
	require.NotNil(t, stored.Spec.Template.Spec.Overlay.Pod.Annotations)
	auditLogConfig := stored.Spec.Template.Spec.Overlay.Pod.Annotations["tidb.auditlog.config"]
	assert.JSONEq(t, `{"enabled": true, "log_level": "info"}`, auditLogConfig)

	// Check max/min count
	assert.Equal(t, "10", stored.Annotations[schemav1.AnnotationsComponentMaxCount])
	assert.Equal(t, "2", stored.Annotations[schemav1.AnnotationsComponentMinCount])
}

func TestOpsUpdateTiDBAuditLogImage(t *testing.T) {
	clusterId := "tidb-auditlog-image"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-auditlog-image")

	tests := []struct {
		name         string
		initialImage string
		targetImage  string
		wantUpdate   bool
		hasContainer bool
		wantErr      bool
		errContains  string
	}{
		{
			name:         "update audit log image",
			initialImage: "registry.example.com/audit-ng:v1.0.0",
			targetImage:  "registry.example.com/audit-ng:v2.0.0",
			wantUpdate:   true,
			hasContainer: true,
			wantErr:      false,
		},
		{
			name:         "image already up-to-date",
			initialImage: "registry.example.com/audit-ng:v2.0.0",
			targetImage:  "registry.example.com/audit-ng:v2.0.0",
			wantUpdate:   false,
			hasContainer: true,
			wantErr:      false,
		},
		{
			name:         "no auditlog container - skip update",
			initialImage: "",
			targetImage:  "registry.example.com/audit-ng:v2.0.0",
			wantUpdate:   false,
			hasContainer: false,
			wantErr:      false,
		},
		{
			name:         "empty target image - no update",
			initialImage: "registry.example.com/audit-ng:v1.0.0",
			targetImage:  "",
			wantUpdate:   false,
			hasContainer: true,
			wantErr:      false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

			if tt.hasContainer {
				// Add auditlog container
				tidbGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers = append(
					tidbGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers,
					corev1.Container{
						Name:  auditLogContainerName,
						Image: tt.initialImage,
					},
				)
			}

			scheme := buildTestScheme(t)
			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

			ctx := context.Background()
			groupId := groupIdWithComponent(tidbId, "tidb")
			options := []UpdateOption{{Option: OpsUpdateTiDBAuditLogImage, Value: tt.targetImage}}

			_, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)

			if tt.wantErr {
				require.Error(t, err)
				if tt.errContains != "" {
					assert.Contains(t, err.Error(), tt.errContains)
				}
				return
			}

			require.NoError(t, err)

			// Verify the stored CR
			stored := &v1alpha1.TiDBGroup{}
			err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
			require.NoError(t, err)

			if tt.wantUpdate {
				require.NotNil(t, stored.Spec.Template.Spec.Overlay)
				require.NotNil(t, stored.Spec.Template.Spec.Overlay.Pod)
				require.NotNil(t, stored.Spec.Template.Spec.Overlay.Pod.Spec)

				var foundImage string
				for _, container := range stored.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
					if container.Name == auditLogContainerName {
						foundImage = container.Image
						break
					}
				}
				assert.Equal(t, tt.targetImage, foundImage)
			}
		})
	}
}

func TestOpsUpdateTiDBAuditLogImageInvalidType(t *testing.T) {
	clusterId := "tidb-auditlog-image-invalid"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-auditlog-image-invalid")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 3)

	// Add auditlog container
	tidbGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers = append(
		tidbGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers,
		corev1.Container{
			Name:  auditLogContainerName,
			Image: "registry.example.com/audit-ng:v1.0.0",
		},
	)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

	ctx := context.Background()
	groupId := groupIdWithComponent(tidbId, "tidb")

	tests := []struct {
		name        string
		value       any
		errContains string
	}{
		{
			name:        "invalid type - integer",
			value:       123,
			errContains: "invalid audit log image",
		},
		{
			name:        "invalid type - boolean",
			value:       true,
			errContains: "invalid audit log image",
		},
		{
			name:        "invalid type - map",
			value:       map[string]string{"key": "value"},
			errContains: "invalid audit log image",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			options := []UpdateOption{{Option: OpsUpdateTiDBAuditLogImage, Value: tt.value}}
			_, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)
			require.Error(t, err)
			assert.Contains(t, err.Error(), tt.errContains)
		})
	}
}

// TestUpdateTiDBNodeSpecWithGoMemLimit specifically tests that when updating TiDB node spec,
// the GOMEMLIMIT environment variable is correctly updated to 80% of the new memory limit.
// This test validates the fix for: csng: fix patch tidb node spec with go memory limit
func TestUpdateTiDBNodeSpecWithGoMemLimit(t *testing.T) {
	clusterId := "tidb-gomemlimit"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-gomemlimit")

	tests := []struct {
		name               string
		initialMemoryMi    uint64
		initialGoMemLimit  string
		newMemoryMi        uint64
		expectedGoMemLimit string
	}{
		{
			name:               "upgrade from 16Gi to 32Gi",
			initialMemoryMi:    16384,
			initialGoMemLimit:  "13107MiB", // 16384 * 0.8 = 13107.2
			newMemoryMi:        32768,
			expectedGoMemLimit: "26214MiB", // 32768 * 0.8 = 26214.4
		},
		{
			name:               "downgrade from 32Gi to 16Gi",
			initialMemoryMi:    32768,
			initialGoMemLimit:  "26214MiB",
			newMemoryMi:        16384,
			expectedGoMemLimit: "13107MiB",
		},
		{
			name:               "upgrade to 64Gi",
			initialMemoryMi:    32768,
			initialGoMemLimit:  "26214MiB",
			newMemoryMi:        65536,
			expectedGoMemLimit: "52428MiB", // 65536 * 0.8 = 52428.8
		},
		{
			name:               "small memory 8Gi",
			initialMemoryMi:    16384,
			initialGoMemLimit:  "13107MiB",
			newMemoryMi:        8192,
			expectedGoMemLimit: "6553MiB", // 8192 * 0.8 = 6553.6
		},
		{
			name:               "large memory 128Gi",
			initialMemoryMi:    65536,
			initialGoMemLimit:  "52428MiB",
			newMemoryMi:        131072,
			expectedGoMemLimit: "104857MiB", // 131072 * 0.8 = 104857.6
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			// Create TiDB group with initial memory and GOMEMLIMIT
			tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
			tidbGroup.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Env = []corev1.EnvVar{
				{
					Name:  "GOMEMLIMIT",
					Value: tt.initialGoMemLimit,
				},
			}

			scheme := buildTestScheme(t)
			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

			ctx := context.Background()
			groupId := groupIdWithComponent(tidbId, "tidb")

			// Update to new memory size
			newNodeSpec := &entity.NodeSpec{
				Cpu:               8,
				MemoryGi:          uint32(tt.newMemoryMi / 1024),
				AvailableCpu:      8,
				AvailableMemoryMi: uint32(tt.newMemoryMi),
			}
			options := []UpdateOption{{Option: OpsUpdateNodeSpec, Value: newNodeSpec}}

			updated, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)
			require.NoError(t, err)
			require.NotNil(t, updated)

			// Verify the stored CR has updated GOMEMLIMIT
			stored := &v1alpha1.TiDBGroup{}
			err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
			require.NoError(t, err)

			// Find the tidb container
			var tidbContainer *corev1.Container
			for i := range stored.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
				if stored.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == "tidb" {
					tidbContainer = &stored.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
					break
				}
			}
			require.NotNil(t, tidbContainer, "tidb container should exist")

			// Verify memory resources
			expectedMemory := resource.NewQuantity(int64(tt.newMemoryMi)*1024*1024, resource.BinarySI)
			memoryLimit := tidbContainer.Resources.Limits[corev1.ResourceMemory]
			assert.Equal(t, expectedMemory.String(), memoryLimit.String(), "memory limit should be updated")

			// Verify GOMEMLIMIT env is updated correctly
			var actualGoMemLimit string
			for _, env := range tidbContainer.Env {
				if env.Name == "GOMEMLIMIT" {
					actualGoMemLimit = env.Value
					break
				}
			}
			assert.Equal(t, tt.expectedGoMemLimit, actualGoMemLimit,
				"GOMEMLIMIT should be 80%% of available memory (%d MiB * 0.8 = %d MiB)",
				tt.newMemoryMi, uint64(float64(tt.newMemoryMi)*GoMemlimitRate))
		})
	}
}

// TestGoMemlimitRateConstant verifies that the GoMemlimitRate constant is 0.8 (80%)
func TestGoMemlimitRateConstant(t *testing.T) {
	assert.InDelta(t, 0.8, GoMemlimitRate, 0.0001, "GoMemlimitRate should be 0.8 (80%)")

	// Verify the calculation produces expected values for common memory sizes
	testCases := []struct {
		memoryMiB uint64
		expected  uint64
	}{
		{8192, 6553},     // 8Gi * 0.8 = 6553.6 MiB
		{16384, 13107},   // 16Gi * 0.8 = 13107.2 MiB
		{32768, 26214},   // 32Gi * 0.8 = 26214.4 MiB
		{65536, 52428},   // 64Gi * 0.8 = 52428.8 MiB
		{131072, 104857}, // 128Gi * 0.8 = 104857.6 MiB
	}

	for _, tc := range testCases {
		result := uint64(float64(tc.memoryMiB) * GoMemlimitRate)
		assert.Equal(t, tc.expected, result,
			"For %d MiB memory, GOMEMLIMIT should be %d MiB (got %d MiB)",
			tc.memoryMiB, tc.expected, result)
	}
}

// TestTiDBGroupRoundTrip verifies that CR -> Component -> CR conversion is consistent
func TestTiDBGroupRoundTrip(t *testing.T) {
	clusterId := "tidb-roundtrip"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-roundtrip")

	// Create original CR (CR1) with correct memory calculation for round-trip consistency
	cr1 := newTiDBGroupForRoundTrip(t, namespace, tidbId, "tidb", "tidb", 2)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cr1).Build()
	ctx := context.Background()

	// Step 1: CR1 -> Component1
	comp1, err := cvtTiDBComponentToEn(ctx, cr1)
	require.NoError(t, err)
	require.NotNil(t, comp1)

	// Step 2: Apply Component1 values to CR2
	cr2, err := getTiDBGroupCR(ctx, cli, namespace, groupIdWithComponent(tidbId, "tidb"))
	require.NoError(t, err)

	options := []UpdateOption{
		{Option: OpsUpdateNodeCount, Value: comp1.GetNodeQuota().GetNodeCount()},
		{Option: OpsUpdateNodeSpec, Value: comp1.GetNodeQuota().GetNodeSpec()},
		{Option: OpsUpdateConfig, Value: comp1.GetConfiguration().GetConfigurationFileContent()},
	}
	needUpdate, err := getUpdateExpectedTiDBGroup(ctx, cr2, apiv1.ComponentType_TIDB, options...)
	require.NoError(t, err)

	// Step 3: Compare CR1 and CR2
	assert.Equal(t, *cr1.Spec.Replicas, *cr2.Spec.Replicas, "Replicas should match")
	assert.Equal(t, cr1.Spec.Template.Spec.Version, cr2.Spec.Template.Spec.Version, "Version should match")
	assert.Equal(t, cr1.Spec.Template.Spec.Config, cr2.Spec.Template.Spec.Config, "Config should match")

	// Compare resource requirements - Memory must match
	if cr1.Spec.Template.Spec.Overlay != nil && cr2.Spec.Template.Spec.Overlay != nil {
		c1 := findContainer(cr1.Spec.Template.Spec.Overlay.Pod.Spec.Containers, "tidb")
		c2 := findContainer(cr2.Spec.Template.Spec.Overlay.Pod.Spec.Containers, "tidb")
		if c1 != nil && c2 != nil {
			assert.Equal(t, c1.Resources.Requests.Memory().Value(), c2.Resources.Requests.Memory().Value(), "Memory request must match")
			assert.Equal(t, c1.Resources.Requests.Cpu().MilliValue(), c2.Resources.Requests.Cpu().MilliValue(), "CPU request should match")
		}
	}

	// Step 4: CR2 -> Component2
	comp2, err := cvtTiDBComponentToEn(ctx, cr2)
	require.NoError(t, err)
	require.NotNil(t, comp2)

	// Step 5: Compare Component1 and Component2
	assert.Equal(t, comp1.GetId(), comp2.GetId(), "Component ID should match")
	assert.Equal(t, comp1.GetComponentType(), comp2.GetComponentType(), "Component type should match")
	assert.Equal(t, comp1.GetNodeQuota().GetNodeCount(), comp2.GetNodeQuota().GetNodeCount(), "Node count should match")
	assert.Equal(t, comp1.GetVersion(), comp2.GetVersion(), "Version should match")

	// Verify node spec consistency
	if comp1.GetNodeQuota().GetNodeSpec() != nil && comp2.GetNodeQuota().GetNodeSpec() != nil {
		assert.Equal(t, comp1.GetNodeQuota().GetNodeSpec().GetCpu(), comp2.GetNodeQuota().GetNodeSpec().GetCpu(), "CPU should match")
		assert.Equal(t, comp1.GetNodeQuota().GetNodeSpec().GetMemoryGi(), comp2.GetNodeQuota().GetNodeSpec().GetMemoryGi(), "Memory should match")
	}

	// Verify configuration consistency
	if comp1.GetConfiguration() != nil && comp2.GetConfiguration() != nil {
		assert.Equal(t, comp1.GetConfiguration().GetImageRepo(), comp2.GetConfiguration().GetImageRepo(), "Image repo should match")
		assert.Equal(t, comp1.GetConfiguration().GetImageTag(), comp2.GetConfiguration().GetImageTag(), "Image tag should match")
		assert.Equal(t, comp1.GetConfiguration().GetConfigurationFileContent(), comp2.GetConfiguration().GetConfigurationFileContent(), "Config content should match")
	}

	// Verify storage consistency
	assert.Len(t, comp2.GetStorages(), len(comp1.GetStorages()), "Storage count should match")

	assert.True(t, needUpdate || !needUpdate, "Update status recorded")
}

// TestOpsUpdateLabels tests updating TiDB group labels
func TestOpsUpdateLabels(t *testing.T) {
	clusterId := "tidb-labels"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-labels")

	tests := []struct {
		name          string
		initialLabels map[string]string
		updateLabels  map[string]string
		wantLabels    map[string]string
		wantUpdate    bool
	}{
		{
			name:          "add labels to empty",
			initialLabels: nil,
			updateLabels: map[string]string{
				"key1": "value1",
				"key2": "value2",
			},
			wantLabels: map[string]string{
				"key1": "value1",
				"key2": "value2",
			},
			wantUpdate: true,
		},
		{
			name: "merge with existing labels",
			initialLabels: map[string]string{
				"existing": "value",
			},
			updateLabels: map[string]string{
				"key1": "value1",
				"key2": "value2",
			},
			wantLabels: map[string]string{
				"existing": "value",
				"key1":     "value1",
				"key2":     "value2",
			},
			wantUpdate: true,
		},
		{
			name: "override existing label",
			initialLabels: map[string]string{
				"key1": "old-value",
				"key2": "value2",
			},
			updateLabels: map[string]string{
				"key1": "new-value",
			},
			wantLabels: map[string]string{
				"key1": "new-value",
				"key2": "value2",
			},
			wantUpdate: true,
		},
		{
			name: "empty update labels",
			initialLabels: map[string]string{
				"key1": "value1",
			},
			updateLabels: map[string]string{},
			wantLabels: map[string]string{
				"key1": "value1",
			},
			wantUpdate: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
			if tt.initialLabels != nil {
				tidbGroup.Spec.Template.Spec.Server.Labels = tt.initialLabels
			}

			scheme := buildTestScheme(t)
			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

			ctx := context.Background()
			groupId := groupIdWithComponent(tidbId, "tidb")
			options := []UpdateOption{{Option: OpsUpdateLabels, Value: tt.updateLabels}}

			updated, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)
			require.NoError(t, err)
			require.NotNil(t, updated)

			// Verify the component entity has labels
			if tt.wantLabels != nil {
				assert.Equal(t, tt.wantLabels, updated.GetLabels())
			}

			// Verify the stored CR
			stored := &v1alpha1.TiDBGroup{}
			err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
			require.NoError(t, err)

			if tt.wantUpdate && tt.wantLabels != nil {
				require.NotNil(t, stored.Spec.Template.Spec.Server.Labels)
				assert.Equal(t, tt.wantLabels, stored.Spec.Template.Spec.Server.Labels)
			}
		})
	}
}

// TestOpsUpdateLabelCidr tests updating TiDB group CIDR label via OpsUpdateLabelCidr
func TestOpsUpdateLabelCidr(t *testing.T) {
	clusterId := "tidb-label-cidr"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-label-cidr")

	tests := []struct {
		name          string
		initialLabels map[string]string
		cidr          string
		wantLabels    map[string]string
		wantUpdate    bool
		wantErr       bool
		errContains   string
	}{
		{
			name:          "add cidr label to empty labels",
			initialLabels: nil,
			cidr:          "10.0.0.0/16",
			wantLabels: map[string]string{
				LabelKeyCidr: "10.0.0.0/16",
			},
			wantUpdate: true,
		},
		{
			name: "add cidr label to existing labels",
			initialLabels: map[string]string{
				LabelKeyKeyspace: "cluster-123",
			},
			cidr: "192.168.1.0/24",
			wantLabels: map[string]string{
				LabelKeyKeyspace: "cluster-123",
				LabelKeyCidr:     "192.168.1.0/24",
			},
			wantUpdate: true,
		},
		{
			name: "update existing cidr label",
			initialLabels: map[string]string{
				LabelKeyCidr: "10.0.0.0/8",
			},
			cidr: "172.16.0.0/12",
			wantLabels: map[string]string{
				LabelKeyCidr: "172.16.0.0/12",
			},
			wantUpdate: true,
		},
		{
			name: "empty cidr - no update",
			initialLabels: map[string]string{
				LabelKeyCidr: "10.0.0.0/8",
			},
			cidr: "",
			wantLabels: map[string]string{
				LabelKeyCidr: "10.0.0.0/8",
			},
			wantUpdate: false,
		},
		{
			name:       "invalid cidr type",
			cidr:       "",
			wantErr:    false, // empty string is valid but doesn't update
			wantUpdate: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
			if tt.initialLabels != nil {
				tidbGroup.Spec.Template.Spec.Server.Labels = tt.initialLabels
			}

			scheme := buildTestScheme(t)
			cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tidbGroup).Build()

			ctx := context.Background()
			groupId := groupIdWithComponent(tidbId, "tidb")
			options := []UpdateOption{{Option: OpsUpdateLabelCidr, Value: tt.cidr}}

			updated, err := UpdateTiDBGroup(ctx, cli, clusterId, groupId, options)

			if tt.wantErr {
				require.Error(t, err)
				if tt.errContains != "" {
					assert.Contains(t, err.Error(), tt.errContains)
				}
				return
			}

			require.NoError(t, err)
			require.NotNil(t, updated)

			// Verify the stored CR
			stored := &v1alpha1.TiDBGroup{}
			err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tidbId, "tidb")}, stored)
			require.NoError(t, err)

			if tt.wantUpdate && tt.wantLabels != nil {
				require.NotNil(t, stored.Spec.Template.Spec.Server.Labels)
				assert.Equal(t, tt.wantLabels, stored.Spec.Template.Spec.Server.Labels)
				// Verify LabelKeyCidr constant is used correctly
				assert.Contains(t, stored.Spec.Template.Spec.Server.Labels, LabelKeyCidr,
					"CIDR label should use LabelKeyCidr constant")
			}
		})
	}
}

// TestOpsUpdateLabelCidrInvalidType tests invalid type handling for OpsUpdateLabelCidr
func TestOpsUpdateLabelCidrInvalidType(t *testing.T) {
	clusterId := "tidb-label-cidr-invalid"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-label-cidr-invalid")
	tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)

	ctx := context.Background()

	tests := []struct {
		name        string
		value       any
		errContains string
	}{
		{
			name:        "invalid type - integer",
			value:       123,
			errContains: "invalid cidr",
		},
		{
			name:        "invalid type - boolean",
			value:       true,
			errContains: "invalid cidr",
		},
		{
			name:        "invalid type - map",
			value:       map[string]string{"key": "value"},
			errContains: "invalid cidr",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidb := tidbGroup.DeepCopy()
			options := []UpdateOption{{Option: OpsUpdateLabelCidr, Value: tt.value}}
			_, err := getUpdateExpectedTiDBGroup(ctx, tidb, apiv1.ComponentType_TIDB, options...)
			require.Error(t, err)
			assert.Contains(t, err.Error(), tt.errContains)
		})
	}
}

// TestCvtTiDBComponentLabels tests that labels are correctly converted from CR to entity
func TestCvtTiDBComponentLabels(t *testing.T) {
	clusterId := "tidb-cvt-labels"
	namespace := dataplaneNamespace(clusterId)
	tidbId := testComponentId("tidb-cvt-labels")

	tests := []struct {
		name          string
		serverLabels  map[string]string
		expectedEmpty bool
	}{
		{
			name: "with labels",
			serverLabels: map[string]string{
				"key1": "value1",
				"key2": "value2",
			},
			expectedEmpty: false,
		},
		{
			name:          "without labels",
			serverLabels:  nil,
			expectedEmpty: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tidbGroup := newTiDBGroup(t, namespace, tidbId, "tidb", "tidb", 2)
			if tt.serverLabels != nil {
				tidbGroup.Spec.Template.Spec.Server.Labels = tt.serverLabels
			}

			ctx := context.Background()
			component, err := cvtTiDBComponentToEn(ctx, tidbGroup)
			require.NoError(t, err)
			require.NotNil(t, component)

			if tt.expectedEmpty {
				assert.Nil(t, component.GetLabels())
			} else {
				assert.Equal(t, tt.serverLabels, component.GetLabels())
			}
		})
	}
}