package tioperator

import (
	"context"
	"errors"
	"testing"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/samber/lo"
	"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"
	"sigs.k8s.io/controller-runtime/pkg/client/interceptor"
)

func TestListTiKVGroups(t *testing.T) {
	clusterId := "tikv-a"
	namespace := dataplaneNamespace(clusterId)

	tikvId := testComponentId("tikv-a")
	otherId := testComponentId("tikv-b")

	tikv := newTiKVGroup(t, namespace, tikvId, 3)
	other := newTiKVGroup(t, namespace, otherId, 2)

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tikv, other).Build()

	ctx := context.Background()
	comps, err := ListTiKVGroups(ctx, cli, clusterId)
	require.NoError(t, err)
	require.Len(t, comps, 2)
	for _, comp := range comps {
		assert.Equal(t, apiv1.ComponentType_TIKV, comp.GetComponentType())
		assert.NotEmpty(t, comp.GetId())
	}
}

func TestListTiKVGroupsSkipsInvalidComponent(t *testing.T) {
	clusterId := "tikv-invalid"
	namespace := dataplaneNamespace(clusterId)

	validId := testComponentId("tikv-valid")
	invalidId := testComponentId("tikv-invalid")

	valid := newTiKVGroup(t, namespace, validId, 3)
	invalid := newTiKVGroup(t, namespace, invalidId, 2)
	invalid.Labels[schemav1.LabelNodeSpecTypeKey] = "bogus"

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(valid, invalid).Build()

	ctx := context.Background()
	comps, err := ListTiKVGroups(ctx, cli, clusterId)
	require.NoError(t, err)
	require.Len(t, comps, 1)
	assert.Equal(t, validId, comps[0].GetId())
}

func TestGetTiKVGroup(t *testing.T) {
	clusterId := "tikv-c"
	namespace := dataplaneNamespace(clusterId)

	tikvId := testComponentId("tikv-c")
	tikv := newTiKVGroup(t, namespace, tikvId, 4)

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

	ctx := context.Background()
	groupId := groupIdWithComponent(tikvId, "tikv")
	comp, err := GetTiKVGroup(ctx, cli, clusterId, groupId)
	require.NoError(t, err)
	require.NotNil(t, comp)
	assert.Equal(t, apiv1.ComponentType_TIKV, comp.GetComponentType())
	assert.Equal(t, tikvId, comp.GetId())
	assert.Equal(t, uint32(4), comp.GetNodeQuota().GetNodeCount())
}

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

	ctx := context.Background()
	_, err := GetTiKVGroup(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 TestUpdateTiKVGroup(t *testing.T) {
	clusterId := "tikv-d"
	namespace := dataplaneNamespace(clusterId)

	tikvId := testComponentId("tikv-d")
	tikv := newTiKVGroup(t, namespace, tikvId, 2)

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

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

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

func TestUpdateTiKVGroupPreservesRemoteWorkersOnConfigUpdate(t *testing.T) {
	clusterId := "tikv-remote-worker"
	namespace := dataplaneNamespace(clusterId)

	tikvId := testComponentId("tikv-remote-worker")
	tikv := newTiKVGroup(t, namespace, tikvId, 2)
	tikv.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
		Worker: v1alpha1.WorkerReference{
			Name: groupNameWithComponent(tikvId, "default"),
		},
		Coprocessor: &v1alpha1.CoprocessorReference{
			Name: groupNameWithComponent(tikvId, "default"),
		},
	}

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

	ctx := context.Background()
	groupId := groupIdWithComponent(tikvId, "tikv")
	_, err := UpdateTiKVGroup(ctx, cli, clusterId, groupId, []UpdateOption{
		{Option: OpsUpdateConfig, Value: "new-config"},
	})
	require.NoError(t, err)

	stored := &v1alpha1.TiKVGroup{}
	err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tikvId, "tikv")}, stored)
	require.NoError(t, err)
	require.NotNil(t, stored.Spec.Template.Spec.RemoteWorkers)
	require.NotNil(t, stored.Spec.Template.Spec.RemoteWorkers.Coprocessor)
	assert.Equal(t, groupNameWithComponent(tikvId, "default"), stored.Spec.Template.Spec.RemoteWorkers.Worker.Name)
	assert.Equal(t, groupNameWithComponent(tikvId, "default"), stored.Spec.Template.Spec.RemoteWorkers.Coprocessor.Name)
}

