package tioperator

import (
	"context"
	"fmt"
	"testing"

	"github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	meta "github.com/pingcap/tidb-operator/api/v2/meta/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"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

func mustDMGroupNameV2(t *testing.T, clusterID string) string {
	t.Helper()
	name, err := entity.DMGroupNameV2(clusterID)
	require.NoError(t, err)
	return name
}

func TestGetDMClusterV2(t *testing.T) {
	clusterId := "dm-v2"
	dmCluster := testDMClusterV2Entity(clusterId, 3, 2)
	dmCluster.DataplaneName = "test-dataplane"
	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())

	replicasMaster := int32(3)
	replicasWorker := int32(2)
	master := &v1alpha1.DMGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:       mustDMGroupNameV2(t, dmCluster.GetId()),
			Namespace:  namespace,
			Generation: 1,
			Labels: map[string]string{
				schemav1.LabelNodeSpecTypeKey: "8C16G",
			},
		},
		Spec: v1alpha1.DMGroupSpec{
			Replicas: &replicasMaster,
			Template: v1alpha1.DMTemplate{
				Spec: v1alpha1.DMTemplateSpec{
					Overlay: &v1alpha1.Overlay{
						Pod: &v1alpha1.PodOverlay{
							Spec: testDMV2PodSpec(
								map[string]string{
									"node-spec":                 "8C16G",
									schemav1.K8sVersionLabelKey: "1.31",
								},
								dmMasterContainer,
								corev1.ResourceRequirements{
									Limits: corev1.ResourceList{
										corev1.ResourceCPU:    resource.MustParse("6"),
										corev1.ResourceMemory: resource.MustParse("12Gi"),
									},
								},
							),
						},
					},
				},
			},
		},
		Status: v1alpha1.DMGroupStatus{
			CommonStatus: v1alpha1.CommonStatus{
				ObservedGeneration: 1,
				CurrentRevision:    "rev1",
				UpdateRevision:     "rev1",
				Conditions:         readySyncedConditions(),
			},
			GroupStatus: v1alpha1.GroupStatus{
				ReadyReplicas:   replicasMaster,
				UpdatedReplicas: replicasMaster,
			},
		},
	}

	worker := &v1alpha1.DMWorkerGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:       mustDMGroupNameV2(t, dmCluster.GetId()),
			Namespace:  namespace,
			Generation: 1,
			Labels: map[string]string{
				schemav1.LabelNodeSpecTypeKey: "4C8G",
			},
		},
		Spec: v1alpha1.DMWorkerGroupSpec{
			Replicas: &replicasWorker,
			Template: v1alpha1.DMWorkerTemplate{
				Spec: v1alpha1.DMWorkerTemplateSpec{
					Overlay: &v1alpha1.Overlay{
						Pod: &v1alpha1.PodOverlay{
							Spec: testDMV2PodSpec(
								map[string]string{
									"node-spec":                 "4C8G",
									schemav1.K8sVersionLabelKey: "1.30",
								},
								dmWorkerContainer,
								corev1.ResourceRequirements{
									Limits: corev1.ResourceList{
										corev1.ResourceCPU:    resource.MustParse("3"),
										corev1.ResourceMemory: resource.MustParse("6Gi"),
									},
								},
							),
						},
					},
				},
			},
		},
		Status: v1alpha1.DMWorkerGroupStatus{
			CommonStatus: v1alpha1.CommonStatus{
				ObservedGeneration: 1,
				CurrentRevision:    "rev1",
				UpdateRevision:     "rev1",
				Conditions:         readySyncedConditions(),
			},
			GroupStatus: v1alpha1.GroupStatus{
				ReadyReplicas:   replicasWorker,
				UpdatedReplicas: replicasWorker,
			},
		},
	}

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(master, worker).Build()

	got, err := GetDMClusterV2(context.Background(), cli, dmCluster)
	require.NoError(t, err)
	require.NotNil(t, got)
	assert.Equal(t, apiv1.DMCluster_AVAILABLE, got.GetPhase())
	assert.Equal(t, dmCluster.GetSharedPoolId(), got.GetSharedPoolId())
	assert.Equal(t, "test-dataplane", got.GetDataplaneName())
	assert.Equal(t, uint32(3), got.GetMasterNodeQuota().GetNodeCount())
	assert.Equal(t, uint32(3), got.GetMasterNodeQuota().GetMinNodeCount())
	assert.Equal(t, uint32(3), got.GetMasterNodeQuota().GetMaxNodeCount())
	assert.Equal(t, uint32(8), got.GetMasterNodeQuota().GetNodeSpec().GetCpu())
	assert.Equal(t, uint32(16), got.GetMasterNodeQuota().GetNodeSpec().GetMemoryGi())
	assert.Equal(t, "1.31", got.GetMasterNodeQuota().GetNodeSpec().GetNodeVersion())
	assert.Equal(t, uint32(2), got.GetWorkerNodeQuota().GetNodeCount())
	assert.Equal(t, uint32(2), got.GetWorkerNodeQuota().GetMinNodeCount())
	assert.Equal(t, uint32(2), got.GetWorkerNodeQuota().GetMaxNodeCount())
	assert.Equal(t, uint32(4), got.GetWorkerNodeQuota().GetNodeSpec().GetCpu())
	assert.Equal(t, uint32(8), got.GetWorkerNodeQuota().GetNodeSpec().GetMemoryGi())
	assert.Equal(t, "1.30", got.GetWorkerNodeQuota().GetNodeSpec().GetNodeVersion())
}

