package dataplane

import (
	"context"
	"errors"
	"fmt"
	"net"
	"strings"
	"testing"

	"buf.build/go/protovalidate"
	ackv1alpha1 "github.com/crossplane-contrib/provider-upjet-alibabacloud/apis/ack/v1alpha1"
	protovalidate_middleware "github.com/grpc-ecosystem/go-grpc-middleware/v2/interceptors/protovalidate"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	ec2v1beta1 "github.com/upbound/provider-aws/apis/ec2/v1beta1"
	eksv1beta1 "github.com/upbound/provider-aws/apis/eks/v1beta1"
	"go.uber.org/mock/gomock"
	"google.golang.org/grpc"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/credentials/insecure"
	"google.golang.org/grpc/status"
	"google.golang.org/grpc/test/bufconn"
	apierrors "k8s.io/apimachinery/pkg/api/errors"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/runtime"
	k8sscheme "k8s.io/client-go/kubernetes/scheme"
	"k8s.io/utils/pointer"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"

	dmpbv1 "github.com/tidbcloud/cluster-service-ng/api/tidbcloud/dataplane-manager/v1"
	"github.com/tidbcloud/infra-provider/regional-server/api/v1alpha1"
	bizerr "github.com/tidbcloud/infra-provider/regional-server/pkg/common/errors"
	rsconfig "github.com/tidbcloud/infra-provider/regional-server/pkg/config"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/consts"
	"github.com/tidbcloud/infra-provider/regional-server/pkg/repository"
	projectscheme "github.com/tidbcloud/infra-provider/regional-server/pkg/scheme"
	dataplane_mock "github.com/tidbcloud/infra-provider/regional-server/pkg/service/dataplane/mock"
	"github.com/tidbcloud/infra-provider/regional-server/test/mockrepository"
)

func init() {
	_ = v1alpha1.AddToScheme(k8sscheme.Scheme)
	_ = v1alpha1.AddToScheme(projectscheme.Scheme)
}

func newTestDataPlaneGRPCServer(t *testing.T) *grpc.Server {
	validator, err := protovalidate.New()
	require.NoError(t, err)

	opts := []grpc.ServerOption{
		grpc.ChainUnaryInterceptor(
			protovalidate_middleware.UnaryServerInterceptor(validator),
		),
	}

	srv := grpc.NewServer(opts...)
	svc := &DataPlaneService{
		config: rsconfig.RootConfig{
			Namespace: "default",
			ControllerConfig: rsconfig.ControllerConfig{
				RegionConfig: rsconfig.RegionConfig{
					Name: "us-west-2",
				},
			},
		},
	}
	dmpbv1.RegisterDataPlaneServiceServer(srv, svc)
	return srv
}

func newTestDataPlaneGRPCClient(t *testing.T, srv *grpc.Server) dmpbv1.DataPlaneServiceClient {
	lis := bufconn.Listen(1024 * 1024)
	go func() { _ = srv.Serve(lis) }()

	dialer := grpc.WithContextDialer(func(ctx context.Context, s string) (net.Conn, error) {
		return lis.Dial()
	})
	conn, err := grpc.DialContext(context.Background(), "bufnet", grpc.WithTransportCredentials(insecure.NewCredentials()), dialer)
	require.NoError(t, err)

	return dmpbv1.NewDataPlaneServiceClient(conn)
}

func TestDataPlaneService_Protovalidate_InvalidRequests(t *testing.T) {
	srv := newTestDataPlaneGRPCServer(t)
	client := newTestDataPlaneGRPCClient(t, srv)

	t.Run("empty dataplane_id in CreateDataPlaneRequest", func(t *testing.T) {
		req := &dmpbv1.CreateDataPlaneRequest{
			DataPlaneId: "",
			DataPlane:   &dmpbv1.DataPlane{},
		}
		resp, err := client.CreateDataPlane(context.Background(), req)
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("missing DataPlane in CreateDataPlaneRequest", func(t *testing.T) {
		req := &dmpbv1.CreateDataPlaneRequest{
			DataPlaneId: "dp-123",
			DataPlane:   nil,
		}
		resp, err := client.CreateDataPlane(context.Background(), req)
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty dataplane_id in GetDataPlaneRequest", func(t *testing.T) {
		req := &dmpbv1.GetDataPlaneRequest{
			DataPlaneId: "",
		}
		resp, err := client.GetDataPlane(context.Background(), req)
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty dataplane_id in UpgradeDataPlaneRequest", func(t *testing.T) {
		req := &dmpbv1.UpgradeDataPlaneRequest{
			DataPlaneId: "",
			K8SVersion:  "v1.28",
		}
		resp, err := client.UpgradeDataPlane(context.Background(), req)
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty k8s_version in UpgradeDataPlaneRequest", func(t *testing.T) {
		req := &dmpbv1.UpgradeDataPlaneRequest{
			DataPlaneId: "dp-123",
			K8SVersion:  "",
		}
		resp, err := client.UpgradeDataPlane(context.Background(), req)
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "k8s_version")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})

	t.Run("empty dataplane_id in DeleteDataPlaneRequest", func(t *testing.T) {
		req := &dmpbv1.DeleteDataPlaneRequest{
			DataPlaneId: "",
		}
		resp, err := client.DeleteDataPlane(context.Background(), req)
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "data_plane_id")
		assert.Equal(t, codes.InvalidArgument, status.Code(err))
	})
}

// TODO: add back after adopting existing nodepools feature is done
// func TestDataPlaneService_ListNodePools(t *testing.T) {
// 	scheme := runtime.NewScheme()
// 	_ = v1alpha1.AddToScheme(scheme)
//
// 	mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
// 		&v1alpha1.DataPlane{
// 			ObjectMeta: metav1.ObjectMeta{
// 				Name:      "dp-id",
// 				Namespace: "ns",
// 			},
// 			Status: v1alpha1.DataPlaneStatus{
// 				Phase: v1alpha1.DataPlanePhaseActive,
// 				AppBox: &v1alpha1.Reference{
// 					Name:      "appbox-1",
// 					Namespace: "ns",
// 				},
// 			},
// 		},
// 		&v1alpha1.NodePool{
// 			ObjectMeta: metav1.ObjectMeta{
// 				Name:      "np-1",
// 				Namespace: "ns",
// 				Labels: map[string]string{
// 					consts.DataplaneLabelKey: "dp-id",
// 				},
// 			},
// 			Spec: v1alpha1.NodePoolSpec{
// 				InstanceType:      "c8g.xlarge",
// 				KubernetesVersion: "1.29.2",
// 				CpuCore:           4,
// 				MemoryGi:          8,
// 			},
// 		},
// 		&v1alpha1.NodePool{
// 			ObjectMeta: metav1.ObjectMeta{
// 				Name:      "np-2",
// 				Namespace: "ns",
// 				Labels: map[string]string{
// 					consts.DataplaneLabelKey: "dp-other",
// 				},
// 			},
// 			Spec: v1alpha1.NodePoolSpec{
// 				InstanceType:      "c8g.2xlarge",
// 				KubernetesVersion: "1.29.2",
// 			},
// 		},
// 	).Build()
//
// 	dps := NewDataPlaneService(
// 		mockClient,
// 		repository.NewDataPlaneReaderWriter(mockClient),
// 		nil, nil, nil,
// 		nil,
// 		rsconfig.RootConfig{
// 			Namespace:          "ns",
// 			ResourceNamePrefix: "tidbcloud",
// 			ControllerConfig: rsconfig.ControllerConfig{
// 				RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
// 			},
// 		},
// 		rsconfig.TemplatesConfig{},
// 	)
//
// 	resp, err := dps.ListNodePools(context.Background(), &dmpbv1.ListNodePoolsRequest{
// 		DataPlaneId: "dp-id",
// 	})
//
// 	assert.NoError(t, err)
// 	assert.Len(t, resp.NodePools, 1)
// 	assert.Equal(t, "c8g.xlarge", resp.NodePools[0].InstanceType)
// }

func TestDataPlaneService_ListNodePools_ACKUsesNodePoolCRs(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1.AddToScheme(scheme))

	mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
		&v1alpha1.DataPlane{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "dp-1",
				Namespace: "ns",
			},
			Status: v1alpha1.DataPlaneStatus{
				Phase: v1alpha1.DataPlanePhaseActive,
				AppBox: &v1alpha1.Reference{
					Name:      "appbox-1",
					Namespace: "ns",
				},
			},
		},
		&v1alpha1.APPBox{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "appbox-1",
				Namespace: "ns",
			},
			Spec: v1alpha1.APPBoxSpec{
				Provider: consts.ProviderTypeACK,
			},
		},
		&v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "np-1",
				Namespace: "ns",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: v1alpha1.NodePoolSpec{
				InstanceType:      "ecs.g7.large",
				KubernetesVersion: "1.31",
				CpuCore:           2,
				MemoryGi:          8,
			},
			Status: v1alpha1.NodePoolStatus{
				Phase: v1alpha1.NodePoolPhaseActive,
			},
		},
	).Build()

	svc := NewDataPlaneService(
		mockClient,
		repository.NewDataPlaneReaderWriter(mockClient),
		nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
		nil,
		rsconfig.RootConfig{
			Namespace:          "ns",
			ResourceNamePrefix: "tidbcloud",
			ControllerConfig: rsconfig.ControllerConfig{
				RegionConfig: rsconfig.RegionConfig{Name: "ap-southeast-1"},
			},
		},
		rsconfig.TemplatesConfig{},
	)

	resp, err := svc.ListNodePools(context.Background(), &dmpbv1.ListNodePoolsRequest{
		DataPlaneId: "dp-1",
	})
	require.NoError(t, err)
	require.Len(t, resp.NodePools, 1)
	assert.Equal(t, "np-1", resp.NodePools[0].Name)
	assert.Equal(t, dmpbv1.NodePoolStatus_NODEPOOL_ACTIVE, resp.NodePools[0].Status)
}