func TestGetTiKVGroupCRAddsPrefix(t *testing.T) {
	clusterId := "tikv-prefix"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-prefix")

	tikv := newTiKVGroup(t, namespace, tikvId, 2)
	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(tikv).Build()

	ctx := context.Background()
	groupId := groupIdWithComponent(tikvId, "tikv")
	fetched, err := getTiKVGroupCR(ctx, cli, namespace, groupId)
	require.NoError(t, err)
	assert.Equal(t, tikv.Name, fetched.Name)

	fetchedWithPrefix, err := getTiKVGroupCR(ctx, cli, namespace, tikv.Name)
	require.NoError(t, err)
	assert.Equal(t, tikv.Name, fetchedWithPrefix.Name)
}

func TestGetTiKVGroupStatus(t *testing.T) {
	clusterId := "tikv-status"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-status")
	base := newTiKVGroup(t, namespace, tikvId, 3)

	cases := []struct {
		name   string
		mutate func(*v1alpha1.TiKVGroup)
		phase  apiv1.Component_Phase
	}{
		{
			name:   "available",
			mutate: func(g *v1alpha1.TiKVGroup) {},
			phase:  apiv1.Component_AVAILABLE,
		},
		{
			name: "not-synced",
			mutate: func(g *v1alpha1.TiKVGroup) {
				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.TiKVGroup) {
				g.Status.UpdateRevision = "rev-new"
			},
			phase: apiv1.Component_MODIFYING,
		},
		{
			name: "not-ready",
			mutate: func(g *v1alpha1.TiKVGroup) {
				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) {
			tikvGroup := base.DeepCopy()
			tc.mutate(tikvGroup)
			phase := getTiKVGroupStatus(ctx, tikvGroup)
			assert.Equal(t, tc.phase, phase)
		})
	}
}

