package action

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strconv"
	"testing"

	certmanagerv1 "github.com/cert-manager/cert-manager/pkg/apis/certmanager/v1"
	v2 "github.com/pingcap/tidb-operator/api/v2/core/v1alpha1"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/config"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	"github.com/tidbcloud/cluster-service-ng/pkg/cert"
	clusterconfigmocks "github.com/tidbcloud/cluster-service-ng/pkg/clusterconfig/mocks"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/template/render"
	"github.com/tidbcloud/cluster-service-ng/pkg/tioperator"
	"google.golang.org/protobuf/encoding/protojson"
	corev1 "k8s.io/api/core/v1"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	"k8s.io/apimachinery/pkg/types"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
)

var (
	dmMasterTestTemplArgs = map[string]string{
		"ClusterId":        "test-dm-cluster",
		"SharedPoolId":     "test-shared-pool",
		"Namespace":        "dmtest-shared-pool",
		"ComponentGroup":   "dbcluster-dm-master",
		"Replicas":         "3",
		"NodeSpec":         "8C16G",
		"SetResource":      "true",
		"CPU":              "7800m",
		"LimitCPU":         "8000m",
		"Memory":           "16384",
		"RequestMemory":    "15360",
		"Image":            "pingcap/dm:v8.1.0",
		"ImagePullPolicy":  "IfNotPresent",
		"NodeVersion":      "v1.0.0",
		"StorageClassName": "cloud-ssd",
		"StorageSize":      "50",
	}

	dmWorkerTestTemplArgs = map[string]string{
		"ClusterId":           "test-dm-cluster",
		"SharedPoolId":        "test-shared-pool",
		"Namespace":           "dmtest-shared-pool",
		"ComponentGroup":      "dbcluster-dm-worker",
		"Replicas":            "2",
		"NodeSpec":            "4C8G",
		"SetResource":         "true",
		"CPU":                 "3800m",
		"LimitCPU":            "4000m",
		"Memory":              "8192",
		"RequestMemory":       "7782",
		"Image":               "pingcap/dm:v8.1.0",
		"ImagePullPolicy":     "IfNotPresent",
		"NodeVersion":         "v1.0.0",
		"DmMasterServiceName": "dbcluster-dm-master",
	}
)

func TestApplyDMClusterActionName(t *testing.T) {
	action := &ApplyDMClusterAction{}
	assert.Equal(t, ApplyDMClusterActionName, action.Name())
	assert.Equal(t, "apply-dm-cluster-action", action.Name())
}