func TestDataPlaneService_CreateNodePool(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = v1alpha1.AddToScheme(scheme)

	t.Run("create nodepool successfully", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc: &v1alpha1.Reference{
						Name:      "vpc-1",
						Namespace: "ns",
					},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.34",
					},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePool: &dmpbv1.NodePool{
				InstanceType:          "c8g.xlarge",
				K8SVersion:            "1.34",
				CpuCore:               8,
				AvailableCpuMilliCore: 7000,
				MemoryGi:              16,
				AvailableMemoryMi:     14336,
				NodeType:              dmpbv1.NodeType_Dedicated,
				WithLocalDisk:         true,
				LocalDiskSizeGi:       150,
				Labels:                map[string]string{"test": "test"},
				Taints: []*dmpbv1.NodePoolTaint{
					{Key: "test", Value: "test", Effect: "NoSchedule"},
				},
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.Equal(t, buildNodePoolName("tidbcloud", "dp-1", "c8g.xlarge", "1.34", 8, 16), resp.Name)
		assert.Equal(t, "c8g.xlarge", resp.InstanceType)
		assert.Equal(t, "1.34", resp.K8SVersion)

		var npList v1alpha1.NodePoolList
		err = mockClient.List(context.Background(), &npList)
		assert.NoError(t, err)
		assert.Len(t, npList.Items, 1)
		assert.Equal(t, buildNodePoolName("tidbcloud", "dp-1", "c8g.xlarge", "1.34", 8, 16), npList.Items[0].Name)
		assert.Empty(t, npList.Items[0].Spec.ResourceNamePrefix)
		assert.Equal(t, "c8g.xlarge", npList.Items[0].Spec.InstanceType)
		assert.Equal(t, "1.34", npList.Items[0].Spec.KubernetesVersion)
		assert.Equal(t, uint32(7000), npList.Items[0].Spec.AvailableCpuMilliCore)
		assert.Equal(t, uint32(7000), npList.Items[0].Spec.MaxCpuMilli)
		assert.Equal(t, uint32(14336), npList.Items[0].Spec.MaxMemoryMi)
	})

	t.Run("create nodepool with custom resource prefix", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc: &v1alpha1.Reference{
						Name:      "vpc-1",
						Namespace: "ns",
					},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.34",
					},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId:    "dp-1",
			ResourcePrefix: "custom",
			NodePool: &dmpbv1.NodePool{
				InstanceType: "c8g.xlarge",
				K8SVersion:   "1.34",
				CpuCore:      8,
				MemoryGi:     16,
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.Equal(t, buildNodePoolName("custom", "dp-1", "c8g.xlarge", "1.34", 8, 16), resp.Name)

		var npList v1alpha1.NodePoolList
		err = mockClient.List(context.Background(), &npList)
		assert.NoError(t, err)
		assert.Len(t, npList.Items, 1)
		assert.Equal(t, buildNodePoolName("custom", "dp-1", "c8g.xlarge", "1.34", 8, 16), npList.Items[0].Name)
		assert.Equal(t, "custom", npList.Items[0].Spec.ResourceNamePrefix)
	})

	t.Run("create nodepool - missing required fields", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.34",
					},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePool: &dmpbv1.NodePool{
				InstanceType: "",
				K8SVersion:   "",
			},
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "instance_type")
	})

	t.Run("create nodepool - missing k8s version", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc: &v1alpha1.Reference{
						Name:      "vpc-1",
						Namespace: "ns",
					},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.34",
					},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePool: &dmpbv1.NodePool{
				InstanceType: "c8g.xlarge",
				K8SVersion:   "",
				CpuCore:      8,
				MemoryGi:     16,
			},
		})
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, buildNodePoolName("tidbcloud", "dp-1", "c8g.xlarge", "1.34", 8, 16), resp.Name)
		assert.Equal(t, "1.34", resp.K8SVersion)
	})

	// placement_group_enabled is the CSNG-supplied durable record:
	// the DPM service honours the request value and persists it on the
	// NodePool Spec, with a topology sanity check that rejects `true` on
	// multi-AZ or non-EKS dataplanes. This lets CSNG's sibling-inheritance
	// chain pin a legacy false primary onto its fallback instead of being
	// silently overwritten by a topology-derived true.
	t.Run("placement_group_enabled honours false on single-AZ (legacy false primary)", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-zonal", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-zonal", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{Name: "vpc-zonal", Namespace: "ns"},
				Spec:       v1alpha1.VPCSpec{Zones: []string{"us-west-2a"}},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeEKS,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		// CSNG sibling-inheritance sends false on the wire (legacy single-AZ
		// PG-disabled primary). The service must honour the wire value and
		// persist false, so the new NodePool stays on the legacy per-zone
		// path instead of being migrated onto the partition path.
		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-zonal",
			NodePool: &dmpbv1.NodePool{
				InstanceType:          "c8g.xlarge",
				K8SVersion:            "1.34",
				CpuCore:               8,
				MemoryGi:              16,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     14336,
				PlacementGroupEnabled: false,
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.False(t, resp.PlacementGroupEnabled, "wire value false on a single-AZ EKS dataplane must persist as the durable record")

		var npList v1alpha1.NodePoolList
		require.NoError(t, mockClient.List(context.Background(), &npList))
		require.Len(t, npList.Items, 1)
		assert.False(t, npList.Items[0].Spec.PlacementGroupEnabled, "the spec is the durable record and must mirror the wire value")
	})

	t.Run("placement_group_enabled honours true on single-AZ", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-zonal-true", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-zonal-true", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{Name: "vpc-zonal-true", Namespace: "ns"},
				Spec:       v1alpha1.VPCSpec{Zones: []string{"us-west-2a"}},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeEKS,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-zonal-true",
			NodePool: &dmpbv1.NodePool{
				InstanceType:          "c8g.xlarge",
				K8SVersion:            "1.34",
				CpuCore:               8,
				MemoryGi:              16,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     14336,
				PlacementGroupEnabled: true,
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.True(t, resp.PlacementGroupEnabled)

		var npList v1alpha1.NodePoolList
		require.NoError(t, mockClient.List(context.Background(), &npList))
		require.Len(t, npList.Items, 1)
		assert.True(t, npList.Items[0].Spec.PlacementGroupEnabled)
	})

	t.Run("placement_group_enabled true on multi-AZ is rejected (topology sanity check)", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-regional", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-regional", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{Name: "vpc-regional", Namespace: "ns"},
				Spec:       v1alpha1.VPCSpec{Zones: []string{"us-west-2a", "us-west-2b", "us-west-2c"}},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeEKS,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		// Producer (a stale or buggy CSNG) tries to enable PG on a
		// multi-AZ pool. The service must reject the request so a
		// partition PG never spans AZs.
		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-regional",
			NodePool: &dmpbv1.NodePool{
				InstanceType:          "c8g.xlarge",
				K8SVersion:            "1.34",
				CpuCore:               8,
				MemoryGi:              16,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     14336,
				PlacementGroupEnabled: true,
			},
		})
		require.Error(t, err)
		assert.Nil(t, resp)

		var npList v1alpha1.NodePoolList
		require.NoError(t, mockClient.List(context.Background(), &npList))
		assert.Empty(t, npList.Items, "a rejected CreateNodePool must not leave a CR behind")
	})

	t.Run("placement_group_enabled false on multi-AZ is accepted", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-regional-false", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-regional-false", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{Name: "vpc-regional-false", Namespace: "ns"},
				Spec:       v1alpha1.VPCSpec{Zones: []string{"us-west-2a", "us-west-2b"}},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeEKS,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-regional-false",
			NodePool: &dmpbv1.NodePool{
				InstanceType:          "c8g.xlarge",
				K8SVersion:            "1.34",
				CpuCore:               8,
				MemoryGi:              16,
				AvailableCpuMilliCore: 7000,
				AvailableMemoryMi:     14336,
				PlacementGroupEnabled: false,
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.False(t, resp.PlacementGroupEnabled)

		var npList v1alpha1.NodePoolList
		require.NoError(t, mockClient.List(context.Background(), &npList))
		require.Len(t, npList.Items, 1)
		assert.False(t, npList.Items[0].Spec.PlacementGroupEnabled)
	})

	// Unresolved topology must surface as an error so the caller retries.
	// Pinning false here would commit a wrong durable spec on first create
	// and then a later retry would hit AlreadyExists and never converge to
	// the correct true value.
	t.Run("placement_group_enabled returns error when topology is unresolved", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-pending", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-pending", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			// VPC CR intentionally absent to exercise the NotFound path.
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeEKS,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-pending",
			NodePool: &dmpbv1.NodePool{
				InstanceType: "c8g.xlarge",
				K8SVersion:   "1.34",
				CpuCore:      8,
				MemoryGi:     16,
			},
		})
		require.Error(t, err, "missing VPC must surface as a retryable error, not a silently-pinned false")
		assert.Nil(t, resp)

		// And no NodePool CR must have been created; otherwise a later
		// retry would hit AlreadyExists and stay stuck on false.
		var npList v1alpha1.NodePoolList
		require.NoError(t, mockClient.List(context.Background(), &npList))
		assert.Empty(t, npList.Items)
	})

	// ACK/AKS dataplane: PG is out of scope per design. Wire value false is
	// accepted as a no-op; wire value true is rejected so a stale or buggy
	// caller cannot pin an unsupported flag onto the spec.
	t.Run("placement_group_enabled true on non-EKS is rejected (topology sanity check)", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-ack", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-ack", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{Name: "vpc-ack", Namespace: "ns"},
				Spec:       v1alpha1.VPCSpec{Zones: []string{"us-west-2a"}},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeACK,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-ack",
			NodePool: &dmpbv1.NodePool{
				InstanceType: "c8g.xlarge",
				K8SVersion:   "1.34",
				CpuCore:      8,
				MemoryGi:     16,
				// Producer (incorrectly) tries to enable PG for a non-EKS pool.
				PlacementGroupEnabled: true,
			},
		})
		require.Error(t, err, "non-EKS providers must reject placement_group_enabled=true")
		assert.Nil(t, resp)

		var npList v1alpha1.NodePoolList
		require.NoError(t, mockClient.List(context.Background(), &npList))
		assert.Empty(t, npList.Items, "a rejected CreateNodePool must not leave a CR behind")
	})

	t.Run("placement_group_enabled false on non-EKS is accepted", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{Name: "dp-ack-false", Namespace: "ns"},
				Status: v1alpha1.DataPlaneStatus{
					Phase: v1alpha1.DataPlanePhaseActive,
					Vpc:   &v1alpha1.Reference{Name: "vpc-ack-false", Namespace: "ns"},
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.VPC{
				ObjectMeta: metav1.ObjectMeta{Name: "vpc-ack-false", Namespace: "ns"},
				Spec:       v1alpha1.VPCSpec{Zones: []string{"us-west-2a"}},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{Name: "appbox-1", Namespace: "ns"},
				Spec: v1alpha1.APPBoxSpec{
					KubernetesConfig: v1alpha1.KubernetesConfig{Version: "1.34"},
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			repository.NewVPCReaderWriter(mockClient),
			nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					Provider:     consts.ProviderTypeACK,
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.CreateNodePool(context.Background(), &dmpbv1.CreateNodePoolRequest{
			DataPlaneId: "dp-ack-false",
			NodePool: &dmpbv1.NodePool{
				InstanceType: "c8g.xlarge",
				K8SVersion:   "1.34",
				CpuCore:      8,
				MemoryGi:     16,
				// Producer sends the default false; the service must accept
				// and persist it. The shared-pool gate on the CSNG side is
				// the authoritative driver for whether PG is enabled.
				PlacementGroupEnabled: false,
			},
		})
		require.NoError(t, err)
		require.NotNil(t, resp)
		assert.False(t, resp.PlacementGroupEnabled, "ACK/AKS pools must stay outside the PG scope")
	})
}

func TestDataPlaneService_CreateDataPlane(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = v1alpha1.AddToScheme(scheme)

	mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	cloudProviderExtractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
	svc := NewDataPlaneService(
		mockClient,
		repository.NewDataPlaneReaderWriter(mockClient),
		nil, nil, nil,
		cloudProviderExtractor,
		rsconfig.RootConfig{
			Namespace: "ns",
			ControllerConfig: rsconfig.ControllerConfig{
				RegionConfig: rsconfig.RegionConfig{
					Name:  "us-east-1",
					Zones: []string{"az1", "az2", "az3"},
				},
				DataPlaneRoleName:      "role",
				AWSDataPlaneExternalID: "extid",
				Provider:               "aws",
			},
		},
		rsconfig.TemplatesConfig{},
	)

	_, err := svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp1",
		DataPlane:   &dmpbv1.DataPlane{Vpc: &dmpbv1.DataPlane_Vpc{}},
	})
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "cidr is required")

	req := &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId:           "dp1",
		EnableCapacityPlanner: true,
		DataPlane: &dmpbv1.DataPlane{
			Vpc: &dmpbv1.DataPlane_Vpc{Cidr: "10.1.0.0/16"},
			Pca: &dmpbv1.DataPlane_PCA{
				CaIdentifier: "arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/pca-1",
			},
			Tags: map[string]string{
				"tag1":                           "val1",
				"tags.tidbcloud.com/tenant":      "12345",
				"tags.tidbcloud.com/servicetype": "byoc",
			},
		},
	}
	resp, err := svc.CreateDataPlane(context.Background(), req)
	assert.NoError(t, err)
	assert.Equal(t, "dp1", resp.Id)
	assert.Equal(t, "10.1.0.0/16", resp.Vpc.Cidr)
	assert.ElementsMatch(t, []string{"az1", "az2", "az3"}, resp.NodeZones)
	assert.Equal(t, dmpbv1.DataPlaneOwnershipMode_SINGLE_POOL_OWNERSHIP, resp.OwnershipMode)
	assert.NotNil(t, resp.Pca)
	assert.Equal(t, "arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/pca-1", resp.Pca.CaIdentifier)

	created := &v1alpha1.DataPlane{}
	err = mockClient.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: "dp1"}, created)
	assert.NoError(t, err)
	assert.Empty(t, created.Spec.NodeZones)
	assert.Equal(t, v1alpha1.DataPlaneOwnershipModeSinglePool, created.Spec.OwnershipMode)
	require.NotNil(t, created.Spec.PCA)
	assert.Equal(t, "arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/pca-1", created.Spec.PCA.CAIdentifier)
	assert.Equal(t, "byoc", created.Labels["tags.tidbcloud.com/servicetype"])
	assert.Equal(t, "byoc", created.Spec.CloudTags["tags.tidbcloud.com/servicetype"])
	assert.Equal(t, "true", created.Annotations[v1alpha1.AnnotationKeyEnableCapacityPlanner])

	_, err = svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp-invalid-pca",
		DataPlane: &dmpbv1.DataPlane{
			Vpc: &dmpbv1.DataPlane_Vpc{Cidr: "10.4.0.0/16"},
			Pca: &dmpbv1.DataPlane_PCA{
				CaIdentifier: "invalid-pca-id",
			},
			Tags: map[string]string{
				"tags.tidbcloud.com/tenant": "12345",
			},
		},
	})
	assert.ErrorContains(t, err, "invalid data_plane.pca.ca_identifier")

	_, err = svc.CreateDataPlane(context.Background(), req)
	assert.Error(t, err)
	assert.Contains(t, err.Error(), "dataplane already exists")

	_, err = svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp-rg-aws",
		DataPlane: &dmpbv1.DataPlane{
			Vpc:               &dmpbv1.DataPlane_Vpc{Cidr: "10.2.0.0/16"},
			ResourceGroupName: "existing-rg",
			Tags: map[string]string{
				"tags.tidbcloud.com/tenant": "12345",
			},
		},
	})
	assert.ErrorContains(t, err, "only supported for AKS dataplanes")

	aksClient := fake.NewClientBuilder().WithScheme(scheme).Build()
	aksSvc := NewDataPlaneService(
		aksClient,
		repository.NewDataPlaneReaderWriter(aksClient),
		nil, nil, nil,
		cloudProviderExtractor,
		rsconfig.RootConfig{
			Namespace: "ns",
			ControllerConfig: rsconfig.ControllerConfig{
				RegionConfig: rsconfig.RegionConfig{
					Name:  "eastus",
					Zones: []string{"az1", "az2", "az3"},
				},
				DataPlaneRoleName: "role",
				Provider:          consts.ProviderTypeAKS,
			},
		},
		rsconfig.TemplatesConfig{},
	)
	_, err = aksSvc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp-rg-aks",
		DataPlane: &dmpbv1.DataPlane{
			Vpc:               &dmpbv1.DataPlane_Vpc{Cidr: "10.3.0.0/16"},
			ResourceGroupName: "existing-rg",
			Tags: map[string]string{
				"tags.tidbcloud.com/tenant": "12345",
				"tag1":                      "val1",
			},
		},
	})
	require.NoError(t, err)

	createdWithRG := &v1alpha1.DataPlane{}
	err = aksClient.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: "dp-rg-aks"}, createdWithRG)
	require.NoError(t, err)
	assert.Equal(t, "existing-rg", createdWithRG.Spec.AccountRef.ResourceGroupName)
}