func TestGetUpdateExpectedTiKVGroup(t *testing.T) {
	clusterId := "tikv-update"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-update")

	base := newTiKVGroup(t, namespace, tikvId, 3)
	base.Spec.Template.Spec.Volumes = []v1alpha1.Volume{
		{
			Name:                      string(schemav1.StorageData),
			Storage:                   resource.MustParse("200Gi"),
			StorageClassName:          lo.ToPtr("cloud-ssd"),
			VolumeAttributesClassName: lo.ToPtr("cloud-essd-1000-200"),
		},
		{
			Name:                      string(schemav1.StorageRaftLog),
			Storage:                   resource.MustParse("50Gi"),
			StorageClassName:          lo.ToPtr("cloud-ssd"),
			VolumeAttributesClassName: lo.ToPtr("cloud-auto-1000-150"),
		},
	}

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

	cases := []struct {
		name            string
		mutate          func(*v1alpha1.TiKVGroup)
		options         []UpdateOption
		wantUpdate      bool
		wantErrContains string
		wantErrCode     codes.Code
		verify          func(*testing.T, *v1alpha1.TiKVGroup)
	}{
		{
			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/custom:t"},
				{Option: OpsUpdateConfig, Value: "new-config"},
				{Option: OpsUpdateUpdateStrategy, Value: v1alpha1.ConfigUpdateStrategyHotReload},
				{Option: OpsUpdateDataSize, Value: uint32(400)},
				{Option: OpsUpdateDataVolumeAttributesClass, Value: "cloud-essd-2000-600"},
				{Option: OpsUpdateRaftLogSize, Value: uint32(120)},
				{Option: OpsUpdateRaftLogVolumeAttributesClass, Value: "cloud-auto-1600-300"},
			},
			wantUpdate: true,
			verify: func(t *testing.T, tikv *v1alpha1.TiKVGroup) {
				t.Helper()
				require.NotNil(t, tikv.Spec.Replicas)
				assert.Equal(t, int32(4), *tikv.Spec.Replicas)

				tikvContainer := findContainerByName(t, tikv, "tikv")
				require.NotNil(t, tikvContainer)

				assertMainContainerResources(t, tikvContainer.Resources, nodeSpec, apiv1.ComponentType_TIKV)

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

				annotations := tikv.GetAnnotations()
				assert.Equal(t, "8", annotations[schemav1.AnnotationsComponentMaxCount])
				assert.Equal(t, "2", annotations[schemav1.AnnotationsComponentMinCount])

				require.Len(t, tikv.Spec.Template.Spec.Volumes, 2)

				dataVolume := volumeByName(tikv.Spec.Template.Spec.Volumes, string(schemav1.StorageData))
				require.NotNil(t, dataVolume)
				assert.Equal(t, resource.MustParse("400Gi"), dataVolume.Storage)
				require.NotNil(t, dataVolume.StorageClassName)
				assert.Equal(t, "cloud-ssd", *dataVolume.StorageClassName)
				require.NotNil(t, dataVolume.VolumeAttributesClassName)
				assert.Equal(t, "cloud-essd-2000-600", *dataVolume.VolumeAttributesClassName)

				raftVolume := volumeByName(tikv.Spec.Template.Spec.Volumes, string(schemav1.StorageRaftLog))
				require.NotNil(t, raftVolume)
				assert.Equal(t, resource.MustParse("120Gi"), raftVolume.Storage)
				require.NotNil(t, raftVolume.StorageClassName)
				assert.Equal(t, "cloud-ssd", *raftVolume.StorageClassName)
				require.NotNil(t, raftVolume.VolumeAttributesClassName)
				assert.Equal(t, "cloud-auto-1600-300", *raftVolume.VolumeAttributesClassName)

				assert.Contains(t, annotations, schemav1.AnnotationsUpdateTimeDataVolumeKey)
				assert.Contains(t, annotations, schemav1.AnnotationsUpdateTimeRaftLogVolumeKey)
			},
		},
		{
			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-data-size",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.Volumes = []v1alpha1.Volume{{Name: string(schemav1.StorageData)}}
			},
			options:         []UpdateOption{{Option: OpsUpdateDataSize, Value: "bad"}},
			wantErrContains: "invalid data storage size",
		},
		{
			name: "invalid-raft-size",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.Volumes = []v1alpha1.Volume{{Name: string(schemav1.StorageRaftLog)}}
			},
			options:         []UpdateOption{{Option: OpsUpdateRaftLogSize, Value: "bad"}},
			wantErrContains: "invalid raft log storage size",
		},
		{
			name: "invalid-data-volume-attributes-type",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.Volumes = []v1alpha1.Volume{{Name: string(schemav1.StorageData)}}
			},
			options:         []UpdateOption{{Option: OpsUpdateDataVolumeAttributesClass, Value: 321}},
			wantErrContains: "invalid data volume attributes class",
		},
		{
			name: "invalid-raftlog-volume-attributes-type",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.Volumes = []v1alpha1.Volume{{Name: string(schemav1.StorageRaftLog)}}
			},
			options:         []UpdateOption{{Option: OpsUpdateRaftLogVolumeAttributesClass, Value: 789}},
			wantErrContains: "invalid raft log volume attributes class",
		},
		{
			name:            "invalid-update-strategy-type",
			options:         []UpdateOption{{Option: OpsUpdateUpdateStrategy, Value: 123}},
			wantErrContains: "invalid update strategy",
		},
		{
			name:       "no-op-options",
			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, tikv *v1alpha1.TiKVGroup) {
				t.Helper()
				require.NotNil(t, tikv.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector)
				assert.Equal(t, "1.28", tikv.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector[schemav1.K8sVersionLabelKey])
			},
		},
		{
			name: "update-node-version-idempotent",
			mutate: func(g *v1alpha1.TiKVGroup) {
				if g.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector == nil {
					g.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector = map[string]string{}
				}
				g.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector[schemav1.K8sVersionLabelKey] = "1.27"
			},
			options:    []UpdateOption{{Option: OpsUpdateNodeVersion, Value: "1.27"}},
			wantUpdate: false,
		},
		{
			name: "update-node-version-nil-nodeSelector",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector = nil
			},
			options:    []UpdateOption{{Option: OpsUpdateNodeVersion, Value: "1.29"}},
			wantUpdate: true,
			verify: func(t *testing.T, tikv *v1alpha1.TiKVGroup) {
				t.Helper()
				require.NotNil(t, tikv.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector)
				assert.Equal(t, "1.29", tikv.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) {
			tikv := base.DeepCopy()
			if tc.mutate != nil {
				tc.mutate(tikv)
			}

			needUpdate, err := getUpdateExpectedTiKVGroup(ctx, tikv, 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, tikv)
			}
		})
	}
}