func TestBuildDMTemplateArgs(t *testing.T) {
	tests := []struct {
		name           string
		dmCluster      *entity.DMCluster
		nodeQuota      *entity.NodeQuota
		nodeVersionMap map[string]string
		expectedError  string
		validate       func(t *testing.T, args map[string]string)
	}{
		{
			name: "successful build template args with available resources",
			dmCluster: &entity.DMCluster{
				Id:           "test-dm-cluster",
				SharedPoolId: "test-shared-pool",
			},
			nodeQuota: &apiv1.NodeQuota{
				NodeCount: 3,
				NodeSpec: &apiv1.NodeSpec{
					Cpu:               8,
					MemoryGi:          16,
					AvailableCpu:      7,
					AvailableMemoryMi: 14336,
				},
			},
			nodeVersionMap: map[string]string{
				"8C16G": "v1.0.0",
			},
			validate: func(t *testing.T, args map[string]string) {
				t.Helper()
				assert.Equal(t, "test-dm-cluster", args["ClusterId"])
				assert.Equal(t, "test-shared-pool", args["SharedPoolId"])
				assert.Equal(t, "dmtest-shared-pool", args["Namespace"])
				assert.Equal(t, "3", args["Replicas"])
				assert.Equal(t, "8C16G", args["NodeSpec"])
				assert.Equal(t, "true", args["SetResource"])
				assert.Equal(t, strconv.FormatUint(tioperator.ComputeRequestCPU(7000, 0), 10)+"m", args["CPU"])
				assert.Equal(t, "7000m", args["LimitCPU"])
				assert.Equal(t, "14336", args["Memory"])
				assert.Equal(t, strconv.FormatUint(tioperator.ComputeRequestMemory(14336, 0), 10), args["RequestMemory"])
				assert.Equal(t, "v1.0.0", args["NodeVersion"])
			},
		},
		{
			name: "fallback memory when available memory not set",
			dmCluster: &entity.DMCluster{
				Id:           "test-dm-cluster",
				SharedPoolId: "test-shared-pool",
			},
			nodeQuota: &apiv1.NodeQuota{
				NodeCount: 3,
				NodeSpec: &apiv1.NodeSpec{
					Cpu:      8,
					MemoryGi: 16,
				},
			},
			nodeVersionMap: map[string]string{
				"8C16G": "v1.0.0",
			},
			validate: func(t *testing.T, args map[string]string) {
				t.Helper()
				assert.Equal(t, "test-dm-cluster", args["ClusterId"])
				assert.Equal(t, "test-shared-pool", args["SharedPoolId"])
				assert.Equal(t, "dmtest-shared-pool", args["Namespace"])
				assert.Equal(t, "3", args["Replicas"])
				assert.Equal(t, "8C16G", args["NodeSpec"])
				assert.Equal(t, "true", args["SetResource"])
				// AvailableMemoryMi unset: (MemoryGi - 1) * 1024 = 15360
				assert.Equal(t, "15360", args["Memory"])
				// AvailableCpu unset: (Cpu - 1) * 1000 = 7000
				assert.Equal(t, strconv.FormatUint(tioperator.ComputeRequestCPU(7000, 0), 10)+"m", args["CPU"])
				assert.Equal(t, "7000m", args["LimitCPU"])
				assert.Equal(t, strconv.FormatUint(tioperator.ComputeRequestMemory(15360, 0), 10), args["RequestMemory"])
				assert.Equal(t, "v1.0.0", args["NodeVersion"])
			},
		},
		{
			name: "node version not found",
			dmCluster: &entity.DMCluster{
				Id:           "test-dm-cluster",
				SharedPoolId: "test-shared-pool",
			},
			nodeQuota: &apiv1.NodeQuota{
				NodeCount: 3,
				NodeSpec: &apiv1.NodeSpec{
					Cpu:               16,
					MemoryGi:          32,
					AvailableCpu:      15,
					AvailableMemoryMi: 30720,
				},
			},
			nodeVersionMap: map[string]string{
				"8C16G": "v1.0.0",
			},
			expectedError: "node version not found for node spec",
		},
		{
			name: "empty node version",
			dmCluster: &entity.DMCluster{
				Id:           "test-dm-cluster",
				SharedPoolId: "test-shared-pool",
			},
			nodeQuota: &apiv1.NodeQuota{
				NodeCount: 3,
				NodeSpec: &apiv1.NodeSpec{
					Cpu:               8,
					MemoryGi:          16,
					AvailableCpu:      7,
					AvailableMemoryMi: 14336,
				},
			},
			nodeVersionMap: map[string]string{
				"8C16G": "",
			},
			expectedError: "get node version is empty",
		},
		{
			name: "node quota without node spec",
			dmCluster: &entity.DMCluster{
				Id:           "test-dm-cluster",
				SharedPoolId: "test-shared-pool",
			},
			nodeQuota: &apiv1.NodeQuota{
				NodeCount: 3,
				NodeSpec:  nil,
			},
			nodeVersionMap: map[string]string{
				"0C0G": "v1.0.0",
			},
			expectedError: "node quota or node spec is nil",
		},
	}

	for _, tc := range tests {
		t.Run(tc.name, func(t *testing.T) {
			action := &ApplyDMClusterAction{
				imageConfig: &config.ImageConfig{
					DmImage:         "pingcap/dm:v8.1.0",
					ImagePullPolicy: "IfNotPresent",
				},
			}

			args, err := action.buildDMTemplateArgs(tc.dmCluster, tc.nodeQuota, tc.nodeVersionMap, false)

			if tc.expectedError != "" {
				require.Error(t, err)
				assert.Contains(t, err.Error(), tc.expectedError)
			} else {
				require.NoError(t, err)
				tc.validate(t, args)
			}
		})
	}
}

func TestBuildDMTemplateArgsForDeletion(t *testing.T) {
	action := &ApplyDMClusterAction{
		imageConfig: &config.ImageConfig{
			DmImage:         "pingcap/dm:v8.1.0",
			ImagePullPolicy: "IfNotPresent",
		},
	}

	dmCluster := &entity.DMCluster{
		Id:           "test-dm-cluster",
		SharedPoolId: "test-shared-pool",
	}

	// For deletion, nodeQuota can be nil or incomplete
	nodeQuota := &apiv1.NodeQuota{
		NodeCount: 3,
		NodeSpec: &apiv1.NodeSpec{
			Cpu:      8,
			MemoryGi: 16,
			// AvailableMemoryMi is not set - this should not cause error during deletion
		},
	}

	nodeVersionMap := map[string]string{}

	// Should not error even with incomplete data when needDelete=true
	args, err := action.buildDMTemplateArgs(dmCluster, nodeQuota, nodeVersionMap, true)
	require.NoError(t, err)
	assert.Equal(t, "test-dm-cluster", args["ClusterId"])
	assert.Equal(t, "test-shared-pool", args["SharedPoolId"])
	assert.Equal(t, "3", args["Replicas"])
	assert.Equal(t, "8C16G", args["NodeSpec"])

	// SetResource should not be set for deletion
	_, hasSetResource := args["SetResource"]
	assert.False(t, hasSetResource, "SetResource should not be set for deletion")

	// Image should still be set
	assert.Equal(t, "pingcap/dm:v8.1.0", args["Image"])
	assert.Equal(t, "IfNotPresent", args["ImagePullPolicy"])

	// NodeVersion should be set to dummy value
	assert.Equal(t, "v1.0.0", args["NodeVersion"])
}