func TestDataPlaneService_CreateDataPlaneOwnershipMode(t *testing.T) {
	t.Parallel()

	t.Run("persists requested effective mode", func(t *testing.T) {
		t.Parallel()
		for _, tt := range []struct {
			name          string
			requestedMode dmpbv1.DataPlaneOwnershipMode
			storedMode    v1alpha1.DataPlaneOwnershipMode
		}{
			{
				name:       "unspecified defaults to single pool",
				storedMode: v1alpha1.DataPlaneOwnershipModeSinglePool,
			},
			{
				name:          "multi pool is explicit",
				requestedMode: dmpbv1.DataPlaneOwnershipMode_MULTI_POOL_OWNERSHIP,
				storedMode:    v1alpha1.DataPlaneOwnershipModeMultiPool,
			},
		} {
			t.Run(tt.name, func(t *testing.T) {
				t.Parallel()
				svc, kubeClient := newCreateDataPlaneTestService(t)
				req := validCreateDataPlaneRequest("dp-create-" + strings.ReplaceAll(tt.name, " ", "-"))
				req.DataPlane.OwnershipMode = tt.requestedMode

				resp, err := svc.CreateDataPlane(context.Background(), req)
				require.NoError(t, err)
				assert.NotEqual(t, dmpbv1.DataPlaneOwnershipMode_DATA_PLANE_OWNERSHIP_MODE_UNSPECIFIED, resp.OwnershipMode)

				created := new(v1alpha1.DataPlane)
				require.NoError(t, kubeClient.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: req.DataPlaneId}, created))
				assert.Equal(t, tt.storedMode, created.Spec.OwnershipMode)
				wantResponseMode, convertErr := dataPlaneOwnershipModeCR2API(tt.storedMode)
				require.NoError(t, convertErr)
				assert.Equal(t, wantResponseMode, resp.OwnershipMode)
			})
		}
	})

	t.Run("retry compares only effective ownership mode", func(t *testing.T) {
		t.Parallel()
		for _, tt := range []struct {
			name          string
			storedMode    v1alpha1.DataPlaneOwnershipMode
			requestedMode dmpbv1.DataPlaneOwnershipMode
			wantError     string
		}{
			{
				name:          "old absent and requested single are the same",
				requestedMode: dmpbv1.DataPlaneOwnershipMode_SINGLE_POOL_OWNERSHIP,
				wantError:     "dataplane already exists",
			},
			{
				name:          "stored and requested multi are the same",
				storedMode:    v1alpha1.DataPlaneOwnershipModeMultiPool,
				requestedMode: dmpbv1.DataPlaneOwnershipMode_MULTI_POOL_OWNERSHIP,
				wantError:     "dataplane already exists",
			},
			{
				name:          "single cannot be retried as multi",
				storedMode:    v1alpha1.DataPlaneOwnershipModeSinglePool,
				requestedMode: dmpbv1.DataPlaneOwnershipMode_MULTI_POOL_OWNERSHIP,
				wantError:     "dataplane ownership mode conflicts",
			},
			{
				name:       "multi cannot be retried as default single",
				storedMode: v1alpha1.DataPlaneOwnershipModeMultiPool,
				wantError:  "dataplane ownership mode conflicts",
			},
			{
				name:       "unknown stored mode fails closed",
				storedMode: "FutureOwnership",
				wantError:  "dataplane ownership mode conflicts",
			},
		} {
			t.Run(tt.name, func(t *testing.T) {
				t.Parallel()
				existing := &v1alpha1.DataPlane{
					ObjectMeta: metav1.ObjectMeta{Namespace: "ns", Name: "dp-retry"},
					Spec: v1alpha1.DataPlaneSpec{
						VpcCidr:       "192.168.0.0/16",
						OwnershipMode: tt.storedMode,
					},
				}
				svc, _ := newCreateDataPlaneTestService(t, existing)
				req := validCreateDataPlaneRequest(existing.Name)
				req.DataPlane.OwnershipMode = tt.requestedMode
				// Deliberately different from the stored create fields. Retry
				// compatibility compares only the effective ownership mode.
				req.DataPlane.Vpc.Cidr = "10.99.0.0/16"

				_, err := svc.CreateDataPlane(context.Background(), req)
				require.Error(t, err)
				assert.True(t, bizerr.Is(err, bizerr.ResourceConflict))
				assert.ErrorContains(t, err, tt.wantError)
			})
		}
	})

	t.Run("unknown public mode is rejected before persistence", func(t *testing.T) {
		t.Parallel()
		svc, kubeClient := newCreateDataPlaneTestService(t)
		req := validCreateDataPlaneRequest("dp-unknown-request")
		req.DataPlane.OwnershipMode = dmpbv1.DataPlaneOwnershipMode(99)

		_, err := svc.CreateDataPlane(context.Background(), req)
		require.Error(t, err)
		assert.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
		assert.ErrorContains(t, err, "unknown dataplane ownership mode")
		getErr := kubeClient.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: req.DataPlaneId}, new(v1alpha1.DataPlane))
		assert.True(t, apierrors.IsNotFound(getErr))
	})
}