func TestApplyDMClusterV2(t *testing.T) {
	t.Run("create resources", func(t *testing.T) {
		scheme := buildTestScheme(t)
		dmCluster := testDMClusterV2Entity("create", 3, 2)
		namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
		cluster := testClusterWithFeatureGates(namespace)
		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()

		err := ApplyDMClusterV2(context.Background(), cli, dmCluster, testDMVersion(), testDMImage(), false)
		require.NoError(t, err)

		master := &v1alpha1.DMGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{
			Namespace: namespace,
			Name:      mustDMGroupNameV2(t, dmCluster.GetId()),
		}, master)
		require.NoError(t, err)
		require.NotNil(t, master.Spec.Replicas)
		assert.Equal(t, int32(3), *master.Spec.Replicas)
		assert.Equal(t, "v8.5.0", master.Spec.Template.Spec.Version)
		require.NotNil(t, master.Spec.Template.Spec.Image)
		assert.Equal(t, testDMImage(), *master.Spec.Template.Spec.Image)
		assert.Equal(t, clusterCRName, master.Spec.Cluster.Name)
		assert.Equal(t, testDMClusterFeatureGates(), master.Spec.Features)
		assert.Equal(t, "8C16G", master.Labels[schemav1.LabelNodeSpecTypeKey])
		require.Len(t, master.Spec.Template.Spec.Overlay.Pod.Spec.Containers, 1)
		assert.Equal(t, dmMasterContainer, master.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Name)
		assertMainContainerResources(t, master.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Resources, dmCluster.GetMasterNodeQuota().GetNodeSpec(), apiv1.ComponentType_DATA_MIGRATION)

		worker := &v1alpha1.DMWorkerGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{
			Namespace: namespace,
			Name:      mustDMGroupNameV2(t, dmCluster.GetId()),
		}, worker)
		require.NoError(t, err)
		require.NotNil(t, worker.Spec.Replicas)
		assert.Equal(t, int32(2), *worker.Spec.Replicas)
		assert.Equal(t, mustDMGroupNameV2(t, dmCluster.GetId()), worker.Spec.DMGroupRef.Name)
		assert.Equal(t, clusterCRName, worker.Spec.Cluster.Name)
		assert.Equal(t, testDMClusterFeatureGates(), worker.Spec.Features)
		assert.Equal(t, "4C8G", worker.Labels[schemav1.LabelNodeSpecTypeKey])
		require.Len(t, worker.Spec.Template.Spec.Overlay.Pod.Spec.Containers, 1)
		assert.Equal(t, dmWorkerContainer, worker.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Name)
		assertMainContainerResources(t, worker.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Resources, dmCluster.GetWorkerNodeQuota().GetNodeSpec(), apiv1.ComponentType_DATA_MIGRATION)
	})

	t.Run("update resources", func(t *testing.T) {
		scheme := buildTestScheme(t)
		dmCluster := testDMClusterV2Entity("update", 4, 1)
		namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
		ownerUID := types.UID("owner-uid")
		cluster := testClusterWithFeatureGates(namespace)

		currentMaster := &v1alpha1.DMGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name:        mustDMGroupNameV2(t, dmCluster.GetId()),
				Namespace:   namespace,
				Labels:      map[string]string{"managed-by": "operator"},
				Annotations: map[string]string{"owner": "user"},
				Finalizers:  []string{"test.finalizer/master"},
				OwnerReferences: []metav1.OwnerReference{{
					APIVersion: "v1",
					Kind:       "ConfigMap",
					Name:       "master-owner",
					UID:        ownerUID,
				}},
			},
			Spec: v1alpha1.DMGroupSpec{Replicas: int32Ptr(1)},
		}
		currentWorker := &v1alpha1.DMWorkerGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name:        mustDMGroupNameV2(t, dmCluster.GetId()),
				Namespace:   namespace,
				Labels:      map[string]string{"managed-by": "operator"},
				Annotations: map[string]string{"owner": "user"},
				Finalizers:  []string{"test.finalizer/worker"},
				OwnerReferences: []metav1.OwnerReference{{
					APIVersion: "v1",
					Kind:       "ConfigMap",
					Name:       "worker-owner",
					UID:        ownerUID,
				}},
			},
			Spec: v1alpha1.DMWorkerGroupSpec{Replicas: int32Ptr(1)},
		}
		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, currentMaster, currentWorker).Build()

		err := ApplyDMClusterV2(context.Background(), cli, dmCluster, testDMVersion(), testDMImage(), true)
		require.NoError(t, err)

		master := &v1alpha1.DMGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: mustDMGroupNameV2(t, dmCluster.GetId())}, master)
		require.NoError(t, err)
		require.NotNil(t, master.Spec.Replicas)
		assert.Equal(t, int32(4), *master.Spec.Replicas)
		assert.Equal(t, clusterCRName, master.Spec.Cluster.Name)
		assert.Equal(t, map[string]string{
			"managed-by":                  "operator",
			schemav1.LabelNodeSpecTypeKey: "8C16G",
		}, master.Labels)
		assert.Equal(t, map[string]string{"owner": "user"}, master.Annotations)
		assert.Equal(t, []string{"test.finalizer/master"}, master.Finalizers)
		require.Len(t, master.OwnerReferences, 1)
		assert.Equal(t, "master-owner", master.OwnerReferences[0].Name)
		assert.Equal(t, ownerUID, master.OwnerReferences[0].UID)

		worker := &v1alpha1.DMWorkerGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: mustDMGroupNameV2(t, dmCluster.GetId())}, worker)
		require.NoError(t, err)
		require.NotNil(t, worker.Spec.Replicas)
		assert.Equal(t, int32(1), *worker.Spec.Replicas)
		assert.Equal(t, clusterCRName, worker.Spec.Cluster.Name)
		assert.Equal(t, map[string]string{
			"managed-by":                  "operator",
			schemav1.LabelNodeSpecTypeKey: "4C8G",
		}, worker.Labels)
		assert.Equal(t, map[string]string{"owner": "user"}, worker.Annotations)
		assert.Equal(t, []string{"test.finalizer/worker"}, worker.Finalizers)
		require.Len(t, worker.OwnerReferences, 1)
		assert.Equal(t, "worker-owner", worker.OwnerReferences[0].Name)
		assert.Equal(t, ownerUID, worker.OwnerReferences[0].UID)
	})

	t.Run("preserve image and version when update image disabled", func(t *testing.T) {
		scheme := buildTestScheme(t)
		dmCluster := testDMClusterV2Entity("preserve-image", 4, 1)
		namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
		cluster := testClusterWithFeatureGates(namespace)
		oldMasterImage := "repo/dm-master:v8.4.0"
		oldWorkerImage := "repo/dm-worker:v8.4.0"
		currentMaster := &v1alpha1.DMGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      mustDMGroupNameV2(t, dmCluster.GetId()),
				Namespace: namespace,
			},
			Spec: v1alpha1.DMGroupSpec{
				Replicas: int32Ptr(1),
				Template: v1alpha1.DMTemplate{
					Spec: v1alpha1.DMTemplateSpec{
						Version: "v8.4.0",
						Image:   &oldMasterImage,
					},
				},
			},
		}
		currentWorker := &v1alpha1.DMWorkerGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name:      mustDMGroupNameV2(t, dmCluster.GetId()),
				Namespace: namespace,
			},
			Spec: v1alpha1.DMWorkerGroupSpec{
				Replicas: int32Ptr(1),
				Template: v1alpha1.DMWorkerTemplate{
					Spec: v1alpha1.DMWorkerTemplateSpec{
						Version: "v8.4.0",
						Image:   &oldWorkerImage,
					},
				},
			},
		}
		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, currentMaster, currentWorker).Build()

		err := ApplyDMClusterV2(context.Background(), cli, dmCluster, testDMVersion(), testDMImage(), false)
		require.NoError(t, err)

		master := &v1alpha1.DMGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: mustDMGroupNameV2(t, dmCluster.GetId())}, master)
		require.NoError(t, err)
		assert.Equal(t, "v8.4.0", master.Spec.Template.Spec.Version)
		require.NotNil(t, master.Spec.Template.Spec.Image)
		assert.Equal(t, oldMasterImage, *master.Spec.Template.Spec.Image)
		require.NotNil(t, master.Spec.Replicas)
		assert.Equal(t, int32(4), *master.Spec.Replicas)

		worker := &v1alpha1.DMWorkerGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: mustDMGroupNameV2(t, dmCluster.GetId())}, worker)
		require.NoError(t, err)
		assert.Equal(t, "v8.4.0", worker.Spec.Template.Spec.Version)
		require.NotNil(t, worker.Spec.Template.Spec.Image)
		assert.Equal(t, oldWorkerImage, *worker.Spec.Template.Spec.Image)
		require.NotNil(t, worker.Spec.Replicas)
		assert.Equal(t, int32(1), *worker.Spec.Replicas)
	})

	t.Run("update image and version when update image enabled", func(t *testing.T) {
		scheme := buildTestScheme(t)
		dmCluster := testDMClusterV2Entity("update-image", 4, 1)
		namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
		cluster := testClusterWithFeatureGates(namespace)
		oldImage := "repo/dm:v8.4.0"
		currentMaster := &v1alpha1.DMGroup{
			ObjectMeta: metav1.ObjectMeta{Name: mustDMGroupNameV2(t, dmCluster.GetId()), Namespace: namespace},
			Spec: v1alpha1.DMGroupSpec{
				Template: v1alpha1.DMTemplate{Spec: v1alpha1.DMTemplateSpec{Version: "v8.4.0", Image: &oldImage}},
			},
		}
		currentWorker := &v1alpha1.DMWorkerGroup{
			ObjectMeta: metav1.ObjectMeta{Name: mustDMGroupNameV2(t, dmCluster.GetId()), Namespace: namespace},
			Spec: v1alpha1.DMWorkerGroupSpec{
				Template: v1alpha1.DMWorkerTemplate{Spec: v1alpha1.DMWorkerTemplateSpec{Version: "v8.4.0", Image: &oldImage}},
			},
		}
		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, currentMaster, currentWorker).Build()

		err := ApplyDMClusterV2(context.Background(), cli, dmCluster, testDMVersion(), testDMImage(), true)
		require.NoError(t, err)

		master := &v1alpha1.DMGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: mustDMGroupNameV2(t, dmCluster.GetId())}, master)
		require.NoError(t, err)
		assert.Equal(t, testDMVersion(), master.Spec.Template.Spec.Version)
		require.NotNil(t, master.Spec.Template.Spec.Image)
		assert.Equal(t, testDMImage(), *master.Spec.Template.Spec.Image)

		worker := &v1alpha1.DMWorkerGroup{}
		err = cli.Get(context.Background(), types.NamespacedName{Namespace: namespace, Name: mustDMGroupNameV2(t, dmCluster.GetId())}, worker)
		require.NoError(t, err)
		assert.Equal(t, testDMVersion(), worker.Spec.Template.Spec.Version)
		require.NotNil(t, worker.Spec.Template.Spec.Image)
		assert.Equal(t, testDMImage(), *worker.Spec.Template.Spec.Image)
	})

	t.Run("create multiple dm clusters in same shared pool", func(t *testing.T) {
		scheme := buildTestScheme(t)
		first := testDMClusterV2Entity("first", 3, 2)
		second := testDMClusterV2Entity("second", 1, 1)
		second.SharedPoolId = first.GetSharedPoolId()
		namespace := entity.GetDataPlaneNamespace(first.GetSharedPoolId())
		cluster := testClusterWithFeatureGates(namespace)
		cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()

		require.NoError(t, ApplyDMClusterV2(context.Background(), cli, first, testDMVersion(), testDMImage(), false))
		require.NoError(t, ApplyDMClusterV2(context.Background(), cli, second, testDMVersion(), testDMImage(), false))

		for _, dmCluster := range []*entity.DMCluster{first, second} {
			master := &v1alpha1.DMGroup{}
			err := cli.Get(context.Background(), types.NamespacedName{
				Namespace: namespace,
				Name:      mustDMGroupNameV2(t, dmCluster.GetId()),
			}, master)
			require.NoError(t, err)
			assert.Equal(t, clusterCRName, master.Spec.Cluster.Name)

			worker := &v1alpha1.DMWorkerGroup{}
			err = cli.Get(context.Background(), types.NamespacedName{
				Namespace: namespace,
				Name:      mustDMGroupNameV2(t, dmCluster.GetId()),
			}, worker)
			require.NoError(t, err)
			assert.Equal(t, mustDMGroupNameV2(t, dmCluster.GetId()), worker.Spec.DMGroupRef.Name)
		}
	})

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

		err := ApplyDMClusterV2(context.Background(), cli, nil, testDMVersion(), testDMImage(), false)
		require.EqualError(t, err, "dm cluster is nil")
	})

	t.Run("missing shared pool cluster", func(t *testing.T) {
		scheme := buildTestScheme(t)
		cli := fake.NewClientBuilder().WithScheme(scheme).Build()
		dmCluster := testDMClusterV2Entity("missing-cluster", 3, 2)

		err := ApplyDMClusterV2(context.Background(), cli, dmCluster, testDMVersion(), testDMImage(), false)
		require.EqualError(t, err, fmt.Sprintf("rpc error: code = NotFound desc = shared pool cluster %s not found in namespace %s", clusterCRName, entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())))
	})
}