func TestRenderDMMasterConfigMap(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_master_configmap", dmMasterTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: ConfigMap")
	assert.Contains(t, yamlStr, "name: dbcluster-dm-master")
	assert.Contains(t, yamlStr, "namespace: dmtest-shared-pool")
	assert.Contains(t, yamlStr, "openapi = true")
}

func TestRenderDMMasterService(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_master_svc", dmMasterTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: Service")
	assert.Contains(t, yamlStr, "name: dbcluster-dm-master")
	assert.Contains(t, yamlStr, "8261") // DM Master API port
}

func TestRenderDMMasterPeerService(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_master_peer_svc", dmMasterTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: Service")
	assert.Contains(t, yamlStr, "clusterIP: None")
	assert.Contains(t, yamlStr, "8291") // DM Master peer port
}

func TestRenderDMMasterStatefulSet(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_master_statefulset", dmMasterTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: StatefulSet")
	assert.Contains(t, yamlStr, "name: dbcluster-dm-master")
	assert.Contains(t, yamlStr, "replicas: 3")
	assert.Contains(t, yamlStr, "volumeClaimTemplates")
	assert.Contains(t, yamlStr, dmMasterTestTemplArgs["Image"])
}

func TestRenderDMWorkerConfigMap(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_worker_configmap", dmWorkerTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: ConfigMap")
	assert.Contains(t, yamlStr, "name: dbcluster-dm-worker")
	assert.Contains(t, yamlStr, "namespace: dmtest-shared-pool")
}

func TestRenderDMWorkerService(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_worker_svc", dmWorkerTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: Service")
	assert.Contains(t, yamlStr, "name: dbcluster-dm-worker")
	assert.Contains(t, yamlStr, "8262") // DM Worker port
}

func TestRenderDMWorkerPeerService(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_worker_peer_svc", dmWorkerTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: Service")
	assert.Contains(t, yamlStr, "clusterIP: None")
}

func TestRenderDMWorkerStatefulSet(t *testing.T) {
	tl := initDMTemplate(t)

	yamlStr, err := tl.Render("dm_worker_statefulset", dmWorkerTestTemplArgs)
	require.NoError(t, err)
	require.NotEmpty(t, yamlStr)

	// Verify that the rendered YAML contains expected fields
	assert.Contains(t, yamlStr, "kind: StatefulSet")
	assert.Contains(t, yamlStr, "name: dbcluster-dm-worker")
	assert.Contains(t, yamlStr, "replicas: 2")
	assert.Contains(t, yamlStr, dmWorkerTestTemplArgs["Image"])
	assert.Contains(t, yamlStr, "emptyDir") // DM Worker uses emptyDir
}

func TestApplyDMClusterAction_Invoke_MissingClusterId(t *testing.T) {
	action := &ApplyDMClusterAction{}

	args := actions.ActionArgs{}

	err := action.Invoke(args)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "dm cluster id is required")
}

func TestApplyDMClusterAction_Invoke_LoadFailed(t *testing.T) {
	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", false).Return(nil, assert.AnError)

	action := &ApplyDMClusterAction{
		dmClusterRepo: dmClusterRepo,
	}

	args := actions.ActionArgs{
		workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
	}

	err := action.Invoke(args)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed to load dm cluster")
}

func TestApplyDMClusterAction_Invoke_UnmarshalFailed(t *testing.T) {
	dmCluster := &entity.DMCluster{
		Id:            "test-dm-cluster",
		SharedPoolId:  "test-shared-pool",
		DataplaneName: "test-dataplane",
	}

	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", false).Return(dmCluster, nil)

	action := &ApplyDMClusterAction{
		dmClusterRepo: dmClusterRepo,
	}

	args := actions.ActionArgs{
		workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
		workflow.DMClusterKey:   actions.ActionArgString("invalid-json"),
	}

	err := action.Invoke(args)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed to unmarshal dmCluster")
}