func newCreateDataPlaneTestService(t *testing.T, objects ...client.Object) (*DataPlaneService, client.Client) {
	t.Helper()
	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1.AddToScheme(scheme))
	kubeClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(objects...).Build()
	return NewDataPlaneService(
		kubeClient,
		repository.NewDataPlaneReaderWriter(kubeClient),
		nil, nil, nil, nil,
		rsconfig.RootConfig{
			Namespace: "ns",
			ControllerConfig: rsconfig.ControllerConfig{
				RegionConfig: rsconfig.RegionConfig{
					Name:  "us-east-1",
					Zones: []string{"az1", "az2", "az3"},
				},
				DataPlaneRoleName: "role",
				Provider:          consts.ProviderTypeEKS,
			},
		},
		rsconfig.TemplatesConfig{},
	), kubeClient
}

func validCreateDataPlaneRequest(id string) *dmpbv1.CreateDataPlaneRequest {
	return &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: id,
		DataPlane: &dmpbv1.DataPlane{
			Vpc: &dmpbv1.DataPlane_Vpc{Cidr: "10.1.0.0/16"},
			Tags: map[string]string{
				consts.TenantLabelKey: "tenant-1",
			},
		},
	}
}

func TestDataPlaneService_CreateDataPlaneValidatesACKPrivateLinkEndpointCIDR(t *testing.T) {
	scheme := runtime.NewScheme()
	require.NoError(t, v1alpha1.AddToScheme(scheme))
	mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()
	svc := NewDataPlaneService(
		mockClient,
		repository.NewDataPlaneReaderWriter(mockClient),
		nil, nil, nil, nil,
		rsconfig.RootConfig{
			Namespace: "ns",
			ControllerConfig: rsconfig.ControllerConfig{
				Provider: consts.ProviderTypeACK,
				RegionConfig: rsconfig.RegionConfig{
					Name:      "cn-hangzhou",
					Zones:     []string{"cn-hangzhou-b", "cn-hangzhou-e"},
					NodeZones: []string{"cn-hangzhou-b"},
				},
			},
		},
		rsconfig.TemplatesConfig{},
	)

	for _, tc := range []struct {
		name string
		cidr string
	}{
		{name: "CIDR is not IPv4", cidr: "2001:db8::/20"},
		{name: "CIDR is malformed", cidr: "not-a-cidr"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			_, err := svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
				DataPlaneId: "dp-" + tc.name,
				DataPlane: &dmpbv1.DataPlane{
					Vpc: &dmpbv1.DataPlane_Vpc{Cidr: tc.cidr},
					Tags: map[string]string{
						consts.TenantLabelKey: "tenant-1",
					},
				},
			})
			require.Error(t, err)
			assert.ErrorContains(t, err, "ACK private link endpoint subnets require an IPv4 VPC CIDR")
			assert.True(t, bizerr.Is(err, bizerr.ParameterInvalid))
		})
	}

	_, err := svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp-valid",
		DataPlane: &dmpbv1.DataPlane{
			Vpc: &dmpbv1.DataPlane_Vpc{Cidr: "10.0.0.0/22"},
			Tags: map[string]string{
				consts.TenantLabelKey: "tenant-1",
			},
		},
	})
	require.NoError(t, err)

	_, err = svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp-multi-az-ipv6",
		DataPlane: &dmpbv1.DataPlane{
			Vpc:                  &dmpbv1.DataPlane_Vpc{Cidr: "2001:db8::/20"},
			NodeZones:            []string{"cn-hangzhou-b", "cn-hangzhou-e"},
			HighAvailabilityType: dmpbv1.DataPlane_REGIONAL,
			Tags: map[string]string{
				consts.TenantLabelKey: "tenant-1",
			},
		},
	})
	require.NoError(t, err)
	multiAZDataPlane := &v1alpha1.DataPlane{}
	require.NoError(t, mockClient.Get(context.Background(), client.ObjectKey{
		Namespace: "ns",
		Name:      "dp-multi-az-ipv6",
	}, multiAZDataPlane))
	assert.Equal(t, []string{"cn-hangzhou-b", "cn-hangzhou-e"}, multiAZDataPlane.Spec.NodeZones)

	svc.config.BYOC.BYOCPrivateLink.ControlPlaneVPCID = "vpc-control-plane"
	_, err = svc.CreateDataPlane(context.Background(), &dmpbv1.CreateDataPlaneRequest{
		DataPlaneId: "dp-byoc-single-az-ipv6",
		DataPlane: &dmpbv1.DataPlane{
			Vpc:                  &dmpbv1.DataPlane_Vpc{Cidr: "2001:db8::/20"},
			NodeZones:            []string{"cn-hangzhou-b"},
			HighAvailabilityType: dmpbv1.DataPlane_ZONAL,
			Tags: map[string]string{
				consts.TenantLabelKey: "tenant-1",
			},
		},
	})
	require.NoError(t, err)
	byocDataPlane := &v1alpha1.DataPlane{}
	require.NoError(t, mockClient.Get(context.Background(), client.ObjectKey{
		Namespace: "ns",
		Name:      "dp-byoc-single-az-ipv6",
	}, byocDataPlane))
	assert.Equal(t, []string{"cn-hangzhou-b"}, byocDataPlane.Spec.NodeZones)
}