func TestDmClusterV2Namespace(t *testing.T) {
	assert.Empty(t, dmClusterV2Namespace(nil))

	emptyPool := &entity.DMCluster{Id: "dm-1"}
	assert.Empty(t, dmClusterV2Namespace(emptyPool))

	dmCluster := testDMClusterV2Entity("dm-1", 1, 1)
	assert.Equal(t, entity.GetDMV2Namespace(dmCluster.GetSharedPoolId()), dmClusterV2Namespace(dmCluster))
}

func TestDeleteDMClusterV2(t *testing.T) {
	dmCluster := testDMClusterV2Entity("dm-v2-delete", 3, 2)
	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())

	ns := &corev1.Namespace{
		ObjectMeta: metav1.ObjectMeta{Name: namespace},
	}

	master := &v1alpha1.DMGroup{
		ObjectMeta: metav1.ObjectMeta{Name: mustDMGroupNameV2(t, dmCluster.GetId()), Namespace: namespace},
	}
	worker := &v1alpha1.DMWorkerGroup{
		ObjectMeta: metav1.ObjectMeta{Name: mustDMGroupNameV2(t, dmCluster.GetId()), Namespace: namespace},
	}

	scheme := buildTestScheme(t)
	cli := fake.NewClientBuilder().WithScheme(scheme).WithObjects(ns, master, worker).Build()

	err := DeleteDMClusterV2(context.Background(), cli, dmCluster)
	require.NoError(t, err)

	for _, obj := range []client.Object{
		&v1alpha1.DMGroup{ObjectMeta: metav1.ObjectMeta{Name: mustDMGroupNameV2(t, dmCluster.GetId()), Namespace: namespace}},
		&v1alpha1.DMWorkerGroup{ObjectMeta: metav1.ObjectMeta{Name: mustDMGroupNameV2(t, dmCluster.GetId()), Namespace: namespace}},
	} {
		getErr := cli.Get(context.Background(), client.ObjectKeyFromObject(obj), obj)
		assert.True(t, apierrors.IsNotFound(getErr))
	}

	namespaceObj := &corev1.Namespace{}
	getErr := cli.Get(context.Background(), client.ObjectKey{Name: namespace}, namespaceObj)
	require.NoError(t, getErr)
}