func TestApplyDMClusterAction_Invoke_OperatorV2DeleteSkipsNodeVersionLookup(t *testing.T) {
	dmCluster := &entity.DMCluster{
		Id:            "test-dm-cluster",
		SharedPoolId:  "test-shared-pool",
		DataplaneName: "test-dataplane",
	}

	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", true).Return(dmCluster, nil)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, v2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).Build()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "test-dataplane").Return(remoteClient, nil)

	dataplaneClient := dataplanemgrmocks.NewClient(t)

	action := &ApplyDMClusterAction{
		dmClusterRepo:    dmClusterRepo,
		dataplaneManager: dataplaneManager,
		dataplaneClient:  dataplaneClient,
		imageConfig:      &config.ImageConfig{DmImage: "pingcap/dm:v8.1.0"},
	}

	args := actions.ActionArgs{
		workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
		workflow.DMClusterKey:   actions.ActionArgString(`{"masterNodeQuota":null,"workerNodeQuota":null}`),
		workflow.DMDeleteKey:    actions.ActionArgString(workflow.DMDeleteValue),
	}

	err := action.Invoke(args)
	require.NoError(t, err)
	dataplaneClient.AssertNotCalled(t, "ListNodePools", mock.Anything, mock.Anything)
}

func TestApplyDMClusterAction_Invoke_OperatorV2DeleteCleansCerts(t *testing.T) {
	dmCluster := &entity.DMCluster{
		Id:            "test-dm-cluster",
		SharedPoolId:  "test-shared-pool",
		DataplaneName: "test-dataplane",
	}
	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())

	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", true).Return(dmCluster, nil)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, v2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))

	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	require.NoError(t, err)
	workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	require.NoError(t, err)

	objects := []client.Object{
		&v2.DMGroup{ObjectMeta: metav1.ObjectMeta{Name: masterGroupName, Namespace: namespace}},
		&v2.DMWorkerGroup{ObjectMeta: metav1.ObjectMeta{Name: workerGroupName, Namespace: namespace}},
	}
	certs, err := cert.DMClusterCertsV2(dmCluster.GetId(), namespace)
	require.NoError(t, err)
	for _, certCR := range certs {
		objects = append(objects, certCR, &corev1.Secret{
			ObjectMeta: metav1.ObjectMeta{Name: certCR.Name, Namespace: namespace},
		})
	}
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "test-dataplane").Return(remoteClient, nil)

	action := &ApplyDMClusterAction{
		dmClusterRepo:    dmClusterRepo,
		dataplaneManager: dataplaneManager,
		imageConfig:      &config.ImageConfig{DmImage: "pingcap/dm:v8.1.0"},
	}

	args := actions.ActionArgs{
		workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
		workflow.DMClusterKey:   actions.ActionArgString(`{"masterNodeQuota":null,"workerNodeQuota":null}`),
		workflow.DMDeleteKey:    actions.ActionArgString(workflow.DMDeleteValue),
	}

	err = action.Invoke(args)
	require.NoError(t, err)

	certs, err = cert.DMClusterCertsV2(dmCluster.GetId(), namespace)
	require.NoError(t, err)
	for _, certCR := range certs {
		getErr := remoteClient.Get(context.Background(), types.NamespacedName{Name: certCR.Name, Namespace: namespace}, &certmanagerv1.Certificate{})
		assert.True(t, apierrors.IsNotFound(getErr))

		secretErr := remoteClient.Get(context.Background(), types.NamespacedName{Name: certCR.Name, Namespace: namespace}, &corev1.Secret{})
		assert.True(t, apierrors.IsNotFound(secretErr))
	}
}