func findContainerByName(t *testing.T, tikv *v1alpha1.TiKVGroup, name string) *corev1.Container {
	t.Helper()
	if tikv.Spec.Template.Spec.Overlay == nil || tikv.Spec.Template.Spec.Overlay.Pod == nil || tikv.Spec.Template.Spec.Overlay.Pod.Spec == nil {
		return nil
	}
	for i := range tikv.Spec.Template.Spec.Overlay.Pod.Spec.Containers {
		if tikv.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i].Name == name {
			return &tikv.Spec.Template.Spec.Overlay.Pod.Spec.Containers[i]
		}
	}
	return nil
}

func volumeByName(volumes []v1alpha1.Volume, name string) *v1alpha1.Volume {
	for i := range volumes {
		if volumes[i].Name == name {
			return &volumes[i]
		}
	}
	return nil
}

func TestCvtTiKVComponentToEn(t *testing.T) {
	clusterId := "tikv-cvt"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-cvt")
	tikvGroup := newTiKVGroup(t, namespace, tikvId, 3)

	ctx := context.Background()
	comp, err := cvtTiKVComponentToEn(ctx, tikvGroup)
	require.NoError(t, err)
	require.NotNil(t, comp)

	assert.Equal(t, apiv1.ComponentType_TIKV, comp.GetComponentType())
	assert.Equal(t, tikvId, comp.GetId())
	assert.Equal(t, uint32(3), comp.GetNodeQuota().GetNodeCount())
	assert.Equal(t, "v1.1.0", comp.GetVersion())
}