func TestBuildDMGroupV2(t *testing.T) {
	dmCluster := testDMClusterV2Entity("build", 3, 2)
	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())

	group, err := buildDMGroupV2(dmCluster, namespace, clusterCRName, testDMClusterFeatureGates(), testDMVersion(), testDMImage())
	require.NoError(t, err)

	assert.Equal(t, namespace, group.Namespace)
	assert.Equal(t, mustDMGroupNameV2(t, dmCluster.GetId()), group.Name)
	require.NotNil(t, group.Spec.Replicas)
	assert.Equal(t, int32(3), *group.Spec.Replicas)
	assert.Equal(t, clusterCRName, group.Spec.Cluster.Name)
	assert.Equal(t, "v8.5.0", group.Spec.Template.Spec.Version)
	require.NotNil(t, group.Spec.Template.Spec.Image)
	assert.Equal(t, testDMImage(), *group.Spec.Template.Spec.Image)
	require.NotNil(t, group.Spec.Template.Spec.DataVolume)
	assert.Equal(t, dmMasterDataVolume, group.Spec.Template.Spec.DataVolume.Name)
	assert.Equal(t, "8C16G", group.Labels[schemav1.LabelNodeSpecTypeKey])
	assert.Equal(t, "8C16G", group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector["node-spec"])
	assert.Equal(t, "1.31", group.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector[schemav1.K8sVersionLabelKey])
	require.Len(t, group.Spec.Template.Spec.Overlay.Pod.Spec.Containers, 1)
	assert.Equal(t, dmMasterContainer, group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Name)
	assertMainContainerResources(t, group.Spec.Template.Spec.Overlay.Pod.Spec.Containers[0].Resources, dmCluster.GetMasterNodeQuota().GetNodeSpec(), apiv1.ComponentType_DATA_MIGRATION)
}