func TestApplyDMClusterAction_Invoke_OperatorV2ResolvesNodeVersions(t *testing.T) {
	dmCluster := &entity.DMCluster{
		Id:            "test-dm-cluster",
		SharedPoolId:  "test-shared-pool",
		DataplaneName: "test-dataplane",
	}
	inputCluster := &entity.DMCluster{
		MasterNodeQuota: &entity.NodeQuota{
			NodeCount: 1,
			NodeSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableCpu: 6, AvailableMemoryMi: 12288},
		},
		WorkerNodeQuota: &entity.NodeQuota{
			NodeCount: 1,
			NodeSpec:  &entity.NodeSpec{Cpu: 4, MemoryGi: 8, AvailableCpu: 3, AvailableMemoryMi: 6144},
		},
	}
	inputClusterStr, err := protojson.Marshal(inputCluster)
	require.NoError(t, err)

	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", false).Return(dmCluster, nil)

	sharedPool := &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id: dmCluster.GetSharedPoolId(),
		},
		TenantId: "123",
	}
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, dmCluster.GetSharedPoolId(), false).Return(sharedPool, nil)

	clusterConfigClient := clusterconfigmocks.NewClient(t)
	clusterConfigClient.On("GetDMComponentVersionAndConfig", mock.Anything, sharedPool).Return("v8.5.0", &entity.ComponentConfiguration{
		ImageRepo: "docker.io/pingcap/data-migration",
		ImageTag:  "dm-hotfix-001",
	}, nil)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, v2.AddToScheme(scheme))
	require.NoError(t, certmanagerv1.AddToScheme(scheme))
	namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
	cluster := &v2.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: namespace}}
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster).Build()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "test-dataplane").Return(remoteClient, nil)

	dataplaneClient := dataplanemgrmocks.NewClient(t)
	dataplaneClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "test-dataplane"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 8, MemoryGi: 16, K8SVersion: "1.31"},
				{CpuCore: 4, MemoryGi: 8, K8SVersion: "1.30"},
			},
		}, nil)

	action := &ApplyDMClusterAction{
		sharedPoolRepo:      sharedPoolRepo,
		dmClusterRepo:       dmClusterRepo,
		clusterConfigClient: clusterConfigClient,
		dataplaneManager:    dataplaneManager,
		dataplaneClient:     dataplaneClient,
		imageConfig:         &config.ImageConfig{Registry: "registry.example.com", RepoPrefix: "tidbcloud"},
	}

	args := actions.ActionArgs{
		workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
		workflow.DMClusterKey:   actions.ActionArgString(string(inputClusterStr)),
	}

	err = action.Invoke(args)
	require.NoError(t, err)

	master := &v2.DMGroup{}
	err = remoteClient.Get(context.Background(), types.NamespacedName{Name: "dbcluster-test-dm-cluster", Namespace: namespace}, master)
	require.NoError(t, err)
	assert.Equal(t, "v8.5.0", master.Spec.Template.Spec.Version)
	require.NotNil(t, master.Spec.Template.Spec.Image)
	assert.Equal(t, "registry.example.com/tidbcloud/data-migration:dm-hotfix-001", *master.Spec.Template.Spec.Image)
	assert.Equal(t, "1.31", master.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector["tidbcloud.com/node-pool-k8s-version"])

	worker := &v2.DMWorkerGroup{}
	err = remoteClient.Get(context.Background(), types.NamespacedName{Name: "dbcluster-test-dm-cluster", Namespace: namespace}, worker)
	require.NoError(t, err)
	assert.Equal(t, "v8.5.0", worker.Spec.Template.Spec.Version)
	require.NotNil(t, worker.Spec.Template.Spec.Image)
	assert.Equal(t, "registry.example.com/tidbcloud/data-migration:dm-hotfix-001", *worker.Spec.Template.Spec.Image)
	assert.Equal(t, "1.30", worker.Spec.Template.Spec.Overlay.Pod.Spec.NodeSelector["tidbcloud.com/node-pool-k8s-version"])

	masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	require.NoError(t, err)
	workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
	require.NoError(t, err)
	masterCert := &certmanagerv1.Certificate{}
	err = remoteClient.Get(context.Background(), types.NamespacedName{
		Name:      cert.DMComponentCertSecretName(masterGroupName, entity.DMMasterComponent),
		Namespace: namespace,
	}, masterCert)
	require.NoError(t, err)
	workerCert := &certmanagerv1.Certificate{}
	err = remoteClient.Get(context.Background(), types.NamespacedName{
		Name:      cert.DMComponentCertSecretName(workerGroupName, entity.DMWorkerComponent),
		Namespace: namespace,
	}, workerCert)
	require.NoError(t, err)
}