func TestListTiKVGroupsEmptyNamespace(t *testing.T) {
	clusterId := "tikv-empty"
	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).Build()

	ctx := context.Background()
	comps, err := ListTiKVGroups(ctx, cli, clusterId)
	require.NoError(t, err)
	assert.Empty(t, comps)
}

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

	// Create original CR (CR1) with correct memory calculation for round-trip consistency
	cr1 := newTiKVGroupForRoundTrip(t, namespace, tikvId, 3)
	cr1.Spec.Template.Spec.Volumes = []v1alpha1.Volume{
		{
			Name:                      "data",
			StorageClassName:          lo.ToPtr("cloud-ssd"),
			VolumeAttributesClassName: lo.ToPtr("gp3-6000-625"),
			Storage:                   resource.MustParse("500Gi"),
			Mounts: []v1alpha1.VolumeMount{
				{MountPath: "/data"},
			},
		},
		{
			Name:                      "raftlog",
			StorageClassName:          lo.ToPtr("cloud-ssd"),
			VolumeAttributesClassName: lo.ToPtr("io2-10000"),
			Storage:                   resource.MustParse("100Gi"),
			Mounts: []v1alpha1.VolumeMount{
				{MountPath: "/var/lib/raft-log"},
			},
		},
	}

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

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

	// Step 2: Apply Component1 values to CR2
	cr2, err := getTiKVGroupCR(ctx, cli, namespace, groupIdWithComponent(tikvId, "tikv"))
	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 := getUpdateExpectedTiKVGroup(ctx, cr2, 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")
	assert.Len(t, cr2.Spec.Template.Spec.Volumes, len(cr1.Spec.Template.Spec.Volumes), "Volume count 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, "tikv")
		c2 := findContainer(cr2.Spec.Template.Spec.Overlay.Pod.Spec.Containers, "tikv")
		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 := cvtTiKVComponentToEn(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")
	for i := range comp1.GetStorages() {
		assert.Equal(t, comp1.GetStorages()[i].GetName(), comp2.GetStorages()[i].GetName(), "Storage name should match")
		assert.Equal(t, comp1.GetStorages()[i].GetType(), comp2.GetStorages()[i].GetType(), "Storage type should match")
		assert.Equal(t, comp1.GetStorages()[i].GetSizeGi(), comp2.GetStorages()[i].GetSizeGi(), "Storage size should match")
	}

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

// TestOpsUpdateTiKVLabels tests updating TiKV group labels
func TestOpsUpdateTiKVLabels(t *testing.T) {
	clusterId := "tikv-labels"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-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{
				"zone": "zone-1",
				"rack": "rack-1",
			},
			wantLabels: map[string]string{
				"zone": "zone-1",
				"rack": "rack-1",
			},
			wantUpdate: true,
		},
		{
			name: "merge with existing labels",
			initialLabels: map[string]string{
				"existing": "value",
			},
			updateLabels: map[string]string{
				"zone": "zone-1",
			},
			wantLabels: map[string]string{
				"existing": "value",
				"zone":     "zone-1",
			},
			wantUpdate: true,
		},
		{
			name: "override existing label",
			initialLabels: map[string]string{
				"zone": "old-zone",
			},
			updateLabels: map[string]string{
				"zone": "new-zone",
			},
			wantLabels: map[string]string{
				"zone": "new-zone",
			},
			wantUpdate: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tikvGroup := newTiKVGroup(t, namespace, tikvId, 3)
			if tt.initialLabels != nil {
				tikvGroup.Spec.Template.Spec.Server.Labels = tt.initialLabels
			}

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

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

			updated, err := UpdateTiKVGroup(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.TiKVGroup{}
			err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tikvId, "tikv")}, 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)
			}
		})
	}
}

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

	tests := []struct {
		name          string
		serverLabels  map[string]string
		expectedEmpty bool
	}{
		{
			name: "with labels",
			serverLabels: map[string]string{
				"zone": "zone-1",
				"rack": "rack-1",
			},
			expectedEmpty: false,
		},
		{
			name:          "without labels",
			serverLabels:  nil,
			expectedEmpty: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tikvGroup := newTiKVGroup(t, namespace, tikvId, 3)
			if tt.serverLabels != nil {
				tikvGroup.Spec.Template.Spec.Server.Labels = tt.serverLabels
			}

			ctx := context.Background()
			component, err := cvtTiKVComponentToEn(ctx, tikvGroup)
			require.NoError(t, err)
			require.NotNil(t, component)

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

func TestGetUpdateExpectedTiKVGroupRemoteWorkers(t *testing.T) {
	clusterId := "tikv-remote-workers"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-remote-workers")
	workerName := groupNameWithComponent(tikvId, "default")
	coprName := groupNameWithComponent(tikvId, "coprocessor")

	tests := []struct {
		name            string
		initial         *v1alpha1.TiKVRemoteWorkers
		target          any
		wantUpdate      bool
		wantErr         bool
		wantWorker      string
		wantCoprocessor string
	}{
		{
			name:            "nil remote workers initializes",
			initial:         nil,
			target:          TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			wantUpdate:      true,
			wantWorker:      workerName,
			wantCoprocessor: coprName,
		},
		{
			name: "unchanged is no-op",
			initial: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: workerName},
				Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
			},
			target:          TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			wantUpdate:      false,
			wantWorker:      workerName,
			wantCoprocessor: coprName,
		},
		{
			name: "changed updates both",
			initial: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: "old-worker"},
				Coprocessor: &v1alpha1.CoprocessorReference{Name: "old-coprocessor"},
			},
			target:          TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			wantUpdate:      true,
			wantWorker:      workerName,
			wantCoprocessor: coprName,
		},
		{
			name: "empty worker is invalid",
			initial: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: workerName},
				Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
			},
			target:  TiKVRemoteWorkerTarget{Worker: "", Coprocessor: coprName},
			wantErr: true,
		},
		{
			name: "empty coprocessor is invalid",
			initial: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: workerName},
				Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
			},
			target:  TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: ""},
			wantErr: true,
		},
		{
			name: "wrong value type is invalid",
			initial: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: workerName},
				Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
			},
			target:  "not-a-target",
			wantErr: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tikv := newTiKVGroup(t, namespace, tikvId, 2)
			tikv.Spec.Template.Spec.RemoteWorkers = tt.initial

			ctx := context.Background()
			needUpdate, err := getUpdateExpectedTiKVGroup(ctx, tikv, UpdateOption{
				Option: OpsUpdateTiKVRemoteWorkers, Value: tt.target,
			})

			if tt.wantErr {
				require.Error(t, err)
				return
			}
			require.NoError(t, err)
			assert.Equal(t, tt.wantUpdate, needUpdate)
			require.NotNil(t, tikv.Spec.Template.Spec.RemoteWorkers)
			assert.Equal(t, tt.wantWorker, tikv.Spec.Template.Spec.RemoteWorkers.Worker.Name)
			require.NotNil(t, tikv.Spec.Template.Spec.RemoteWorkers.Coprocessor)
			assert.Equal(t, tt.wantCoprocessor, tikv.Spec.Template.Spec.RemoteWorkers.Coprocessor.Name)
		})
	}
}