func TestDMGroupNameV2(t *testing.T) {
	shortClusterId := "shortid"
	assert.Equal(t, "dbcluster-shortid", mustDMGroupNameV2(t, shortClusterId))

	longClusterId := "abcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyzabcdefghijklmnopqrstuvwxyz"
	_, err := entity.DMGroupNameV2(longClusterId)
	require.Error(t, err)
}

func TestDetermineDMClusterPhaseV2(t *testing.T) {
	tests := []struct {
		name   string
		master *v1alpha1.DMGroup
		worker *v1alpha1.DMWorkerGroup
		want   apiv1.DMCluster_Phase
	}{
		{
			name:   "available with ready and synced conditions",
			master: testDMGroupForPhase(1, 1, readySyncedConditions(), 3, 3, "rev1", "rev1"),
			worker: testDMWorkerGroupForPhase(readySyncedConditions(), 2, 2),
			want:   apiv1.DMCluster_AVAILABLE,
		},
		{
			name:   "creating when not observed yet",
			master: testDMGroupForPhase(1, 0, syncedNotReadyConditions(), 0, 0, "", ""),
			worker: testDMWorkerGroupForPhase(syncedNotReadyConditions(), 0, 0),
			want:   apiv1.DMCluster_CREATING,
		},
		{
			name:   "modifying when generation not up to date",
			master: testDMGroupForPhase(2, 1, readySyncedConditions(), 3, 3, "rev1", "rev2"),
			worker: testDMWorkerGroupForPhase(readySyncedConditions(), 2, 2),
			want:   apiv1.DMCluster_MODIFYING,
		},
		{
			name:   "modifying when replicas not ready",
			master: testDMGroupForPhase(1, 1, readySyncedConditions(), 2, 3, "rev1", "rev1"),
			worker: testDMWorkerGroupForPhase(readySyncedConditions(), 2, 2),
			want:   apiv1.DMCluster_MODIFYING,
		},
		{
			name:   "unavailable when up to date but not ready",
			master: testDMGroupForPhase(1, 1, syncedNotReadyConditions(), 3, 3, "rev1", "rev1"),
			worker: testDMWorkerGroupForPhase(syncedNotReadyConditions(), 2, 2),
			want:   apiv1.DMCluster_UNAVAILABLE,
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			assert.Equal(t, tc.want, determineDMClusterPhaseV2(tc.master, tc.worker))
		})
	}
}