func TestApplyDMClusterAction_Invoke_OperatorV2UpdateImageFlag(t *testing.T) {
	tests := []struct {
		name             string
		args             actions.ActionArgs
		wantVersion      string
		wantImage        string
		wantConfigLookup bool
	}{
		{
			name:        "default preserves existing image and version",
			args:        actions.ActionArgs{},
			wantVersion: "v8.4.0",
			wantImage:   "repo/dm:v8.4.0",
		},
		{
			name: "explicit config-only preserves existing image and version",
			args: actions.ActionArgs{
				workflow.DMUpdateImageKey: actions.ActionArgBool(false),
			},
			wantVersion: "v8.4.0",
			wantImage:   "repo/dm:v8.4.0",
		},
		{
			name: "explicit update image applies resolved image and version",
			args: actions.ActionArgs{
				workflow.DMUpdateImageKey: actions.ActionArgBool(true),
			},
			wantVersion:      "v8.5.0",
			wantImage:        "registry.example.com/tidbcloud/data-migration:dm-hotfix-001",
			wantConfigLookup: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			dmCluster := &entity.DMCluster{
				Id:            "test-dm-cluster",
				SharedPoolId:  "test-shared-pool",
				DataplaneName: "test-dataplane",
			}
			inputCluster := &entity.DMCluster{
				MasterNodeQuota: &entity.NodeQuota{
					NodeCount: 1,
					NodeSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16, AvailableCpu: 6, AvailableMemoryMi: 12288},
				},
				WorkerNodeQuota: &entity.NodeQuota{
					NodeCount: 1,
					NodeSpec:  &entity.NodeSpec{Cpu: 4, MemoryGi: 8, AvailableCpu: 3, AvailableMemoryMi: 6144},
				},
			}
			inputClusterStr, err := protojson.Marshal(inputCluster)
			require.NoError(t, err)

			dmClusterRepo := repomnocks.NewDMClusterRepo(t)
			dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", false).Return(dmCluster, nil)

			sharedPool := &entity.SharedPool{
				SharedPool: apiv1.SharedPool{
					Id: dmCluster.GetSharedPoolId(),
				},
				TenantId: "123",
			}
			sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)

			clusterConfigClient := clusterconfigmocks.NewClient(t)
			if tt.wantConfigLookup {
				sharedPoolRepo.On("Load", mock.Anything, dmCluster.GetSharedPoolId(), false).Return(sharedPool, nil)
				clusterConfigClient.On("GetDMComponentVersionAndConfig", mock.Anything, sharedPool).Return("v8.5.0", &entity.ComponentConfiguration{
					ImageRepo: "docker.io/pingcap/data-migration",
					ImageTag:  "dm-hotfix-001",
				}, nil)
			}

			scheme := runtime.NewScheme()
			require.NoError(t, corev1.AddToScheme(scheme))
			require.NoError(t, v2.AddToScheme(scheme))
			require.NoError(t, certmanagerv1.AddToScheme(scheme))
			namespace := entity.GetDMV2Namespace(dmCluster.GetSharedPoolId())
			cluster := &v2.Cluster{ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: namespace}}
			masterGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
			require.NoError(t, err)
			workerGroupName, err := entity.DMGroupNameV2(dmCluster.GetId())
			require.NoError(t, err)
			oldImage := "repo/dm:v8.4.0"
			master := &v2.DMGroup{
				ObjectMeta: metav1.ObjectMeta{Name: masterGroupName, Namespace: namespace},
				Spec: v2.DMGroupSpec{
					Template: v2.DMTemplate{
						Spec: v2.DMTemplateSpec{
							Version: "v8.4.0",
							Image:   &oldImage,
						},
					},
				},
			}
			worker := &v2.DMWorkerGroup{
				ObjectMeta: metav1.ObjectMeta{Name: workerGroupName, Namespace: namespace},
				Spec: v2.DMWorkerGroupSpec{
					Template: v2.DMWorkerTemplate{
						Spec: v2.DMWorkerTemplateSpec{
							Version: "v8.4.0",
							Image:   &oldImage,
						},
					},
				},
			}
			remoteClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(cluster, master, worker).Build()

			dataplaneManager := dataplanemocks.NewManager(t)
			dataplaneManager.On("GetClient", "test-dataplane").Return(remoteClient, nil)

			dataplaneClient := dataplanemgrmocks.NewClient(t)
			dataplaneClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "test-dataplane"}).
				Return(&dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{CpuCore: 8, MemoryGi: 16, K8SVersion: "1.31"},
						{CpuCore: 4, MemoryGi: 8, K8SVersion: "1.30"},
					},
				}, nil)

			action := &ApplyDMClusterAction{
				sharedPoolRepo:      sharedPoolRepo,
				dmClusterRepo:       dmClusterRepo,
				clusterConfigClient: clusterConfigClient,
				dataplaneManager:    dataplaneManager,
				dataplaneClient:     dataplaneClient,
				imageConfig:         &config.ImageConfig{Registry: "registry.example.com", RepoPrefix: "tidbcloud"},
			}

			args := actions.ActionArgs{
				workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
				workflow.DMClusterKey:   actions.ActionArgString(string(inputClusterStr)),
			}
			for k, v := range tt.args {
				args[k] = v
			}

			err = action.Invoke(args)
			require.NoError(t, err)

			gotMaster := &v2.DMGroup{}
			err = remoteClient.Get(context.Background(), types.NamespacedName{Name: masterGroupName, Namespace: namespace}, gotMaster)
			require.NoError(t, err)
			assert.Equal(t, tt.wantVersion, gotMaster.Spec.Template.Spec.Version)
			require.NotNil(t, gotMaster.Spec.Template.Spec.Image)
			assert.Equal(t, tt.wantImage, *gotMaster.Spec.Template.Spec.Image)

			gotWorker := &v2.DMWorkerGroup{}
			err = remoteClient.Get(context.Background(), types.NamespacedName{Name: workerGroupName, Namespace: namespace}, gotWorker)
			require.NoError(t, err)
			assert.Equal(t, tt.wantVersion, gotWorker.Spec.Template.Spec.Version)
			require.NotNil(t, gotWorker.Spec.Template.Spec.Image)
			assert.Equal(t, tt.wantImage, *gotWorker.Spec.Template.Spec.Image)
			if !tt.wantConfigLookup {
				sharedPoolRepo.AssertNotCalled(t, "Load", mock.Anything, mock.Anything, mock.Anything)
				clusterConfigClient.AssertNotCalled(t, "GetDMComponentVersionAndConfig", mock.Anything, mock.Anything)
			}
		})
	}
}