func TestUpdateTiKVGroupRemoteWorkersPreservesCompactor(t *testing.T) {
	clusterId := "tikv-compactor"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-compactor")
	oldWorkerName := "old-worker"
	oldCoprName := "old-coprocessor"
	newWorkerName := groupNameWithComponent(tikvId, "default")
	newCoprName := groupNameWithComponent(tikvId, "coprocessor")
	compactorName := "compactor-group"

	tikv := newTiKVGroup(t, namespace, tikvId, 2)
	tikv.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
		Worker:      v1alpha1.WorkerReference{Name: oldWorkerName},
		Coprocessor: &v1alpha1.CoprocessorReference{Name: oldCoprName},
		Compactor:   &v1alpha1.CompactorReference{Name: compactorName},
	}

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

	ctx := context.Background()
	groupId := groupIdWithComponent(tikvId, "tikv")
	_, err := UpdateTiKVGroup(ctx, cli, clusterId, groupId, []UpdateOption{
		{Option: OpsUpdateTiKVRemoteWorkers, Value: TiKVRemoteWorkerTarget{Worker: newWorkerName, Coprocessor: newCoprName}},
	})
	require.NoError(t, err)

	stored := &v1alpha1.TiKVGroup{}
	err = cli.Get(ctx, client.ObjectKey{Namespace: namespace, Name: groupNameWithComponent(tikvId, "tikv")}, stored)
	require.NoError(t, err)
	require.NotNil(t, stored.Spec.Template.Spec.RemoteWorkers)
	assert.Equal(t, newWorkerName, stored.Spec.Template.Spec.RemoteWorkers.Worker.Name)
	require.NotNil(t, stored.Spec.Template.Spec.RemoteWorkers.Coprocessor)
	assert.Equal(t, newCoprName, stored.Spec.Template.Spec.RemoteWorkers.Coprocessor.Name)
	require.NotNil(t, stored.Spec.Template.Spec.RemoteWorkers.Compactor)
	assert.Equal(t, compactorName, stored.Spec.Template.Spec.RemoteWorkers.Compactor.Name)
}

func newHealthyTiKVGroup(t *testing.T, namespace, id string, replicas int32) *v1alpha1.TiKVGroup {
	t.Helper()
	group := newTiKVGroup(t, namespace, id, replicas)
	group.Status.CurrentReplicas = replicas
	group.Status.Conditions = []metav1.Condition{
		{Type: v1alpha1.CondReady, Status: metav1.ConditionTrue, ObservedGeneration: group.Generation},
		{Type: v1alpha1.CondSynced, Status: metav1.ConditionTrue, ObservedGeneration: group.Generation},
	}
	return group
}

func TestTiKVGroupRemoteWorkersReady(t *testing.T) {
	namespace := "test-ns"
	tikvId := testComponentId("tikv-ready")
	workerName := groupNameWithComponent(tikvId, "default")
	coprName := groupNameWithComponent(tikvId, "coprocessor")

	tests := []struct {
		name     string
		mutate   func(*v1alpha1.TiKVGroup)
		expected TiKVRemoteWorkerTarget
		want     bool
	}{
		{
			name:     "nil group",
			mutate:   func(g *v1alpha1.TiKVGroup) {},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "empty expected worker",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: "", Coprocessor: coprName},
			want:     false,
		},
		{
			name: "empty expected coprocessor",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: ""},
			want:     false,
		},
		{
			name: "generation mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Generation = 2
				g.Status.ObservedGeneration = 1
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "current revision empty",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.CurrentRevision = ""
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "update revision empty",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.UpdateRevision = ""
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "current and update revision mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.CurrentRevision = "rev-old"
				g.Status.UpdateRevision = "rev-new"
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "spec replicas nil",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Replicas = nil
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "status replicas not equal spec replicas",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.Replicas = 2
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "ready replicas mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.ReadyReplicas = 2
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "updated replicas mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.UpdatedReplicas = 2
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "current replicas mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.CurrentReplicas = 2
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "ready condition not true",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.Conditions = []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionFalse, ObservedGeneration: g.Generation},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionTrue, ObservedGeneration: g.Generation},
				}
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "synced condition not true",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Status.Conditions = []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionTrue, ObservedGeneration: g.Generation},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionFalse, ObservedGeneration: g.Generation},
				}
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "ready condition stale observed generation",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Generation = 2
				g.Status.ObservedGeneration = 2
				g.Status.Conditions = []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionTrue, ObservedGeneration: 1},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionTrue, ObservedGeneration: 2},
				}
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "synced condition stale observed generation",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Generation = 2
				g.Status.ObservedGeneration = 2
				g.Status.Conditions = []metav1.Condition{
					{Type: v1alpha1.CondReady, Status: metav1.ConditionTrue, ObservedGeneration: 2},
					{Type: v1alpha1.CondSynced, Status: metav1.ConditionTrue, ObservedGeneration: 1},
				}
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "worker name mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: "wrong-worker"},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "coprocessor name mismatch",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: "wrong-coprocessor"},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     false,
		},
		{
			name: "all conditions met",
			mutate: func(g *v1alpha1.TiKVGroup) {
				g.Spec.Template.Spec.RemoteWorkers = &v1alpha1.TiKVRemoteWorkers{
					Worker:      v1alpha1.WorkerReference{Name: workerName},
					Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
				}
			},
			expected: TiKVRemoteWorkerTarget{Worker: workerName, Coprocessor: coprName},
			want:     true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			var group *v1alpha1.TiKVGroup
			if tt.name != "nil group" {
				group = newHealthyTiKVGroup(t, namespace, tikvId, 3)
				tt.mutate(group)
			}
			got := TiKVGroupRemoteWorkersReady(group, tt.expected)
			assert.Equal(t, tt.want, got)
		})
	}
}