func testDMV2PodSpec(nodeSelector map[string]string, containerName string, resources corev1.ResourceRequirements) *corev1.PodSpec {
	return &corev1.PodSpec{
		NodeSelector: nodeSelector,
		Containers: []corev1.Container{
			{
				Name:      containerName,
				Resources: resources,
			},
		},
	}
}

func testDMClusterV2Entity(clusterId string, masterCount, workerCount uint32) *entity.DMCluster {
	return &entity.DMCluster{
		Id:           clusterId,
		SharedPoolId: "shared-pool-" + clusterId,
		MasterNodeQuota: &entity.NodeQuota{
			NodeCount:    masterCount,
			MinNodeCount: masterCount,
			MaxNodeCount: masterCount,
			NodeSpec: &entity.NodeSpec{
				Cpu:               8,
				MemoryGi:          16,
				AvailableCpu:      6,
				AvailableMemoryMi: 12288,
				NodeVersion:       "1.31",
			},
		},
		WorkerNodeQuota: &entity.NodeQuota{
			NodeCount:    workerCount,
			MinNodeCount: workerCount,
			MaxNodeCount: workerCount,
			NodeSpec: &entity.NodeSpec{
				Cpu:               4,
				MemoryGi:          8,
				AvailableCpu:      3,
				AvailableMemoryMi: 6144,
				NodeVersion:       "1.30",
			},
		},
	}
}