func TestDataPlaneService_UpdateDataPlaneCloudTags(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = v1alpha1.AddToScheme(scheme)

	existing := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "dp1",
			Namespace: "ns",
			Labels: map[string]string{
				"existing":                    "keep",
				"tags.tidbcloud.com/owner":    "system-owner",
				"tags.tidbcloud.com/tenant":   "tenant-1",
				"tags.tidbcloud.com/obsolete": "kept-on-labels",
			},
		},
		Spec: v1alpha1.DataPlaneSpec{
			CloudTags: map[string]string{
				"tags.tidbcloud.com/owner": "system-owner",
				"tags.tidbcloud.com/old":   "old-value",
			},
		},
	}
	mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(existing).Build()

	ctrl := gomock.NewController(t)
	defer ctrl.Finish()
	cloudProviderExtractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
	cloudProviderExtractor.EXPECT().GetDataPlaneDomain().Return("").AnyTimes()

	svc := NewDataPlaneService(
		mockClient,
		repository.NewDataPlaneReaderWriter(mockClient),
		nil, nil, nil,
		cloudProviderExtractor,
		rsconfig.RootConfig{Namespace: "ns"},
		rsconfig.TemplatesConfig{},
	)

	_, err := svc.UpdateDataPlaneCloudTags(context.Background(), &dmpbv1.UpdateDataPlaneCloudTagsRequest{
		DataPlaneId: "dp1",
		Tags: map[string]string{
			"tags.tidbcloud.com/tenant": "tenant-1",
			"tags.tidbcloud.com/owner":  "",
		},
	})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "empty value")

	_, err = svc.UpdateDataPlaneCloudTags(context.Background(), &dmpbv1.UpdateDataPlaneCloudTagsRequest{
		DataPlaneId: "dp1",
		Tags: map[string]string{
			"tags.tidbcloud.com/owner": "custom-owner",
		},
	})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "missing required tag")

	resp, err := svc.UpdateDataPlaneCloudTags(context.Background(), &dmpbv1.UpdateDataPlaneCloudTagsRequest{
		DataPlaneId: "dp1",
		Tags: map[string]string{
			"tags.tidbcloud.com/tenant":      "tenant-1",
			"tags.tidbcloud.com/owner":       "custom-owner",
			"tags.tidbcloud.com/cost-center": "cc-1",
		},
	})
	require.NoError(t, err)
	assert.Equal(t, map[string]string{
		"tags.tidbcloud.com/tenant":      "tenant-1",
		"tags.tidbcloud.com/owner":       "custom-owner",
		"tags.tidbcloud.com/cost-center": "cc-1",
	}, resp.GetTags())

	updated := &v1alpha1.DataPlane{}
	err = mockClient.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: "dp1"}, updated)
	require.NoError(t, err)
	assert.Equal(t, map[string]string{
		"tags.tidbcloud.com/tenant":      "tenant-1",
		"tags.tidbcloud.com/owner":       "custom-owner",
		"tags.tidbcloud.com/cost-center": "cc-1",
	}, updated.Spec.CloudTags)
	assert.Equal(t, "keep", updated.Labels["existing"])
	assert.Equal(t, "custom-owner", updated.Labels["tags.tidbcloud.com/owner"])
	assert.Equal(t, "tenant-1", updated.Labels["tags.tidbcloud.com/tenant"])
	assert.Equal(t, "kept-on-labels", updated.Labels["tags.tidbcloud.com/obsolete"])
	assert.Equal(t, "cc-1", updated.Labels["tags.tidbcloud.com/cost-center"])
}

func TestDataPlaneService_resolveNodeZonesForCreate(t *testing.T) {
	svc := &DataPlaneService{
		config: rsconfig.RootConfig{
			ControllerConfig: rsconfig.ControllerConfig{
				RegionConfig: rsconfig.RegionConfig{
					Zones: []string{"az1", "az2", "az3"},
				},
			},
		},
	}

	t.Run("empty request and empty config node zones keeps node zones unset", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(nil, dmpbv1.DataPlane_HIGH_AVAILABILITY_TYPE_UNSPECIFIED)
		assert.NoError(t, err)
		assert.Nil(t, zones)
	})

	t.Run("duplicate node zones in request should fail", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{"az1", "az1"},
			dmpbv1.DataPlane_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
		)
		assert.Error(t, err)
		assert.Nil(t, zones)
		assert.Contains(t, err.Error(), "duplicate zone")
	})

	t.Run("strict subset check should fail for full zone set", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{"az1", "az2", "az3"},
			dmpbv1.DataPlane_HIGH_AVAILABILITY_TYPE_UNSPECIFIED,
		)
		assert.Error(t, err)
		assert.Nil(t, zones)
		assert.Contains(t, err.Error(), "strict subset")
	})

	t.Run("regional high availability with three caller-provided node zones passes through", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{"az1", "az2", "az3"},
			dmpbv1.DataPlane_REGIONAL,
		)
		assert.NoError(t, err)
		assert.Equal(t, []string{"az1", "az2", "az3"}, zones)
	})

	t.Run("regional high availability ignores configured node zone restriction", func(t *testing.T) {
		regionalSvc := &DataPlaneService{
			config: rsconfig.RootConfig{
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{
						Zones:     []string{"az1", "az2", "az3"},
						NodeZones: []string{"az2"},
					},
				},
			},
		}
		zones, err := regionalSvc.resolveNodeZonesForCreate(nil, dmpbv1.DataPlane_REGIONAL)
		assert.NoError(t, err)
		assert.Nil(t, zones)
	})

	t.Run("regional high availability with empty caller zones passes through regardless of regionConfig.zones count", func(t *testing.T) {
		// dataplane-manager no longer rejects regionConfig shape at API time.
		// Empty req_node_zones signals "use regionConfig baseline" and lets
		// bring-up own the validation. This replaces the legacy "requires three
		// configured baseline zones" case, which used to reject any
		// regionConfig.zones whose length != 3.
		regionalSvc := &DataPlaneService{
			config: rsconfig.RootConfig{
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{
						Zones: []string{"az1", "az2"},
					},
				},
			},
		}
		zones, err := regionalSvc.resolveNodeZonesForCreate(nil, dmpbv1.DataPlane_REGIONAL)
		assert.NoError(t, err)
		assert.Nil(t, zones)
	})

	t.Run("regional high availability with one caller zone is rejected as semantically zonal", func(t *testing.T) {
		regionalSvc := &DataPlaneService{
			config: rsconfig.RootConfig{
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{
						Zones: []string{"az1", "az2"},
					},
				},
			},
		}
		zones, err := regionalSvc.resolveNodeZonesForCreate(
			[]string{"az1"}, dmpbv1.DataPlane_REGIONAL,
		)
		assert.Error(t, err)
		assert.Nil(t, zones)
		assert.Contains(t, err.Error(), "1 zone")
	})

	t.Run("zonal high availability accepts one explicit node zone", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{"az1"},
			dmpbv1.DataPlane_ZONAL,
		)
		assert.NoError(t, err)
		assert.Equal(t, []string{"az1"}, zones)
	})

	t.Run("zonal high availability with empty node zones auto-fills from configured regionConfig.nodeZones", func(t *testing.T) {
		zonalSvc := &DataPlaneService{
			config: rsconfig.RootConfig{
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{
						Zones:     []string{"az1", "az2", "az3"},
						NodeZones: []string{"az2"},
					},
				},
			},
		}
		zones, err := zonalSvc.resolveNodeZonesForCreate(nil, dmpbv1.DataPlane_ZONAL)
		assert.NoError(t, err)
		assert.Equal(t, []string{"az2"}, zones)
	})

	t.Run("zonal high availability with empty node zones and unconfigured regionConfig.nodeZones fails the auto-fill guard", func(t *testing.T) {
		// svc here is the function-level fixture: Zones=[az1,az2,az3], NodeZones unset.
		zones, err := svc.resolveNodeZonesForCreate(nil, dmpbv1.DataPlane_ZONAL)
		assert.Error(t, err)
		assert.Nil(t, zones)
		assert.Contains(t, err.Error(), "regionConfig.nodeZones must contain exactly 1 zone")
	})

	t.Run("zonal high availability with empty node zones and malformed regionConfig.nodeZones fails the auto-fill guard", func(t *testing.T) {
		zonalSvc := &DataPlaneService{
			config: rsconfig.RootConfig{
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{
						Zones:     []string{"az1", "az2", "az3"},
						NodeZones: []string{"az1", "az2"},
					},
				},
			},
		}
		zones, err := zonalSvc.resolveNodeZonesForCreate(nil, dmpbv1.DataPlane_ZONAL)
		assert.Error(t, err)
		assert.Nil(t, zones)
		assert.Contains(t, err.Error(), "regionConfig.nodeZones must contain exactly 1 zone")
	})

	t.Run("regional high availability with two caller-provided node zones passes through", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{"az1", "az2"},
			dmpbv1.DataPlane_REGIONAL,
		)
		assert.NoError(t, err)
		assert.Equal(t, []string{"az1", "az2"}, zones)
	})

	t.Run("regional high availability rejects exactly one caller-provided node zone as semantically zonal", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{"az2"},
			dmpbv1.DataPlane_REGIONAL,
		)
		assert.Error(t, err)
		assert.Nil(t, zones)
		assert.Contains(t, err.Error(), "1 zone")
	})

	t.Run("regional high availability with empty caller zones returns nil", func(t *testing.T) {
		zones, err := svc.resolveNodeZonesForCreate(
			[]string{},
			dmpbv1.DataPlane_REGIONAL,
		)
		assert.NoError(t, err)
		assert.Nil(t, zones)
	})
}