func TestGetTiKVGroupRemoteWorkerNames(t *testing.T) {
	clusterId := "tikv-remote-names"
	namespace := dataplaneNamespace(clusterId)
	tikvId := testComponentId("tikv-remote-names")
	workerName := groupNameWithComponent(tikvId, "default")
	coprName := groupNameWithComponent(tikvId, "coprocessor")

	tests := []struct {
		name       string
		remote     *v1alpha1.TiKVRemoteWorkers
		wantWorker string
		wantCopr   string
	}{
		{
			name:       "nil remote workers",
			remote:     nil,
			wantWorker: "",
			wantCopr:   "",
		},
		{
			name: "worker only",
			remote: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: workerName},
				Coprocessor: nil,
			},
			wantWorker: workerName,
			wantCopr:   "",
		},
		{
			name: "both worker and coprocessor",
			remote: &v1alpha1.TiKVRemoteWorkers{
				Worker:      v1alpha1.WorkerReference{Name: workerName},
				Coprocessor: &v1alpha1.CoprocessorReference{Name: coprName},
			},
			wantWorker: workerName,
			wantCopr:   coprName,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			tikv := newTiKVGroup(t, namespace, tikvId, 2)
			tikv.Spec.Template.Spec.RemoteWorkers = tt.remote

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

			ctx := context.Background()
			groupId := groupIdWithComponent(tikvId, "tikv")
			got, err := GetTiKVGroupRemoteWorkerNames(ctx, cli, clusterId, groupId)
			require.NoError(t, err)
			assert.Equal(t, tt.wantWorker, got.Worker)
			assert.Equal(t, tt.wantCopr, got.Coprocessor)
		})
	}
}

func TestGetTiKVGroupRemoteWorkerNamesErrors(t *testing.T) {
	clusterId := "tikv-remote-names-err"
	tikvId := testComponentId("tikv-remote-names-err")
	groupId := groupIdWithComponent(tikvId, "tikv")
	ctx := context.Background()

	t.Run("NotFound", func(t *testing.T) {
		scheme := buildTestScheme(t)
		cli := fake.NewClientBuilder().WithScheme(scheme).Build()

		target, err := GetTiKVGroupRemoteWorkerNames(ctx, cli, clusterId, groupId)
		require.Error(t, err)
		st, ok := status.FromError(err)
		require.True(t, ok)
		assert.Equal(t, codes.NotFound, st.Code())
		assert.Equal(t, TiKVRemoteWorkerTarget{}, target)
	})

	t.Run("Internal", func(t *testing.T) {
		scheme := buildTestScheme(t)
		cli := fake.NewClientBuilder().
			WithScheme(scheme).
			WithInterceptorFuncs(interceptor.Funcs{
				Get: func(ctx context.Context, c client.WithWatch, key client.ObjectKey, obj client.Object, opts ...client.GetOption) error {
					return errors.New("k8s api error")
				},
			}).
			Build()

		target, err := GetTiKVGroupRemoteWorkerNames(ctx, cli, clusterId, groupId)
		require.Error(t, err)
		st, ok := status.FromError(err)
		require.True(t, ok)
		assert.Equal(t, codes.Internal, st.Code())
		assert.Equal(t, TiKVRemoteWorkerTarget{}, target)
	})
}