func testDMVersion() string {
	return "v8.5.0"
}

func testDMImage() string {
	return "repo/dm:v8.5.0-hotfix"
}

func testDMClusterFeatureGates() []meta.Feature {
	return []meta.Feature{
		meta.FeatureModification,
		meta.DisablePDDefaultReadinessProbe,
		meta.SessionTokenSigning,
		meta.ClusterSubdomain,
		meta.TerminableLogTailer,
		meta.UsePDReadyAPIV2,
		meta.UseTSOReadyAPI,
		meta.UseSchedulingReadyAPI,
		meta.UseTiKVReadyAPI,
		meta.UseTiFlashReadyAPI,
		meta.VolumeAttributesClass,
		meta.IndependentKVEngineWorker,
	}
}

func testDMClusterFeatureGateSpecs() []meta.FeatureGate {
	return []meta.FeatureGate{
		{Name: meta.FeatureModification},
		{Name: meta.DisablePDDefaultReadinessProbe},
		{Name: meta.SessionTokenSigning},
		{Name: meta.ClusterSubdomain},
		{Name: meta.TerminableLogTailer},
		{Name: meta.UsePDReadyAPIV2},
		{Name: meta.UseTSOReadyAPI},
		{Name: meta.UseSchedulingReadyAPI},
		{Name: meta.UseTiKVReadyAPI},
		{Name: meta.UseTiFlashReadyAPI},
		{Name: meta.VolumeAttributesClass},
		{Name: meta.IndependentKVEngineWorker},
	}
}