func TestApplyDMClusterAction_ResolveDMComponentImageFallback(t *testing.T) {
	dmCluster := &entity.DMCluster{
		Id:           "test-dm-cluster",
		SharedPoolId: "test-shared-pool",
	}
	sharedPool := &entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:      dmCluster.GetSharedPoolId(),
			Version: "v8.4.0",
		},
	}

	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	sharedPoolRepo.On("Load", mock.Anything, dmCluster.GetSharedPoolId(), false).Return(sharedPool, nil)

	resolveErr := errors.New("component image not found for data-migration")
	clusterConfigClient := clusterconfigmocks.NewClient(t)
	clusterConfigClient.On("GetDMComponentVersionAndConfig", mock.Anything, sharedPool).
		Return("v8.5.0", &entity.ComponentConfiguration{}, resolveErr)

	action := &ApplyDMClusterAction{
		sharedPoolRepo:      sharedPoolRepo,
		clusterConfigClient: clusterConfigClient,
		imageConfig:         &config.ImageConfig{DmImage: "pingcap/dm:v8.1.0"},
	}

	version, image, err := action.resolveDMComponentImage(context.Background(), dmCluster)
	require.NoError(t, err)
	assert.Equal(t, "v8.5.0", version)
	assert.Equal(t, "pingcap/dm:v8.1.0", image)
}

func TestApplyDMClusterAction_Invoke_OperatorV2FailsWhenNodeVersionMissing(t *testing.T) {
	dmCluster := &entity.DMCluster{
		Id:            "test-dm-cluster",
		SharedPoolId:  "test-shared-pool",
		DataplaneName: "test-dataplane",
	}
	inputCluster := &entity.DMCluster{
		MasterNodeQuota: &entity.NodeQuota{
			NodeCount: 1,
			NodeSpec:  &entity.NodeSpec{Cpu: 8, MemoryGi: 16},
		},
	}
	inputClusterStr, err := protojson.Marshal(inputCluster)
	require.NoError(t, err)

	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	dmClusterRepo.On("Load", mock.Anything, "test-dm-cluster", false).Return(dmCluster, nil)

	scheme := runtime.NewScheme()
	require.NoError(t, corev1.AddToScheme(scheme))
	require.NoError(t, v2.AddToScheme(scheme))
	remoteClient := fake.NewClientBuilder().WithScheme(scheme).Build()

	dataplaneManager := dataplanemocks.NewManager(t)
	dataplaneManager.On("GetClient", "test-dataplane").Return(remoteClient, nil)

	dataplaneClient := dataplanemgrmocks.NewClient(t)
	dataplaneClient.On("ListNodePools", mock.Anything, &dplanemgrv1.ListNodePoolsRequest{DataPlaneId: "test-dataplane"}).
		Return(&dplanemgrv1.ListNodePoolsResponse{
			NodePools: []*dplanemgrv1.NodePool{
				{CpuCore: 4, MemoryGi: 8, K8SVersion: "1.30"},
			},
		}, nil)

	action := &ApplyDMClusterAction{
		dmClusterRepo:    dmClusterRepo,
		dataplaneManager: dataplaneManager,
		dataplaneClient:  dataplaneClient,
		imageConfig:      &config.ImageConfig{DmImage: "pingcap/dm:v8.1.0"},
	}

	args := actions.ActionArgs{
		workflow.DMClusterIdKey: actions.ActionArgString("test-dm-cluster"),
		workflow.DMClusterKey:   actions.ActionArgString(string(inputClusterStr)),
	}

	err = action.Invoke(args)
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed to resolve node versions for dm cluster v2")
	assert.Contains(t, err.Error(), "node version not found for node spec: 8C16G")
}