func TestDataPlaneService_LegacyDeleteNodePool(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = k8sscheme.AddToScheme(scheme)
	_ = v1alpha1.AddToScheme(scheme)
	_ = eksv1beta1.AddToScheme(scheme)
	_ = ec2v1beta1.AddToScheme(scheme)
	_ = ackv1alpha1.AddToScheme(scheme)

	t.Run("delete EKS nodegroups successfully", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-1",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.34"),
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}
		ng2 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-2",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.34"),
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}
		ng3 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-35-1",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.35"),
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.35"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(2),
						},
					},
				},
			},
		}

		lt1 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-1",
			},
		}
		lt2 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-2",
			},
		}
		lt3 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-35-1",
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			ng1, ng2, ng3, lt1, lt2, lt3,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			K8SVersion:  "1.34",
		})

		assert.NoError(t, err)
		assert.NotNil(t, resp)

		// Verify that only nodegroups with version 1.34 were deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 1)
		assert.Equal(t, "ng-1-35-1", ngList.Items[0].Name)
		assert.Equal(t, "1.35", *ngList.Items[0].Spec.ForProvider.Version)

		// Verify that only launch templates for version 1.34 were deleted
		var ltList ec2v1beta1.LaunchTemplateList
		err = mockClient.List(context.Background(), &ltList)
		assert.NoError(t, err)
		assert.Len(t, ltList.Items, 1)
		assert.Equal(t, "ng-1-35-1", ltList.Items[0].Name)
	})

	t.Run("delete EKS nodegroups - error when nodegroup has nodes", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-1",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.34"),
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(2),
						},
					},
				},
			},
		}
		ng2 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-2",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.34"),
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			ng1, ng2,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			K8SVersion:  "1.34",
		})

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "still has nodes")
		assert.Contains(t, err.Error(), "ng-1-34-1")

		// Verify that no nodegroups were deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 2)
	})

	t.Run("delete EKS nodegroups - error when spec and status versions mismatch", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-1",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.34"),
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.33"), // status version is different from spec
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			ng1,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			K8SVersion:  "1.34",
		})

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "version in status")
		assert.Contains(t, err.Error(), "does not match")
		assert.Contains(t, err.Error(), "1.33")
		assert.Contains(t, err.Error(), "1.34")

		// Verify that no nodegroups were deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 1)
		assert.Equal(t, "ng-1-34-1", ngList.Items[0].Name)
	})

	t.Run("delete ACK nodepools - should return error not implemented", func(t *testing.T) {
		np1 := &ackv1alpha1.KubernetesNodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name: "np-1-34-1",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeACK,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			np1,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			K8SVersion:  "1.34",
		})

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "method is not implemented")

		// Verify that nodepools were NOT deleted (ACK upgrades in-place)
		var npList ackv1alpha1.KubernetesNodePoolList
		err = mockClient.List(context.Background(), &npList)
		assert.NoError(t, err)
		assert.Len(t, npList.Items, 1)
		assert.Equal(t, "np-1-34-1", npList.Items[0].Name)
	})

	t.Run("delete nodepools with non-existent dataplane", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "non-existent",
			K8SVersion:  "1.34",
		})

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "failed to get dataplane")
	})

	t.Run("delete nodepools with unsupported provider", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: "unsupported-provider",
				},
			},
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			K8SVersion:  "1.34",
		})

		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "unsupported provider")
	})

	t.Run("delete nodepools - no matching version", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-35-1",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.35"),
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
				},
			},
			ng1,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			K8SVersion:  "1.34",
		})

		assert.NoError(t, err)
		assert.NotNil(t, resp)

		// Verify that no nodegroups were deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 1)
		assert.Equal(t, "ng-1-35-1", ngList.Items[0].Name)
	})

	t.Run("delete EKS nodegroups with specific instance type", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-c5-xlarge",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version:       pointer.String("1.34"),
					InstanceTypes: []*string{pointer.String("c5.xlarge")},
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}
		ng2 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-c5-2xlarge",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version:       pointer.String("1.34"),
					InstanceTypes: []*string{pointer.String("c5.2xlarge")},
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}
		ng3 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-m5-xlarge",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version:       pointer.String("1.34"),
					InstanceTypes: []*string{pointer.String("m5.xlarge")},
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}

		lt1 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-c5-xlarge",
			},
		}
		lt2 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-c5-2xlarge",
			},
		}
		lt3 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-m5-xlarge",
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			ng1, ng2, ng3, lt1, lt2, lt3,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId:  "dp-1",
			K8SVersion:   "1.34",
			InstanceType: "c5.xlarge",
		})

		assert.NoError(t, err)
		assert.NotNil(t, resp)

		// Verify that only the nodegroup with c5.xlarge was deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 2)

		ngNames := make([]string, 0, len(ngList.Items))
		for _, ng := range ngList.Items {
			ngNames = append(ngNames, ng.Name)
		}
		assert.Contains(t, ngNames, "ng-1-34-c5-2xlarge")
		assert.Contains(t, ngNames, "ng-1-34-m5-xlarge")
		assert.NotContains(t, ngNames, "ng-1-34-c5-xlarge")

		// Verify that only the launch template for c5.xlarge was deleted
		var ltList ec2v1beta1.LaunchTemplateList
		err = mockClient.List(context.Background(), &ltList)
		assert.NoError(t, err)
		assert.Len(t, ltList.Items, 2)

		ltNames := make([]string, 0, len(ltList.Items))
		for _, lt := range ltList.Items {
			ltNames = append(ltNames, lt.Name)
		}
		assert.Contains(t, ltNames, "ng-1-34-c5-2xlarge")
		assert.Contains(t, ltNames, "ng-1-34-m5-xlarge")
		assert.NotContains(t, ltNames, "ng-1-34-c5-xlarge")
	})

	t.Run("delete EKS nodegroups with instance type - no matching instance type", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-c5-xlarge",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version:       pointer.String("1.34"),
					InstanceTypes: []*string{pointer.String("c5.xlarge")},
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			ng1,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId:  "dp-1",
			K8SVersion:   "1.34",
			InstanceType: "m5.xlarge", // This instance type doesn't exist
		})

		assert.NoError(t, err)
		assert.NotNil(t, resp)

		// Verify that no nodegroups were deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 1)
		assert.Equal(t, "ng-1-34-c5-xlarge", ngList.Items[0].Name)
	})

	t.Run("delete EKS nodegroups with instance type in multiple instance types array", func(t *testing.T) {
		ng1 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-multi",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version: pointer.String("1.34"),
					InstanceTypes: []*string{
						pointer.String("c5.xlarge"),
						pointer.String("c5.2xlarge"),
						pointer.String("m5.xlarge"),
					},
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}
		ng2 := &eksv1beta1.NodeGroup{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-single",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: eksv1beta1.NodeGroupSpec{
				ForProvider: eksv1beta1.NodeGroupParameters{
					Version:       pointer.String("1.34"),
					InstanceTypes: []*string{pointer.String("r5.xlarge")},
				},
			},
			Status: eksv1beta1.NodeGroupStatus{
				AtProvider: eksv1beta1.NodeGroupObservation{
					Version: pointer.String("1.34"),
					ScalingConfig: []eksv1beta1.ScalingConfigObservation{
						{
							DesiredSize: pointer.Float64(0),
						},
					},
				},
			},
		}

		lt1 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-multi",
			},
		}
		lt2 := &ec2v1beta1.LaunchTemplate{
			ObjectMeta: metav1.ObjectMeta{
				Name: "ng-1-34-single",
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(
			&v1alpha1.DataPlane{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "dp-1",
					Namespace: "ns",
				},
				Status: v1alpha1.DataPlaneStatus{
					AppBox: &v1alpha1.Reference{
						Name:      "appbox-1",
						Namespace: "ns",
					},
				},
			},
			&v1alpha1.APPBox{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "appbox-1",
					Namespace: "ns",
				},
				Spec: v1alpha1.APPBoxSpec{
					Provider: consts.ProviderTypeEKS,
					KubernetesConfig: v1alpha1.KubernetesConfig{
						Version: "1.35",
					},
				},
			},
			ng1, ng2, lt1, lt2,
		).Build()

		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, repository.NewAPPBoxReaderWriter(mockClient),
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.LegacyDeleteNodePool(context.Background(), &dmpbv1.LegacyDeleteNodePoolRequest{
			DataPlaneId:  "dp-1",
			K8SVersion:   "1.34",
			InstanceType: "c5.2xlarge",
		})

		assert.NoError(t, err)
		assert.NotNil(t, resp)

		// Verify that only the nodegroup with c5.2xlarge in its instance types was deleted
		var ngList eksv1beta1.NodeGroupList
		err = mockClient.List(context.Background(), &ngList)
		assert.NoError(t, err)
		assert.Len(t, ngList.Items, 1)
		assert.Equal(t, "ng-1-34-single", ngList.Items[0].Name)

		// Verify that only the launch template for the multi-instance nodegroup was deleted
		var ltList ec2v1beta1.LaunchTemplateList
		err = mockClient.List(context.Background(), &ltList)
		assert.NoError(t, err)
		assert.Len(t, ltList.Items, 1)
		assert.Equal(t, "ng-1-34-single", ltList.Items[0].Name)
	})
}