func testClusterWithFeatureGates(namespace string) *v1alpha1.Cluster {
	return &v1alpha1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      clusterCRName,
			Namespace: namespace,
		},
		Spec: v1alpha1.ClusterSpec{
			FeatureGates: testDMClusterFeatureGateSpecs(),
		},
	}
}

func TestFeatureGatesToFeatures(t *testing.T) {
	assert.Nil(t, featureGatesToFeatures(nil))
	assert.Equal(t, testDMClusterFeatureGates(), featureGatesToFeatures(testDMClusterFeatureGateSpecs()))
}

func testDMGroupForPhase(
	generation int64,
	observedGeneration int64,
	conditions []metav1.Condition,
	readyReplicas int32,
	updatedReplicas int32,
	currentRevision string,
	updateRevision string,
) *v1alpha1.DMGroup {
	replicas := int32(3)
	return &v1alpha1.DMGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:       "dbcluster-test",
			Namespace:  "test",
			Generation: generation,
		},
		Spec: v1alpha1.DMGroupSpec{
			Replicas: &replicas,
		},
		Status: v1alpha1.DMGroupStatus{
			CommonStatus: v1alpha1.CommonStatus{
				ObservedGeneration: observedGeneration,
				CurrentRevision:    currentRevision,
				UpdateRevision:     updateRevision,
				Conditions:         conditions,
			},
			GroupStatus: v1alpha1.GroupStatus{
				ReadyReplicas:   readyReplicas,
				UpdatedReplicas: updatedReplicas,
			},
		},
	}
}

func testDMWorkerGroupForPhase(
	conditions []metav1.Condition,
	readyReplicas int32,
	updatedReplicas int32,
) *v1alpha1.DMWorkerGroup {
	replicas := int32(2)
	generation := int64(1)
	observedGeneration := int64(1)
	currentRevision := "rev1"
	updateRevision := "rev1"
	return &v1alpha1.DMWorkerGroup{
		ObjectMeta: metav1.ObjectMeta{
			Name:       "dbcluster-test",
			Namespace:  "test",
			Generation: generation,
		},
		Spec: v1alpha1.DMWorkerGroupSpec{
			Replicas: &replicas,
		},
		Status: v1alpha1.DMWorkerGroupStatus{
			CommonStatus: v1alpha1.CommonStatus{
				ObservedGeneration: observedGeneration,
				CurrentRevision:    currentRevision,
				UpdateRevision:     updateRevision,
				Conditions:         conditions,
			},
			GroupStatus: v1alpha1.GroupStatus{
				ReadyReplicas:   readyReplicas,
				UpdatedReplicas: updatedReplicas,
			},
		},
	}
}

func int32Ptr(v int32) *int32 {
	return &v
}