func TestNewApplyDMClusterAction(t *testing.T) {
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dmClusterRepo := repomnocks.NewDMClusterRepo(t)
	imageConfig := &config.ImageConfig{
		DmImage: "pingcap/dm:v8.1.0",
	}

	action := NewApplyDMClusterAction(
		sharedPoolRepo,
		dmClusterRepo,
		nil, // dataplaneManager
		nil, // dataplaneClient
		imageConfig,
		nil, // clusterConfigClient
	)

	require.NotNil(t, action)
	assert.Equal(t, ApplyDMClusterActionName, action.Name())
}

func initDMTemplate(t *testing.T) *render.TemplateLibrary {
	t.Helper()
	tplPath := "../../../deployments/templates"
	templateLib := render.NewTemplateLibrary()

	dmTemplates := []string{
		"dm_master_configmap",
		"dm_master_svc",
		"dm_master_peer_svc",
		"dm_master_statefulset",
		"dm_worker_configmap",
		"dm_worker_svc",
		"dm_worker_peer_svc",
		"dm_worker_statefulset",
	}

	for _, tpl := range dmTemplates {
		data, err := os.ReadFile(filepath.Join(tplPath, tpl+".templ"))
		require.NoError(t, err, "Failed to read template file: %s", tpl)
		templateLib.SafeNewTemplate(tpl, string(data))
	}
	return templateLib
}

// buildDMTemplateArgs keeps the historical template rendering tests isolated
// from the production Operator v2 apply path.
func (a *ApplyDMClusterAction) buildDMTemplateArgs(
	dmCluster *entity.DMCluster,
	nodeQuota *entity.NodeQuota,
	nodeVersionMap map[string]string,
	needDelete bool,
) (map[string]string, error) {
	if nodeQuota == nil || nodeQuota.GetNodeSpec() == nil {
		return nil, fmt.Errorf("node quota or node spec is nil")
	}
	templArgs := map[string]string{
		"ClusterId":    dmCluster.GetId(),
		"SharedPoolId": dmCluster.GetSharedPoolId(),
		"Namespace":    "dm" + dmCluster.GetSharedPoolId(),
		"Replicas":     strconv.FormatUint(uint64(nodeQuota.GetNodeCount()), 10),
		"NodeSpec":     entity.GetComponentNodeSpec(nodeQuota.GetNodeSpec()),
	}
	if !needDelete {
		ns := nodeQuota.GetNodeSpec()
		availMem := uint64(ns.GetAvailableMemoryMi())
		if availMem == 0 {
			availMem = uint64(ns.GetMemoryGi()-1) * 1024
		}
		limitMem := tioperator.EffectiveMaxMemoryMi(ns)
		if limitMem == 0 {
			limitMem = availMem
		}
		templArgs["RequestMemory"] = strconv.FormatUint(tioperator.ComputeRequestMemory(availMem, 0), 10)
		templArgs["Memory"] = strconv.FormatUint(limitMem, 10)
		templArgs["CPU"] = strconv.FormatUint(tioperator.ComputeRequestCPU(tioperator.EffectiveAvailableCPUMilli(ns), 0), 10) + "m"
		templArgs["LimitCPU"] = strconv.FormatUint(tioperator.EffectiveMaxCPUMilli(ns), 10) + "m"
		templArgs["SetResource"] = "true"
		nodeSpec := entity.GetComponentNodeSpec(nodeQuota.GetNodeSpec())
		nodeVersion, ok := nodeVersionMap[nodeSpec]
		if !ok {
			return nil, fmt.Errorf("node version not found for node spec: %s", nodeSpec)
		}
		if nodeVersion == "" {
			return nil, fmt.Errorf("get node version is empty")
		}
		templArgs["NodeVersion"] = nodeVersion
	} else {
		templArgs["NodeVersion"] = "v1.0.0"
	}
	dmImage := ""
	if a.imageConfig != nil {
		dmImage = a.imageConfig.DmImage
	}
	templArgs["Image"] = dmImage
	if a.imageConfig == nil || a.imageConfig.ImagePullPolicy == "" {
		templArgs["ImagePullPolicy"] = "IfNotPresent"
	} else {
		templArgs["ImagePullPolicy"] = a.imageConfig.ImagePullPolicy
	}
	return templArgs, nil
}