func TestDataPlaneService_DeleteNodePool(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = v1alpha1.AddToScheme(scheme)

	t.Run("delete nodepool by name successfully", func(t *testing.T) {
		np1 := &v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "np-1",
				Namespace: "ns",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
		}
		np2 := &v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "np-2",
				Namespace: "ns",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(np1, np2).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.DeleteNodePool(context.Background(), &dmpbv1.DeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePoolId:  "np-1",
		})
		assert.NoError(t, err)
		assert.NotNil(t, resp)

		var npList v1alpha1.NodePoolList
		err = mockClient.List(context.Background(), &npList)
		assert.NoError(t, err)
		assert.Len(t, npList.Items, 1)
		assert.Equal(t, "np-2", npList.Items[0].Name)
	})

	t.Run("delete nodepool by name - not found", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.DeleteNodePool(context.Background(), &dmpbv1.DeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePoolId:  "np-1",
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "nodepool not found")
	})

	t.Run("delete nodepool by name - dataplane mismatch", func(t *testing.T) {
		np1 := &v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "np-1",
				Namespace: "ns",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-2",
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(np1).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.DeleteNodePool(context.Background(), &dmpbv1.DeleteNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePoolId:  "np-1",
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "nodepool not found")
	})

	t.Run("delete nodepool by name - missing params", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.DeleteNodePool(context.Background(), &dmpbv1.DeleteNodePoolRequest{
			DataPlaneId: "dp-1",
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "node_pool_id")
	})
}

func TestDataPlaneService_UpdateNodePool(t *testing.T) {
	scheme := runtime.NewScheme()
	_ = v1alpha1.AddToScheme(scheme)

	t.Run("update nodepool labels and taints", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		cloudProviderExtractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)

		np1 := &v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "np-1",
				Namespace: "ns",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-1",
				},
			},
			Spec: v1alpha1.NodePoolSpec{
				Labels: map[string]string{"a": "b"},
				Taints: []v1alpha1.NodePoolTaint{{Key: "old", Value: "old", Effect: "NoSchedule"}},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(np1).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			cloudProviderExtractor,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.UpdateNodePool(context.Background(), &dmpbv1.UpdateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePoolId:  "np-1",
			NodePool: &dmpbv1.NodePool{
				Labels: map[string]string{"env": "prod"},
				Taints: []*dmpbv1.NodePoolTaint{
					{Key: "dedicated", Value: "true", Effect: "NoSchedule"},
				},
			},
		})
		assert.NoError(t, err)
		assert.NotNil(t, resp)
		assert.Equal(t, "prod", resp.Labels["env"])
		assert.Len(t, resp.Taints, 1)
		assert.Equal(t, "dedicated", resp.Taints[0].Key)

		np := &v1alpha1.NodePool{}
		err = mockClient.Get(context.Background(), client.ObjectKey{Namespace: "ns", Name: "np-1"}, np)
		assert.NoError(t, err)
		assert.Equal(t, map[string]string{"env": "prod"}, np.Spec.Labels)
		assert.Len(t, np.Spec.Taints, 1)
		assert.Equal(t, "dedicated", np.Spec.Taints[0].Key)
	})

	t.Run("update nodepool - not found", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		cloudProviderExtractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)

		mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			cloudProviderExtractor,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.UpdateNodePool(context.Background(), &dmpbv1.UpdateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePoolId:  "np-1",
			NodePool:    &dmpbv1.NodePool{},
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "nodepool not found")
	})

	t.Run("update nodepool - dataplane mismatch", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		cloudProviderExtractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)

		np1 := &v1alpha1.NodePool{
			ObjectMeta: metav1.ObjectMeta{
				Name:      "np-1",
				Namespace: "ns",
				Labels: map[string]string{
					v1alpha1.DataPlaneIDKey: "dp-2",
				},
			},
		}

		mockClient := fake.NewClientBuilder().WithScheme(scheme).WithObjects(np1).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			cloudProviderExtractor,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.UpdateNodePool(context.Background(), &dmpbv1.UpdateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePoolId:  "np-1",
			NodePool:    &dmpbv1.NodePool{},
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "nodepool not found")
	})

	t.Run("update nodepool - missing params", func(t *testing.T) {
		mockClient := fake.NewClientBuilder().WithScheme(scheme).Build()
		svc := NewDataPlaneService(
			mockClient,
			repository.NewDataPlaneReaderWriter(mockClient),
			nil, nil, nil,
			nil,
			rsconfig.RootConfig{
				Namespace:          "ns",
				ResourceNamePrefix: "tidbcloud",
				ControllerConfig: rsconfig.ControllerConfig{
					RegionConfig: rsconfig.RegionConfig{Name: "us-east-1"},
				},
			},
			rsconfig.TemplatesConfig{},
		)

		resp, err := svc.UpdateNodePool(context.Background(), &dmpbv1.UpdateNodePoolRequest{
			DataPlaneId: "dp-1",
			NodePool:    &dmpbv1.NodePool{},
		})
		assert.Error(t, err)
		assert.Nil(t, resp)
		assert.Contains(t, err.Error(), "node_pool_id")
	})
}

func TestDataPlaneService_ListNodePoolConfigs(t *testing.T) {
	cfg := rsconfig.RootConfig{
		ControllerConfig: rsconfig.ControllerConfig{
			Provider: consts.ProviderTypeEKS,
			RegionConfig: rsconfig.RegionConfig{
				Zones: []string{"us-west-2a", "us-west-2b", "us-west-2c"},
			},
			APPBoxConfig: rsconfig.APPBoxConfig{
				SharedNodePools: []rsconfig.NodePoolConfig{
					{InstanceType: "t2.medium", CPU: 2, AvailableCPU: 1, MemoryGi: 4, AvailableMemoryMi: 3000},
				},
				NodePools: []rsconfig.NodePoolConfig{
					{InstanceType: "t3.large", CPU: 4, AvailableCPU: 3, MemoryGi: 8, AvailableMemoryMi: 6000},
				},
			},
		},
	}
	svc := NewDataPlaneService(nil, nil, nil, nil, nil, nil, cfg, rsconfig.TemplatesConfig{})
	resp, err := svc.ListNodePoolConfigs(context.Background(), nil)
	assert.NoError(t, err)
	assert.Len(t, resp.NodePoolConfigs, 2)
	assert.Equal(t, []string{"us-west-2a", "us-west-2b", "us-west-2c"}, resp.GetAvailableNodeZones())
	require.Equal(t, dmpbv1.NodeType_Dedicated, resp.NodePoolConfigs[1].GetNodeType())
	assert.Equal(t, uint32(3000), resp.NodePoolConfigs[1].GetAvailableCpuMilliCore())
	assert.Equal(t, uint32(3000), resp.NodePoolConfigs[1].GetMaxCpuMilli())
	assert.Equal(t, uint32(6000), resp.NodePoolConfigs[1].GetMaxMemoryMi())
}

func Test_dataPlaneCapabilities(t *testing.T) {
	// Multi-port LB clients probe for "loadbalancer.multi-port" on the
	// DataPlane response before submitting requests that set Ports.
	caps := dataPlaneCapabilities()
	assert.Contains(t, caps, "loadbalancer.multi-port")
}

func Test_dataPlaneCR2API(t *testing.T) {
	cr := &v1alpha1.DataPlane{
		ObjectMeta: metav1.ObjectMeta{Name: "dp1"},
		Spec: v1alpha1.DataPlaneSpec{
			AccountRef: v1alpha1.AccountReference{
				AccountID:         "acc1",
				ResourceGroupName: "existing-rg",
			},
			PCA: &v1alpha1.PCAConfig{
				CAIdentifier: "arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/pca-1",
			},
			VpcCidr:           "10.0.0.0/16",
			AvailabilityZones: []string{"az1", "az2"},
			CloudTags:         map[string]string{"tag": "val"},
		},
		Status: v1alpha1.DataPlaneStatus{Phase: v1alpha1.DataPlanePhaseActive},
	}
	api, err := dataPlaneCR2API(cr)
	require.NoError(t, err)
	assert.Equal(t, "dp1", api.Id)
	assert.Equal(t, "acc1", api.AccountId)
	assert.Equal(t, "10.0.0.0/16", api.Vpc.Cidr)
	assert.Equal(t, dmpbv1.DataPlane_Active, api.Status)
	assert.Equal(t, "existing-rg", api.ResourceGroupName)
	assert.ElementsMatch(t, []string{"az1", "az2"}, api.NodeZones)
	assert.Equal(t, "val", api.Tags["tag"])
	require.NotNil(t, api.Pca)
	assert.Equal(t, "arn:aws:acm-pca:us-east-1:123456789012:certificate-authority/pca-1", api.Pca.CaIdentifier)
	assert.Equal(t, dmpbv1.DataPlaneOwnershipMode_SINGLE_POOL_OWNERSHIP, api.OwnershipMode)
	assert.Empty(t, cr.Spec.OwnershipMode, "response defaulting must not backfill old CRs")

	cr.Spec.OwnershipMode = v1alpha1.DataPlaneOwnershipModeMultiPool
	api, err = dataPlaneCR2API(cr)
	require.NoError(t, err)
	assert.Equal(t, dmpbv1.DataPlaneOwnershipMode_MULTI_POOL_OWNERSHIP, api.OwnershipMode)

	cr.Spec.OwnershipMode = "FutureOwnership"
	_, err = dataPlaneCR2API(cr)
	require.ErrorContains(t, err, "unknown stored dataplane ownership mode")
}

func Test_fillVPC2DataPlane(t *testing.T) {
	t.Run("expose workload zones only", func(t *testing.T) {
		svc := &DataPlaneService{}
		vpc := &v1alpha1.VPC{
			Spec: v1alpha1.VPCSpec{
				Zones:     []string{"az1", "az2", "az3"},
				NodeZones: []string{"az1", "az2"},
			},
			Status: v1alpha1.VPCStatus{
				Phase: v1alpha1.VPCPhaseActive,
				NodeSubnets: []v1alpha1.Subnet{
					{AvailabilityZone: "az1", AvailabilityZoneID: "azid1"},
					{AvailabilityZone: "az2", AvailabilityZoneID: "azid2"},
					{AvailabilityZone: "az3", AvailabilityZoneID: "azid3"},
				},
			},
		}
		dp := &dmpbv1.DataPlane{Vpc: &dmpbv1.DataPlane_Vpc{}}

		svc.fillVPC2DataPlane(vpc, dp)

		assert.Equal(t, []string{"az1", "az2"}, dp.Vpc.AvailabilityZones)
		assert.Equal(t, []string{"azid1", "azid2"}, dp.Vpc.AvailabilityZoneIds)
	})

	t.Run("fallback to zones when workload zones empty", func(t *testing.T) {
		svc := &DataPlaneService{}
		vpc := &v1alpha1.VPC{
			Spec: v1alpha1.VPCSpec{
				Zones: []string{"az1", "az2", "az3"},
			},
			Status: v1alpha1.VPCStatus{
				Phase: v1alpha1.VPCPhaseActive,
				NodeSubnets: []v1alpha1.Subnet{
					{AvailabilityZone: "az1", AvailabilityZoneID: "azid1"},
					{AvailabilityZone: "az2", AvailabilityZoneID: "azid2"},
					{AvailabilityZone: "az3", AvailabilityZoneID: "azid3"},
				},
			},
		}
		dp := &dmpbv1.DataPlane{Vpc: &dmpbv1.DataPlane_Vpc{}}

		svc.fillVPC2DataPlane(vpc, dp)

		assert.Equal(t, []string{"az1", "az2", "az3"}, dp.Vpc.AvailabilityZones)
		assert.Equal(t, []string{"azid1", "azid2", "azid3"}, dp.Vpc.AvailabilityZoneIds)
	})
}

func Test_fillAuthTokenApiGateway2DataPlane(t *testing.T) {
	mockAppBox := &v1alpha1.APPBox{}
	mockAppBox.Name = "mock-appbox"
	mockAppBox.Namespace = "mock-namespace"
	mockDataPlane := &dmpbv1.DataPlane{
		Id:        "mock-dataplane",
		AccountId: "abc",
	}

	t.Run("get api gateway error", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		mockErr := errors.New("get api gateway error")
		extractor.EXPECT().GetDataPlaneApiGateway(gomock.Any(), gomock.Any()).Return("", mockErr)
		svc := &DataPlaneService{cloudProviderExtractor: extractor}
		err := svc.fillAuthTokenApiGateway2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.Error(t, err) {
			assert.Equal(t, mockErr, err)
		}
	})

	t.Run("get auth token error", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		mockErr := errors.New("get auth token error")
		extractor.EXPECT().GetDataPlaneApiGateway(gomock.Any(), gomock.Any()).Return("", nil)
		extractor.EXPECT().GetDataPlaneAuthToken(gomock.Any(), gomock.Any()).Return("", mockErr)
		svc := &DataPlaneService{cloudProviderExtractor: extractor}
		err := svc.fillAuthTokenApiGateway2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.Error(t, err) {
			assert.Equal(t, mockErr, err)
		}
	})

	t.Run("fill auth token and api gateway successfully", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		expectedApiGateway := "https://abc.com"
		expectedAuthToken := "token"
		extractor.EXPECT().GetDataPlaneApiGateway(gomock.Any(), mockAppBox).Return(expectedApiGateway, nil)
		extractor.EXPECT().GetDataPlaneAuthToken(mockDataPlane.AccountId, mockAppBox).Return(expectedAuthToken, nil)
		svc := &DataPlaneService{cloudProviderExtractor: extractor}
		err := svc.fillAuthTokenApiGateway2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.NoError(t, err) {
			assert.Equal(t, expectedApiGateway, mockDataPlane.ApiGateway)
			assert.Equal(t, expectedAuthToken, mockDataPlane.AuthToken)
		}
	})
}

func Test_fillAppBox2DataPlane(t *testing.T) {
	mockDataPlane := &dmpbv1.DataPlane{
		Id:         "mock-dataplane",
		AccountId:  "abc",
		Kubernetes: &dmpbv1.DataPlane_Kubernetes{},
	}

	mockAppBox := &v1alpha1.APPBox{}
	mockAppBox.Name = "mock-appbox"
	mockAppBox.Namespace = "mock-namespace"
	expectedK8sVersion := "v1.19.0"
	mockAppBox.Spec.KubernetesConfig.Version = expectedK8sVersion
	expectedK8sName := "mock-kubernetes"
	mockAppBox.Status.KubernetesClusterName = expectedK8sName

	t.Run("build kubeconfig error", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		appBoxRW := mockrepository.NewMockAPPBoxReaderWriter(ctrl)
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		mockErr := errors.New("not found")
		extractor.EXPECT().BuildKubeconfig(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, mockErr)
		svc := &DataPlaneService{appBoxRW: appBoxRW, cloudProviderExtractor: extractor}
		err := svc.fillAppBox2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.Error(t, err) {
			assert.Equal(t, mockErr, err)
		}
	})

	t.Run("k8s cluster is not active", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		appBoxRW := mockrepository.NewMockAPPBoxReaderWriter(ctrl)
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		extractor.EXPECT().BuildKubeconfig(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errK8sNotActive)
		svc := &DataPlaneService{appBoxRW: appBoxRW, cloudProviderExtractor: extractor}
		err := svc.fillAppBox2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.NoError(t, err) {
			assert.Equal(t, expectedK8sName, mockDataPlane.Kubernetes.Name)
			assert.Equal(t, expectedK8sVersion, mockDataPlane.Kubernetes.Version)
		}
	})

	expectedKubeconfig := "mock-kubeconfig"

	t.Run("appbox is not active", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		appBoxRW := mockrepository.NewMockAPPBoxReaderWriter(ctrl)
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		extractor.EXPECT().BuildKubeconfig(gomock.Any(), mockDataPlane.AccountId, mockAppBox).
			Return([]byte(expectedKubeconfig), nil)
		svc := &DataPlaneService{appBoxRW: appBoxRW, cloudProviderExtractor: extractor}
		err := svc.fillAppBox2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.NoError(t, err) {
			assert.Equal(t, expectedK8sName, mockDataPlane.Kubernetes.Name)
			assert.Equal(t, expectedK8sVersion, mockDataPlane.Kubernetes.Version)
			assert.Equal(t, expectedKubeconfig, mockDataPlane.Kubernetes.Kubeconfig)
		}
	})

	expectedNodeRole := "mock-node-role"
	expectedNodeClientID := "mock-node-client-id"
	expectedApiGateway := "https://abc.com"
	expectedAuthToken := "token"
	mockAppBox.Status.Phase = v1alpha1.APPBoxPhaseActive

	t.Run("appbox is active", func(t *testing.T) {
		ctrl := gomock.NewController(t)
		defer ctrl.Finish()
		appBoxRW := mockrepository.NewMockAPPBoxReaderWriter(ctrl)
		extractor := dataplane_mock.NewMockCloudProviderExtractor(ctrl)
		extractor.EXPECT().BuildKubeconfig(gomock.Any(), mockDataPlane.AccountId, mockAppBox).
			Return([]byte(expectedKubeconfig), nil)
		extractor.EXPECT().ParseNodeRole(mockAppBox).Return(expectedNodeRole)
		extractor.EXPECT().ParseNodeClientID(mockAppBox).Return(expectedNodeClientID)
		extractor.EXPECT().GetDataPlaneApiGateway(gomock.Any(), mockAppBox).Return(expectedApiGateway, nil)
		extractor.EXPECT().GetDataPlaneAuthToken(mockDataPlane.AccountId, mockAppBox).Return(expectedAuthToken, nil)

		svc := &DataPlaneService{appBoxRW: appBoxRW, cloudProviderExtractor: extractor}
		err := svc.fillAppBox2DataPlane(context.Background(), mockAppBox, mockDataPlane)
		if assert.NoError(t, err) {
			assert.Equal(t, expectedK8sName, mockDataPlane.Kubernetes.Name)
			assert.Equal(t, expectedK8sVersion, mockDataPlane.Kubernetes.Version)
			assert.Equal(t, expectedKubeconfig, mockDataPlane.Kubernetes.Kubeconfig)
			assert.Equal(t, expectedNodeRole, mockDataPlane.NodeRole)
			assert.Equal(t, expectedNodeClientID, mockDataPlane.NodeClientId)
			assert.Equal(t, expectedApiGateway, mockDataPlane.ApiGateway)
			assert.Equal(t, expectedAuthToken, mockDataPlane.AuthToken)
		}
	})
}

func checkNodePoolsValid(nodePools []*dmpbv1.NodePool) bool {
	if len(nodePools) == 0 {
		return true
	}

	k8sVersion := nodePools[0].K8SVersion
	if k8sVersion == "" {
		return false
	}

	nodePoolsSet := make(map[string]struct{}, len(nodePools))
	for _, nodePool := range nodePools {
		nodeKey := fmt.Sprintf("%s-%s", nodePool.NodeType.String(), nodePool.InstanceType)
		if _, ok := nodePoolsSet[nodeKey]; ok {
			return false
		}
		if nodePool.K8SVersion != k8sVersion {
			return false
		}
		nodePoolsSet[nodeKey] = struct{}{}
	}
	return true
}