package common

import (
	"context"
	"fmt"
	"reflect"
	"testing"
	"unsafe"

	gkev1beta1 "github.com/crossplane-contrib/provider-gcp/apis/container/v1beta1"
	opv1alpha1 "github.com/pingcap/tidb-operator/pkg/apis/pingcap/v1alpha1"
	"go.uber.org/mock/gomock"
	gcpcontainer "google.golang.org/api/container/v1"
	"helm.sh/helm/v3/pkg/action"
	appsv1 "k8s.io/api/apps/v1"
	corev1 "k8s.io/api/core/v1"
	"k8s.io/apimachinery/pkg/api/resource"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	k8slabels "k8s.io/apimachinery/pkg/labels"
	"k8s.io/client-go/rest"
	"k8s.io/utils/ptr"
	"sigs.k8s.io/controller-runtime/pkg/client"
	"sigs.k8s.io/controller-runtime/pkg/client/fake"
	"sigs.k8s.io/yaml"

	"github.com/google/go-cmp/cmp"
	"github.com/google/go-cmp/cmp/cmpopts"
	"github.com/samber/lo"
	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"

	"github.com/tidbcloud/eks-provider/controllers/cluster/task"
	"github.com/tidbcloud/eks-provider/controllers/utils"
	"github.com/tidbcloud/eks-provider/utils/tidb-operator/tidbcluster"
	cloudv1 "github.com/tidbcloud/infra-api/api/v1"
	"github.com/tidbcloud/infra-provider/pkg/cloud/gcp"
	"github.com/tidbcloud/infra-provider/pkg/consts"
	shared "github.com/tidbcloud/infra-provider/pkg/controllers/common"
	"github.com/tidbcloud/infra-provider/pkg/dataplane"
	"github.com/tidbcloud/infra-provider/pkg/gkeheterogeneous"
	"github.com/tidbcloud/infra-provider/pkg/gkemindomains"
	"github.com/tidbcloud/infra-provider/pkg/k8s"
	"github.com/tidbcloud/infra-provider/pkg/namer"
	nodepoolutil "github.com/tidbcloud/infra-provider/pkg/nodepool"
	scheme2 "github.com/tidbcloud/infra-provider/pkg/scheme"
)

func updateTestGKEComputeClassBinding(t *testing.T, cluster *cloudv1.Cluster, key, _ string, instanceTypes, nodePoolNames []string) string {
	t.Helper()
	require.NotEmpty(t, instanceTypes)
	if len(instanceTypes) == 1 {
		extraInstanceType := testGKEExtraInstanceType(instanceTypes[0])
		instanceTypes = append(append([]string{}, instanceTypes...), extraInstanceType)
		if len(nodePoolNames) == 1 {
			nodePoolNames = append(append([]string{}, nodePoolNames...), gkeheterogeneous.NodePoolName(cluster, extraInstanceType))
		}
	}
	require.Len(t, nodePoolNames, len(instanceTypes))
	annotation := cloudv1.GKEComputeClassBindingAnnotation{
		NodePools: make([]cloudv1.GKEComputeClassNodePoolBinding, 0, len(instanceTypes)),
	}
	for i, instanceType := range instanceTypes {
		role := cloudv1.GKEComputeClassNodePoolRoleExtra
		if i == 0 {
			role = cloudv1.GKEComputeClassNodePoolRoleBase
		}
		annotation.NodePools = append(annotation.NodePools, cloudv1.GKEComputeClassNodePoolBinding{
			Role:         role,
			InstanceType: instanceType,
			NodePoolName: nodePoolNames[i],
		})
	}
	computeClassName, err := cloudv1.GKEComputeClassNameForBindings(annotation.NodePools)
	require.NoError(t, err)
	annotation.ComputeClassName = computeClassName
	value, err := cloudv1.EncodeGKEComputeClassBindingAnnotation(annotation, key, instanceTypes[0], nodePoolNames[0])
	require.NoError(t, err)
	cluster.UpdateAnnotation(key, value)
	return computeClassName
}

func testGKEExtraInstanceType(instanceType string) string {
	if instanceType == "n2-standard-64" {
		return "n2-standard-96"
	}
	return "n2-standard-64"
}

func testGKEComputeClassNodePoolNames(cluster *cloudv1.Cluster, instanceTypes ...string) []string {
	out := make([]string, 0, len(instanceTypes))
	for _, instanceType := range instanceTypes {
		out = append(out, gkeheterogeneous.NodePoolName(cluster, instanceType))
	}
	return out
}

type testEnvProvider struct {
	client client.Client
}

type recordingClusterListClient struct {
	client.Client
	clusterListSelectors []string
}

func (c *recordingClusterListClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
	if _, ok := list.(*cloudv1.ClusterList); ok {
		listOptions := &client.ListOptions{}
		listOptions.ApplyOptions(opts)
		if listOptions.LabelSelector == nil {
			c.clusterListSelectors = append(c.clusterListSelectors, "")
		} else {
			c.clusterListSelectors = append(c.clusterListSelectors, listOptions.LabelSelector.String())
		}
	}
	return c.Client.List(ctx, list, opts...)
}

type nodePoolListFailingClient struct {
	client.Client
}

func (c *nodePoolListFailingClient) List(ctx context.Context, list client.ObjectList, opts ...client.ListOption) error {
	if _, ok := list.(*gkev1beta1.NodePoolList); ok {
		return fmt.Errorf("nodepool list failed")
	}
	return c.Client.List(ctx, list, opts...)
}

func testGCPClusterRef() string {
	return "projects/test-project/locations/us-central1/clusters/gke-test"
}

func setTestGKEAPPBoxLabels(cluster *cloudv1.Cluster) {
	setTestGKEAPPBoxLabelsWithProject(cluster, "project-1")
}

// Tests that expect GKE shared-nodepool peer scans must use APPBox labels.
// Missing-label cases construct the Cluster without this helper on purpose.
func setTestGKEAPPBoxLabelsWithProject(cluster *cloudv1.Cluster, projectID string) {
	if cluster.Labels == nil {
		cluster.Labels = map[string]string{}
	}
	if cluster.Spec.Kubernetes.Name == "" {
		cluster.Spec.Kubernetes.Name = "gke-test"
	}
	cluster.Labels[cloudv1.RegionLabel] = "us-central1"
	cluster.Labels[cloudv1.TenantTagKey] = "tenant-1"
	cluster.Labels[cloudv1.ProjectTagKey] = projectID
}

func (t *testEnvProvider) GetClient() client.Client             { return t.client }
func (t *testEnvProvider) GetHelmConfig() *action.Configuration { return nil }
func (t *testEnvProvider) GetRestConfig() *rest.Config          { return nil }
func (t *testEnvProvider) GetRegion() string                    { return "us-central1" }
func (t *testEnvProvider) GetClusterName() string               { return "gke-test" }
func (t *testEnvProvider) GetClusterEndpoint() string           { return "" }
func (t *testEnvProvider) GetClusterCA() []byte                 { return nil }
func (t *testEnvProvider) GetAccount() string                   { return "test-project" }
func (t *testEnvProvider) GetNetwork() string                   { return "" }
func (t *testEnvProvider) AWS() dataplane.AWSInterface          { return nil }
func (t *testEnvProvider) GCP() dataplane.GCPInterface          { return nil }

func setContextZones(ctx *task.Context, zones []string) {
	field := reflect.ValueOf(ctx).Elem().FieldByName("zones")
	reflect.NewAt(field.Type(), unsafe.Pointer(field.UnsafeAddr())).Elem().Set(reflect.ValueOf(zones))
}

func testNodePoolForCluster(cluster *cloudv1.Cluster) *NodePool {
	return &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Cluster: cluster,
			},
		},
	}
}

func TestNodePool_resolveDisableGKEHeterogeneousOPNodeSelector(t *testing.T) {
	tests := []struct {
		name           string
		provider       string
		workflowInputs map[string]string
		wantOK         bool
	}{
		{
			name:     "flag absent skips fallback",
			provider: cloudv1.ProviderTypeGKE,
			workflowInputs: map[string]string{
				consts.WfInputKeyNamespace:     "ns1",
				consts.WfInputKeyClusterID:     "cluster1",
				consts.WfInputKeyGroupID:       "",
				consts.WfInputKeyComponentType: string(cloudv1.ComponentTiKV),
			},
		},
		{
			name:     "component group mismatch skips fallback",
			provider: cloudv1.ProviderTypeGKE,
			workflowInputs: map[string]string{
				consts.WfInputKeyNamespace:                           "ns1",
				consts.WfInputKeyClusterID:                           "cluster1",
				consts.WfInputKeyGroupID:                             "other",
				consts.WfInputKeyComponentType:                       string(cloudv1.ComponentTiKV),
				consts.WfInputKeyScaleClusterDisableGKEHeterogeneous: "true",
			},
		},
		{
			name:     "matching disable gke heterogeneous workflow builds ordinary selector",
			provider: cloudv1.ProviderTypeGKE,
			workflowInputs: map[string]string{
				consts.WfInputKeyNamespace:                           "ns1",
				consts.WfInputKeyClusterID:                           "cluster1",
				consts.WfInputKeyGroupID:                             "",
				consts.WfInputKeyComponentType:                       string(cloudv1.ComponentTiKV),
				consts.WfInputKeyScaleClusterDisableGKEHeterogeneous: "true",
			},
			wantOK: true,
		},
		{
			name:     "non gke cluster skips fallback",
			provider: cloudv1.ProviderTypeEKS,
			workflowInputs: map[string]string{
				consts.WfInputKeyNamespace:                           "ns1",
				consts.WfInputKeyClusterID:                           "cluster1",
				consts.WfInputKeyGroupID:                             "",
				consts.WfInputKeyComponentType:                       string(cloudv1.ComponentTiKV),
				consts.WfInputKeyScaleClusterDisableGKEHeterogeneous: "true",
			},
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "cluster1",
					Namespace: "ns1",
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: tt.provider,
					},
					Annotations: map[string]string{},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiKV: cloudv1.TiKV{
							Replicas:     3,
							InstanceType: "n2-standard-64",
						},
					},
				},
			}
			workflow := &cloudv1.Workflow{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "scale-tikv",
					Namespace: "ns1",
				},
				Spec: cloudv1.WorkflowSpec{
					Type:   consts.WorkflowTypeScaleCluster,
					Inputs: tt.workflowInputs,
				},
			}
			if tt.wantOK {
				originalSpec, err := yaml.Marshal(&cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				})
				require.NoError(t, err)
				workflow.Spec.Inputs[consts.WfInputKeyScaleClusterOriginalComponentSpec] = string(originalSpec)
			}
			rawWorkflow, err := yaml.Marshal(workflow)
			require.NoError(t, err)
			cluster.UpdateAnnotation(consts.AnnoKeyScaleClusterWorkflow, string(rawWorkflow))

			np := testNodePoolForCluster(cluster)
			np.Context.CommonK8sClusterStatus = &cloudv1.CommonK8sClusterStatus{NodeVersion: "1.29"}
			comp, err := cluster.GetComponentSpecInGroup("", cloudv1.ComponentTiKV)
			require.NoError(t, err)

			currentNodeSelector := map[string]string{
				consts.GKEComputeClassLabelKey: "compute-class-old",
				consts.K8sVersionLabelKey:      "1.28",
				consts.LabelKeyNodeGroupType:   consts.LabelValNodeGroupTypeDataPlane,
				consts.ZoneLabelKey:            "us-central1-a",
				"custom-selector":              "preserved",
			}
			got, originalIt, ok, err := np.resolveDisableGKEHeterogeneousOPNodeSelector(comp, currentNodeSelector)
			require.NoError(t, err)
			assert.Equal(t, tt.wantOK, ok)
			if !tt.wantOK {
				assert.Nil(t, got)
				assert.Empty(t, originalIt)
				return
			}
			assert.Equal(t, "n2-standard-8", originalIt)
			assert.Equal(t, "n2-standard-64", got[consts.InstanceTypeLabelKey])
			assert.Equal(t, "1.28", got[consts.K8sVersionLabelKey])
			assert.Equal(t, consts.LabelValNodeGroupTypeDataPlane, got[consts.LabelKeyNodeGroupType])
			assert.Equal(t, "us-central1-a", got[consts.ZoneLabelKey])
			assert.Equal(t, "preserved", got["custom-selector"])
			assert.NotContains(t, got, consts.GKEComputeClassLabelKey)
		})
	}
}

func TestNodePool_createOverProvisioningPodsSkipsDisableGKEHeterogeneousFallbackWhenTargetStillOriginal(t *testing.T) {
	const (
		originalIt = "n2-standard-8"
		desiredIt  = "n2-standard-64"
	)

	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster1",
			Namespace: "ns1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
			Annotations: map[string]string{},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: originalIt,
				},
			},
		},
	}
	originalSpec, err := yaml.Marshal(&cloudv1.TiKV{
		Replicas:     3,
		InstanceType: originalIt,
	})
	require.NoError(t, err)
	desiredSpec, err := yaml.Marshal(&cloudv1.TiKV{
		Replicas:     3,
		InstanceType: desiredIt,
	})
	require.NoError(t, err)
	workflow := &cloudv1.Workflow{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "scale-tikv",
			Namespace: cluster.Namespace,
		},
		Spec: cloudv1.WorkflowSpec{
			Type: consts.WorkflowTypeScaleCluster,
			Inputs: map[string]string{
				consts.WfInputKeyNamespace:                             cluster.Namespace,
				consts.WfInputKeyClusterID:                             cluster.Name,
				consts.WfInputKeyGroupID:                               "",
				consts.WfInputKeyComponentType:                         string(cloudv1.ComponentTiKV),
				consts.WfInputKeyScaleClusterOriginalComponentSpec:     string(originalSpec),
				consts.WfInputKeyScaleClusterDesiredComponentSpec:      string(desiredSpec),
				consts.WfInputKeyScaleClusterDisableGKEHeterogeneous:   "true",
				consts.WfInputKeyScaleClusterCurrentConfigRevisionName: "current-config-revision",
				consts.WfInputKeyScaleClusterDesiredConfigRevisionName: "desired-config-revision",
			},
		},
	}
	rawWorkflow, err := yaml.Marshal(workflow)
	require.NoError(t, err)
	cluster.UpdateAnnotation(consts.AnnoKeyScaleClusterWorkflow, string(rawWorkflow))

	n := namer.NewNamer().
		WithClusterName(cluster.Name).
		WithTiDBNamespace(cluster.PrefixedId())
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "tikv-0",
			Namespace: n.GetTiDBNamespace(),
			Labels: map[string]string{
				consts.InstanceLabelKey:  n.GetTiDBClusterName(),
				consts.ComponentLabelKey: string(cloudv1.ComponentTiKV),
			},
		},
		Spec: corev1.PodSpec{
			NodeSelector: map[string]string{
				consts.GKEComputeClassLabelKey: "compute-class-old",
				consts.K8sVersionLabelKey:      "1.29",
			},
		},
	}
	remoteCli := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(pod).Build()
	tc := &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      n.GetTiDBClusterName(),
			Namespace: n.GetTiDBNamespace(),
		},
		Spec: opv1alpha1.TidbClusterSpec{
			TiKV: &opv1alpha1.TiKVSpec{Replicas: 3},
		},
	}
	ctx := &task.Context{
		Cluster:  cluster,
		Namer:    n,
		TCGetter: tidbcluster.NewV1Getter(tc),
		Provider: fakeCloudConfigProvider{
			instanceTypes: map[string]cloudv1.InstanceType{
				originalIt: {Name: originalIt},
			},
		},
	}
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}
	var comp cloudv1.ClusterComp
	for _, candidate := range cluster.GetAllComponentsV2() {
		if candidate.GetComponentType() == cloudv1.ComponentTiKV {
			comp = candidate
			break
		}
	}
	require.NotNil(t, comp)

	err = np.createOverProvisioningPods(context.Background(), remoteCli, comp, 3)
	require.NoError(t, err)

	deployments := &appsv1.DeploymentList{}
	require.NoError(t, remoteCli.List(context.Background(), deployments, client.InNamespace(n.GetTiDBNamespace())))
	assert.Empty(t, deployments.Items)
	assert.Equal(t, cloudv1.InstanceTypeScalePhaseNone, comp.GetInstanceTypeScalePhase())
}

func TestNodePool_createOverProvisioningPodsAddsArchTolerationForGKEARM64Target(t *testing.T) {
	const (
		currentIt = "n2-standard-8"
		targetIt  = "c4a-standard-8"
	)

	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster1",
			Namespace: "ns1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: targetIt,
				},
			},
		},
	}
	n := namer.NewNamer().
		WithClusterName(cluster.Name).
		WithTiDBNamespace(cluster.PrefixedId())
	pod := &corev1.Pod{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "tikv-0",
			Namespace: n.GetTiDBNamespace(),
			Labels: map[string]string{
				consts.InstanceLabelKey:  n.GetTiDBClusterName(),
				consts.ComponentLabelKey: string(cloudv1.ComponentTiKV),
			},
		},
		Spec: corev1.PodSpec{
			NodeSelector: map[string]string{
				consts.InstanceTypeLabelKey: currentIt,
				consts.K8sVersionLabelKey:   "1.29",
			},
		},
	}
	remoteCli := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(pod).Build()
	tc := &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      n.GetTiDBClusterName(),
			Namespace: n.GetTiDBNamespace(),
		},
		Spec: opv1alpha1.TidbClusterSpec{
			TiKV: &opv1alpha1.TiKVSpec{Replicas: 3},
		},
	}
	ctx := &task.Context{
		Cluster:  cluster,
		Namer:    n,
		TCGetter: tidbcluster.NewV1Getter(tc),
		Provider: fakeCloudConfigProvider{
			instanceTypes: map[string]cloudv1.InstanceType{
				currentIt: {
					Name:   currentIt,
					CPU:    resource.MustParse("8"),
					Memory: resource.MustParse("32Gi"),
				},
				targetIt: {
					Name:   targetIt,
					Arch:   consts.ARM64ArchName,
					CPU:    resource.MustParse("8"),
					Memory: resource.MustParse("32Gi"),
				},
			},
		},
	}
	setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}
	var comp cloudv1.ClusterComp
	for _, candidate := range cluster.GetAllComponentsV2() {
		if candidate.GetComponentType() == cloudv1.ComponentTiKV {
			comp = candidate
			break
		}
	}
	require.NotNil(t, comp)

	err := np.createOverProvisioningPods(context.Background(), remoteCli, comp, 3)
	require.NoError(t, err)

	deployments := &appsv1.DeploymentList{}
	require.NoError(t, remoteCli.List(context.Background(), deployments, client.InNamespace(n.GetTiDBNamespace())))
	require.Len(t, deployments.Items, 1)
	assert.Contains(t, deployments.Items[0].Spec.Template.Spec.Tolerations, *utils.TolerationWithArch(consts.ARM64ArchName))
	assert.Equal(t, targetIt, deployments.Items[0].Spec.Template.Spec.NodeSelector[consts.InstanceTypeLabelKey])
	assert.Equal(t, cloudv1.InstanceTypeScalePhaseCreatingNodes, comp.GetInstanceTypeScalePhase())
}

func TestNodePool_createOverProvisioningPodsMarksMinDomainsRolloutWhenApplicable(t *testing.T) {
	const (
		currentIt = "n2-standard-4"
		targetIt  = "n2-standard-8"
	)
	tests := []struct {
		name       string
		applicable bool
	}{
		{name: "applicable", applicable: true},
		{name: "inapplicable", applicable: false},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := testGKEV1MinDomainsCluster()
			if !tt.applicable {
				cluster.Spec.Kubernetes.Cloud.GCP.NodeZones = []string{"us-central1-a", "us-central1-b"}
			}
			cluster.Status.Phase = cloudv1.ClusterPhaseRunning
			cluster.Spec.Cluster.TiDB = cloudv1.TiDB{Replicas: 1, InstanceType: targetIt}
			n := namer.NewNamer().WithClusterName(cluster.Name).WithTiDBNamespace(cluster.PrefixedId())
			pod := &corev1.Pod{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "tidb-0",
					Namespace: n.GetTiDBNamespace(),
					Labels: map[string]string{
						consts.InstanceLabelKey:  n.GetTiDBClusterName(),
						consts.ComponentLabelKey: string(cloudv1.ComponentTiDB),
					},
				},
				Spec: corev1.PodSpec{NodeSelector: map[string]string{
					consts.InstanceTypeLabelKey: currentIt,
					consts.K8sVersionLabelKey:   "1.29",
				}},
			}
			remoteCli := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(pod).Build()
			tc := &opv1alpha1.TidbCluster{
				ObjectMeta: metav1.ObjectMeta{Name: n.GetTiDBClusterName(), Namespace: n.GetTiDBNamespace()},
				Spec: opv1alpha1.TidbClusterSpec{TiDB: &opv1alpha1.TiDBSpec{
					Replicas: 1,
					ComponentSpec: opv1alpha1.ComponentSpec{NodeSelector: map[string]string{
						consts.InstanceTypeLabelKey: currentIt,
					}},
				}},
			}
			ctx := &task.Context{
				Cluster:     cluster,
				Namer:       n,
				Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
				EnvProvider: &testEnvProvider{client: remoteCli},
				TCGetter:    tidbcluster.NewV1Getter(tc),
				Provider: fakeCloudConfigProvider{instanceTypes: map[string]cloudv1.InstanceType{
					currentIt: {Name: currentIt, CPU: resource.MustParse("4"), Memory: resource.MustParse("16Gi")},
					targetIt:  {Name: targetIt, CPU: resource.MustParse("8"), Memory: resource.MustParse("32Gi")},
				}},
			}
			setContextZones(ctx, cluster.Spec.Kubernetes.Cloud.GCP.NodeZones)
			np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}
			var comp cloudv1.ClusterComp
			for _, candidate := range cluster.GetAllComponentsV2() {
				if candidate.GetComponentType() == cloudv1.ComponentTiDB {
					comp = candidate
					break
				}
			}
			require.NotNil(t, comp)

			err := np.createOverProvisioningPods(context.Background(), remoteCli, comp, 3)
			require.NoError(t, err)
			deployments := &appsv1.DeploymentList{}
			require.NoError(t, remoteCli.List(context.Background(), deployments, client.InNamespace(n.GetTiDBNamespace())))
			require.Len(t, deployments.Items, 1)
			require.Equal(t, tt.applicable,
				deployments.Items[0].Annotations[annoGKEV1MinDomainsInstanceTypeRollout] == consts.AnnoValTrue)
			require.Equal(t, cloudv1.InstanceTypeScalePhaseCreatingNodes, comp.GetInstanceTypeScalePhase())
		})
	}
}

func TestMergeDesiredNodePoolDoesNotMergeComputeClassBeforeSpecCheck(t *testing.T) {
	newNodePool := func(computeClass string, locations []string) *gkev1beta1.NodePool {
		nodePool := &gkev1beta1.NodePool{
			ObjectMeta: metav1.ObjectMeta{Name: "shared-pool"},
			Spec: gkev1beta1.NodePoolSpec{
				ForProvider: gkev1beta1.NodePoolParameters{
					Config: &gkev1beta1.NodeConfig{
						MachineType: ptr.To("n2-standard-8"),
					},
					Locations: locations,
				},
			},
		}
		nodepoolutil.SetGKEComputeClassForNodePool(nodePool, computeClass)
		return nodePool
	}

	tests := []struct {
		name                string
		currentComputeClass string
		desiredComputeClass string
		desiredLocations    []string
		wantErr             bool
		wantComputeClass    string
	}{
		{
			name:             "accepts matching spec without compute class",
			desiredLocations: []string{"us-central1-a"},
		},
		{
			name:                "rejects stale current compute class as spec drift",
			currentComputeClass: "shared-priority",
			desiredLocations:    []string{"us-central1-a"},
			wantErr:             true,
			wantComputeClass:    "shared-priority",
		},
		{
			name:                "rejects desired compute class as spec drift",
			desiredComputeClass: "shared-priority",
			desiredLocations:    []string{"us-central1-a"},
			wantErr:             true,
		},
		{
			name:             "checks spec mismatch",
			desiredLocations: []string{"us-central1-b"},
			wantErr:          true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			current := newNodePool(tt.currentComputeClass, []string{"us-central1-a"})
			desired := newNodePool(tt.desiredComputeClass, tt.desiredLocations)

			err := (&NodePool{}).mergeDesiredNodePool(current, desired, desiredNodePoolRequest{
				instanceType:  "n2-standard-8",
				componentType: cloudv1.ComponentTiDB,
			})

			if tt.wantErr {
				assert.ErrorContains(t, err, "inconsistent nodepool spec")
			} else {
				assert.NoError(t, err)
			}
			assert.Equal(t, tt.wantComputeClass, nodepoolutil.GKEComputeClassNameForNodePool(current))
		})
	}
}

func TestNodePool_desiredNodePoolsIgnoresLocalComputeClassIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority-v2", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "custom-extra-pool"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			existingTestNodePool(baseName, "n2-standard-8"),
			existingTestNodePool("custom-extra-pool", "n2-standard-64"),
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)

	ordinaryName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	if assert.Contains(t, nodePoolsByName, ordinaryName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[ordinaryName]))
		assertNoComputeClassTaint(t, nodePoolsByName[ordinaryName])
	}
	assert.NotContains(t, nodePoolsByName, baseName)
	assert.NotContains(t, nodePoolsByName, "custom-extra-pool")
	assert.NotContains(t, nodePoolsByName, nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-64", "1.35"))
}

func TestNodePool_desiredNodePoolsIgnoresMissingHeterogeneousNodePools(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority-v2", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(cluster, "n2-standard-8"), "custom-extra-pool"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)

	ordinaryName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	if assert.Contains(t, nodePoolsByName, ordinaryName) {
		assert.Equal(t, "n2-standard-8", *nodePoolsByName[ordinaryName].Spec.ForProvider.Config.MachineType)
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[ordinaryName]))
	}
	assert.NotContains(t, nodePoolsByName, gkeheterogeneous.NodePoolName(cluster, "n2-standard-8"))
	assert.NotContains(t, nodePoolsByName, "custom-extra-pool")
}

func TestNodePool_desiredNodePoolsAllowsOrdinaryGKEClusterMissingRawScopeLabels(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)
	if assert.Len(t, nps, 1) {
		assert.NotContains(t, nps[0].Labels, cloudv1.RegionLabel)
		assert.NotContains(t, nps[0].Labels, cloudv1.TenantTagKey)
		assert.NotContains(t, nps[0].Labels, cloudv1.ProjectTagKey)
	}
}

func TestNodePool_desiredNodePoolsOrdinaryGKE3LayerKeepsMainNodePoolLabels(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster1",
			Namespace: "ns1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}

	nps, err := np.desiredNodePools(context.Background())
	require.NoError(t, err)
	require.Len(t, nps, 1)

	assert.Equal(t, "project-1", nps[0].Labels[cloudv1.TagLabelPrefix+shared.Project])
	assert.Equal(t, "tenant-1", nps[0].Labels[cloudv1.TagLabelPrefix+shared.Tenant])
	assert.Equal(t, consts.LabelValTrue, nps[0].Spec.ForProvider.Config.Labels[consts.LabelKeyFor3LayerArch])
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nps[0]))
}

func TestNodePool_desiredNodePoolsAddsArchTaintFromInstanceCatalog(t *testing.T) {
	tests := []struct {
		name          string
		instanceType  string
		arch          string
		threeLayer    bool
		wantArchTaint bool
	}{
		{
			name:          "ARM64 instance with 3-layer taint",
			instanceType:  "c4a-standard-4",
			arch:          consts.ARM64ArchName,
			threeLayer:    true,
			wantArchTaint: true,
		},
		{
			name:          "ARM64 instance without 3-layer taint",
			instanceType:  "c4a-standard-4",
			arch:          consts.ARM64ArchName,
			wantArchTaint: true,
		},
		{
			name:         "AMD64 instance with 3-layer taint",
			instanceType: "n2-standard-4",
			arch:         consts.AMD64ArchName,
			threeLayer:   true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name: "cluster1",
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
					},
				},
				Spec: cloudv1.ClusterSpec{
					Cluster: cloudv1.ClusterTopology{
						TiKV: cloudv1.TiKV{
							Replicas:     3,
							InstanceType: tt.instanceType,
						},
					},
				},
			}
			setTestGKEAPPBoxLabels(cluster)
			if tt.threeLayer {
				cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
			}

			ctx := &task.Context{
				Cluster: cluster,
				Namer: namer.NewNamer().
					WithClusterName(cluster.Name).
					WithTiDBNamespace(cluster.PrefixedId()),
				Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
				EnvProvider: &testEnvProvider{},
				Provider: fakeCloudConfigProvider{
					instanceTypes: map[string]cloudv1.InstanceType{
						tt.instanceType: {
							Name: tt.instanceType,
							Arch: tt.arch,
						},
					},
				},
				CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
					NodeVersion: "1.35",
				},
			}
			setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})

			nps, err := (&NodePool{BaseTask: &task.BaseTask{Context: ctx}}).desiredNodePools(context.Background())
			require.NoError(t, err)
			require.Len(t, nps, 1)

			wantTaints := []*gkev1beta1.NodeTaint{
				{
					Key:    "dedicated",
					Value:  "tidb-" + shared.ComponentShared,
					Effect: "NO_SCHEDULE",
				},
			}
			if tt.threeLayer {
				wantTaints = append(wantTaints, taintToNodeTaint(utils.TaintWith3LayerArch()))
			}
			if tt.wantArchTaint {
				wantTaints = append(wantTaints, taintToNodeTaint(utils.TaintWithArch(consts.ARM64ArchName)))
			}
			assert.Equal(t, wantTaints, nps[0].Spec.ForProvider.Config.Taints)
		})
	}
}

func assertNoComputeClassTaint(t *testing.T, nodePool *gkev1beta1.NodePool) {
	t.Helper()
	for _, taint := range nodePool.Spec.ForProvider.Config.Taints {
		assert.NotEqual(t, consts.GKEComputeClassLabelKey, taint.Key)
	}
}

func testGKENodePoolsByName(nodePools []*gkev1beta1.NodePool) map[string]*gkev1beta1.NodePool {
	out := make(map[string]*gkev1beta1.NodePool, len(nodePools))
	for _, nodePool := range nodePools {
		out[nodePool.Name] = nodePool
	}
	return out
}

func existingTestNodePool(name, instanceType string) *gkev1beta1.NodePool {
	np := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{Name: name},
	}
	if instanceType == "" {
		return np
	}
	np.Spec.ForProvider.Cluster = testGCPClusterRef()
	np.Spec.ForProvider.Config = &gkev1beta1.NodeConfig{
		MachineType: ptr.To(instanceType),
	}
	return np
}

func syncedAPIManagedGKEHeterogeneousTestNodePool(cluster *cloudv1.Cluster, name, instanceType string) *gkev1beta1.NodePool {
	np := existingTestNodePool(name, instanceType)
	np.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(np)
	np.Annotations = map[string]string{
		annoKeyResourceLabels: "tidb_cluster:" + cluster.Name,
	}
	np.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	np.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(name)
	return np
}

func sharedTestNodePoolLabels(cluster *cloudv1.Cluster) map[string]string {
	labels := map[string]string{
		namer.ComponentSelectorKey: shared.ComponentShared,
		namer.ClusterSelectorKey:   shared.ClustersShared,
	}
	scopeLabels, _ := nodepoolutil.GKEHeterogeneousNodePoolScopeLabels(cluster)
	for key, value := range scopeLabels {
		labels[key] = value
	}
	return labels
}

func existingSharedComputeClassTestNodePool(cluster *cloudv1.Cluster, name, instanceType, computeClass string) *gkev1beta1.NodePool {
	annotations := map[string]string{
		annoKeyResourceLabels: "tidb_cluster:" + cluster.Name,
	}
	labels := map[string]string{}
	if computeClass != "" {
		annotations[consts.AnnoKeyGKENodePoolComputeClass] = computeClass
		labels[consts.GKEComputeClassLabelKey] = computeClass
	}
	return &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:        name,
			Labels:      sharedTestNodePoolLabels(cluster),
			Annotations: annotations,
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: testGCPClusterRef(),
				Config: &gkev1beta1.NodeConfig{
					MachineType: ptr.To(instanceType),
					Labels:      labels,
				},
			},
		},
		Status: gkev1beta1.NodePoolStatus{
			AtProvider: gkev1beta1.NodePoolObservation{
				Status: gkev1beta1.NodePoolStateRunning,
			},
		},
	}
}

func testGKENodePoolResourceName(name string) string {
	return testGCPClusterRef() + "/nodePools/" + name
}

func testGKENodePoolSelfLink(name string) string {
	return "https://container.googleapis.com/v1/" + testGKENodePoolResourceName(name)
}

func expectGKENodePoolComputeClassNodeMetadataSync(t *testing.T, gcpCli *gcp.MockClient, nodePoolName, computeClass string) {
	t.Helper()
	nodePoolResourceName := testGKENodePoolResourceName(nodePoolName)
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), nodePoolResourceName).
		Return(map[string]string{
			"existing":                     "keep",
			consts.GKEComputeClassLabelKey: computeClass,
		}, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), nodePoolResourceName).
		Return([]*gcpcontainer.NodeTaint{
			testGKEComputeClassNodeTaint(computeClass),
		}, nil)
}

func expectGKENodePoolComputeClassNodeMetadataUpdate(t *testing.T, gcpCli *gcp.MockClient, nodePoolName string, currentLabels map[string]string, currentTaints []*gcpcontainer.NodeTaint, wantLabels map[string]string, wantTaints []*gcpcontainer.NodeTaint, op *gcpcontainer.Operation) {
	t.Helper()
	nodePoolResourceName := testGKENodePoolResourceName(nodePoolName)
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), nodePoolResourceName).
		Return(currentLabels, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), nodePoolResourceName).
		Return(currentTaints, nil)
	gcpCli.EXPECT().
		UpdateNodePoolNodeLabelsAndTaints(gomock.Any(), nodePoolResourceName, gomock.Any(), gomock.Any()).
		DoAndReturn(func(_ context.Context, _ string, labels map[string]string, taints []*gcpcontainer.NodeTaint) (*gcpcontainer.Operation, error) {
			assert.Equal(t, wantLabels, labels)
			assert.ElementsMatch(t, wantTaints, taints)
			return op, nil
		})
}

func testGKEComputeClassNodeTaint(computeClass string) *gcpcontainer.NodeTaint {
	return &gcpcontainer.NodeTaint{
		Key:    consts.GKEComputeClassLabelKey,
		Value:  computeClass,
		Effect: "NO_SCHEDULE",
	}
}

func testGKEHeterogeneousIsolationNodeTaint() *gcpcontainer.NodeTaint {
	return &gcpcontainer.NodeTaint{
		Key:    consts.GKEHeterogeneousNodePoolIsolationTaintKey,
		Value:  consts.GKEHeterogeneousNodePoolIsolationTaintValue,
		Effect: "NO_SCHEDULE",
	}
}

func testGKEInstanceGroupURL(instanceGroupName string) string {
	return "https://www.googleapis.com/compute/v1/projects/test-project/zones/us-central1-a/instanceGroupManagers/" + instanceGroupName
}

func markTestAPIManagedGKEHeterogeneousNodePool(nodePool *gkev1beta1.NodePool) {
	if nodePool.Labels == nil {
		nodePool.Labels = map[string]string{}
	}
	nodePool.Labels[consts.GKEHeterogeneousNodePoolManagedByLabelKey] = consts.GKEHeterogeneousNodePoolManagedByLabelValue
}

func markTestGKEHeterogeneousNodePoolEverBound(nodePool *gkev1beta1.NodePool) {
	if nodePool.Annotations == nil {
		nodePool.Annotations = map[string]string{}
	}
	nodePool.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound] = "true"
}

func testGKESharedScopeCluster(name string) *cloudv1.Cluster {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      name,
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	return cluster
}

func TestNodePool_patchGKENodePoolComputeClassMetadataMarksEverBoundWhenMetadataAlreadyMatches(t *testing.T) {
	nodePool := existingTestNodePool("extra-pool", "n2-standard-8")
	nodepoolutil.ApplyGKEComputeClassMetadataForNodePool(nodePool, "tikv-priority", "")

	local := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(nodePool).Build()
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Local: local,
			},
		},
	}

	err := np.patchGKENodePoolComputeClassMetadata(context.Background(), nodePool, gkeNodePoolComputeClassMetadata{
		desired:       "tikv-priority",
		markEverBound: true,
	})
	assert.NoError(t, err)
	assert.Equal(t, "true", nodePool.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound])

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, "tikv-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, "true", got.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound])
}

func TestNodePool_patchGKENodePoolComputeClassMetadataDoesNotReparkActiveAPIManagedPoolOnDesiredProjection(t *testing.T) {
	computeClass := "tikv-priority"
	nodePool := existingTestNodePool("extra-pool", "n2-standard-8")
	markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
	nodepoolutil.ApplyGKEComputeClassMetadataForNodePool(nodePool, computeClass, computeClass)

	local := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(nodePool).Build()
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Local: local,
			},
		},
	}

	err := np.patchGKENodePoolComputeClassMetadata(context.Background(), nodePool, gkeNodePoolComputeClassMetadata{
		desired:       computeClass,
		markEverBound: true,
	})
	assert.NoError(t, err)
	assert.Equal(t, computeClass, nodePool.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClass, nodePool.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, "true", nodePool.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound])
	assert.Equal(t, computeClass, nodePool.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
	assert.Equal(t, computeClass, nodepoolutil.GKEComputeClassSpecTaintForNodePool(nodePool))
	assert.False(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(nodePool))

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, computeClass, got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.False(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(got))
}

func TestNodePool_patchGKENodePoolComputeClassMetadataClearsResidualIsolationTaintForActivePool(t *testing.T) {
	computeClass := "tikv-priority"
	nodePool := existingTestNodePool("extra-pool", "n2-standard-8")
	markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
	nodepoolutil.ApplyGKEComputeClassMetadataForNodePool(nodePool, computeClass, computeClass)
	nodepoolutil.SetGKEHeterogeneousNodePoolIsolationTaint(nodePool)

	local := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(nodePool).Build()
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Local: local,
			},
		},
	}

	err := np.patchGKENodePoolComputeClassMetadata(context.Background(), nodePool, gkeNodePoolComputeClassMetadata{
		desired: computeClass,
		applied: computeClass,
	})
	assert.NoError(t, err)
	assert.Equal(t, computeClass, nodePool.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, computeClass, nodepoolutil.GKEComputeClassSpecTaintForNodePool(nodePool))
	assert.False(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(nodePool))

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.False(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(got))
}

func TestNodePool_patchGKENodePoolComputeClassMetadataRestoresIsolationTaintForCleanupGuard(t *testing.T) {
	computeClass := "tikv-priority"
	nodePool := existingTestNodePool("extra-pool", "n2-standard-8")
	markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
	nodepoolutil.ApplyGKEComputeClassMetadataForNodePool(nodePool, "", computeClass)

	local := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(nodePool).Build()
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Local: local,
			},
		},
	}

	err := np.patchGKENodePoolComputeClassMetadata(context.Background(), nodePool, gkeNodePoolComputeClassMetadata{
		applied: computeClass,
	})
	assert.NoError(t, err)
	assert.Equal(t, computeClass, nodePool.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, computeClass, nodepoolutil.GKEComputeClassSpecTaintForNodePool(nodePool))
	assert.True(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(nodePool))

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.True(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(got))
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataPatchesExistingAPIManagedPool(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingTestNodePool("extra-pool", "n2-standard-64")
	extra.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(extra)
	extra.Annotations = map[string]string{
		annoKeyResourceLabels: "tidb_cluster:" + cluster.Name,
	}
	extra.Spec.ForProvider.Autoscaling = &gkev1beta1.NodePoolAutoscaling{
		MinNodeCount: ptr.To(int64(2)),
		MaxNodeCount: ptr.To(int64(8)),
	}
	extra.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	extra.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("extra-pool")
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, "extra-pool", computeClassName)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, "true", got.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound])
	assert.Equal(t, computeClassName, got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
	assert.Equal(t, int64(2), *got.Spec.ForProvider.Autoscaling.MinNodeCount)
	assert.Equal(t, int64(8), *got.Spec.ForProvider.Autoscaling.MaxNodeCount)
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataRejectsUnmanagedExtraBindingPool(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingSharedComputeClassTestNodePool(cluster, "extra-pool", "n2-standard-64", "")
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "nodepool extra-pool is not managed by heterogeneous nodepool API")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKEHeterogeneousNodePoolEverBound)
	assert.NotContains(t, got.Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataKeepsSpecLabelWhenAnnotationMatches(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingSharedComputeClassTestNodePool(cluster, "extra-pool", "n2-standard-64", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(extra)
	extra.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("extra-pool")

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName("extra-pool")).
		Return(map[string]string{
			consts.GKEComputeClassLabelKey: computeClassName,
			"existing":                     "keep",
		}, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName("extra-pool")).
		Return([]*gcpcontainer.NodeTaint{
			testGKEComputeClassNodeTaint(computeClassName),
		}, nil)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, computeClassName, got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataWaitsWhenGKENodeLabelOperationSubmitted(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingTestNodePool("extra-pool", "n2-standard-64")
	extra.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(extra)
	extra.Annotations = map[string]string{
		annoKeyResourceLabels: "tidb_cluster:" + cluster.Name,
	}
	extra.Spec.ForProvider.Config.Labels = map[string]string{
		consts.GKEComputeClassLabelKey: "legacy-priority",
	}
	extra.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	extra.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("extra-pool")

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataUpdate(t, gcpCli, "extra-pool",
		map[string]string{"existing": "keep"},
		[]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
		},
		map[string]string{
			"existing":                     "keep",
			consts.GKEComputeClassLabelKey: computeClassName,
		},
		[]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
			testGKEComputeClassNodeTaint(computeClassName),
		},
		&gcpcontainer.Operation{Name: "operation-1"},
	)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "set compute class node metadata for nodepool extra-pool")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClassApplied)
	assert.Equal(t, "legacy-priority", got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataPatchesLocalMetadataAfterGKENodeLabelConfirmed(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingTestNodePool("extra-pool", "n2-standard-64")
	extra.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(extra)
	extra.Annotations = map[string]string{
		annoKeyResourceLabels: "tidb_cluster:" + cluster.Name,
	}
	extra.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	extra.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("extra-pool")

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName("extra-pool")).
		Return(map[string]string{
			"existing":                     "keep",
			consts.GKEComputeClassLabelKey: computeClassName,
		}, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName("extra-pool")).
		Return([]*gcpcontainer.NodeTaint{
			testGKEComputeClassNodeTaint(computeClassName),
		}, nil)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, computeClassName, got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataRequiresScopeLabels(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	existing.Spec.ForProvider.Autoscaling = &gkev1beta1.NodePoolAutoscaling{
		MinNodeCount: ptr.To(int64(2)),
		MaxNodeCount: ptr.To(int64(8)),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "missing required GKE heterogeneous nodepool scope labels")
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataRefreshesExtraResourceLabels(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingTestNodePool("extra-pool", "n2-standard-64")
	extra.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(extra)
	extra.Spec.ForProvider.Autoscaling = &gkev1beta1.NodePoolAutoscaling{
		MinNodeCount: ptr.To(int64(2)),
		MaxNodeCount: ptr.To(int64(8)),
	}
	extra.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	extra.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/extra-pool"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	nodePoolResourceName := "projects/test-project/locations/us-central1/clusters/gke-test/nodePools/extra-pool"
	gcpCli.EXPECT().
		GetNodePoolResourceLabels(gomock.Any(), nodePoolResourceName).
		Return(map[string]string{"existing": "keep"}, nil)
	gcpCli.EXPECT().
		UpdateNodePoolResourceLabels(gomock.Any(), nodePoolResourceName, gomock.Any()).
		DoAndReturn(func(_ context.Context, _ string, labels map[string]string) (*gcpcontainer.Operation, error) {
			assert.Equal(t, "keep", labels["existing"])
			assert.Equal(t, "project-1", labels["project"])
			assert.Equal(t, "tenant-1", labels["tenant"])
			assert.Equal(t, "dedicated", labels["servicetype"])
			assert.Equal(t, "customer-tidb", labels["usedby"])
			return nil, nil
		})
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, "extra-pool", computeClassName)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	resourceLabels := got.Annotations[annoKeyResourceLabels]
	assert.Contains(t, resourceLabels, "existing:keep")
	assert.Contains(t, resourceLabels, "project:project-1")
	assert.Contains(t, resourceLabels, "tenant:tenant-1")
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, computeClassName, got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
	assert.Equal(t, int64(2), *got.Spec.ForProvider.Autoscaling.MinNodeCount)
	assert.Equal(t, int64(8), *got.Spec.ForProvider.Autoscaling.MaxNodeCount)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataRefreshesPreparedResourceLabels(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"

	prepared := existingTestNodePool("prepared-pool", "n2-standard-64")
	prepared.Labels = sharedTestNodePoolLabels(cluster)
	prepared.Labels[consts.GKEHeterogeneousNodePoolManagedByLabelKey] = consts.GKEHeterogeneousNodePoolManagedByLabelValue
	prepared.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	prepared.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/prepared-pool"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	nodePoolResourceName := "projects/test-project/locations/us-central1/clusters/gke-test/nodePools/prepared-pool"
	gcpCli.EXPECT().
		GetNodePoolResourceLabels(gomock.Any(), nodePoolResourceName).
		Return(map[string]string{"existing": "keep"}, nil)
	gcpCli.EXPECT().
		UpdateNodePoolResourceLabels(gomock.Any(), nodePoolResourceName, gomock.Any()).
		DoAndReturn(func(_ context.Context, _ string, labels map[string]string) (*gcpcontainer.Operation, error) {
			assert.Equal(t, "keep", labels["existing"])
			assert.Equal(t, "project-1", labels["project"])
			assert.Equal(t, "tenant-1", labels["tenant"])
			assert.Equal(t, "dedicated", labels["servicetype"])
			assert.Equal(t, "customer-tidb", labels["usedby"])
			return nil, nil
		})
	local := &recordingClusterListClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(prepared).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "prepared-pool"}, got))
	resourceLabels := got.Annotations[annoKeyResourceLabels]
	assert.Contains(t, resourceLabels, "existing:keep")
	assert.Contains(t, resourceLabels, "project:project-1")
	assert.Contains(t, resourceLabels, "tenant:tenant-1")
	assert.Empty(t, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Empty(t, got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
	assert.Empty(t, local.clusterListSelectors)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataRestoresPreparedIsolationTaintWhenResourceLabelsRecorded(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
				cloudv1.RegionLabel:       "us-west1",
				cloudv1.ProjectTagKey:     "project-1",
				cloudv1.TenantTagKey:      "tenant-1",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	cluster.Spec.Kubernetes.Name = "gke-test"

	prepared := existingTestNodePool("prepared-pool", "n2-standard-64")
	prepared.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(prepared)
	prepared.Annotations = map[string]string{
		annoKeyResourceLabels: "existing:keep",
	}
	prepared.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	prepared.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("prepared-pool")
	assert.False(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(prepared))

	local := &recordingClusterListClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(prepared).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "prepared-pool"}, got))
	assert.True(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(got))
	assert.Equal(t, "existing:keep", got.Annotations[annoKeyResourceLabels])
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.Empty(t, local.clusterListSelectors)
}

func TestNodePool_ensureGKENodePoolComputeClassNodeMetadataWithRequeue(t *testing.T) {
	nodePoolResourceName := testGKENodePoolResourceName("shared-pool")

	tests := []struct {
		name               string
		computeClass       string
		apiManaged         bool
		currentLabels      map[string]string
		currentTaints      []*gcpcontainer.NodeTaint
		wantLabels         map[string]string
		wantTaints         []*gcpcontainer.NodeTaint
		updateErr          error
		wantCombinedUpdate bool
		wantTaintUpdate    bool
		wantSynced         bool
		wantSubmitted      bool
		wantRequeueOnError bool
		wantErrorContains  string
	}{
		{
			name:         "skips when label already matches",
			computeClass: "tikv-priority",
			currentLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantSynced: true,
		},
		{
			name:         "returns submitted after compute class metadata update succeeds",
			computeClass: "tikv-priority",
			currentLabels: map[string]string{
				"existing": "keep",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				{Key: "dedicated", Value: "tidb-shared", Effect: "NO_SCHEDULE"},
			},
			wantLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				{Key: "dedicated", Value: "tidb-shared", Effect: "NO_SCHEDULE"},
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantCombinedUpdate: true,
			wantSubmitted:      true,
		},
		{
			name:         "returns submitted after label update succeeds",
			computeClass: "tikv-priority",
			currentLabels: map[string]string{
				"existing": "keep",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantCombinedUpdate: true,
			wantSubmitted:      true,
		},
		{
			name:         "returns submitted after taint update succeeds",
			computeClass: "tikv-priority",
			currentLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				{Key: "dedicated", Value: "tidb-shared", Effect: "NO_SCHEDULE"},
			},
			wantLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				{Key: "dedicated", Value: "tidb-shared", Effect: "NO_SCHEDULE"},
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantCombinedUpdate: true,
			wantSubmitted:      true,
		},
		{
			name:         "api-managed nodepool keeps isolation while adding compute class metadata",
			computeClass: "tikv-priority",
			apiManaged:   true,
			currentLabels: map[string]string{
				"existing": "keep",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
			},
			wantLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantCombinedUpdate: true,
			wantSubmitted:      true,
		},
		{
			name:         "returns submitted after isolation taint removal succeeds",
			computeClass: "tikv-priority",
			apiManaged:   true,
			currentLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantTaintUpdate: true,
			wantSubmitted:   true,
		},
		{
			name:         "returns submitted after isolation taint restore succeeds",
			computeClass: "",
			apiManaged:   true,
			currentLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "stale-priority",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("stale-priority"),
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("stale-priority"),
				testGKEHeterogeneousIsolationNodeTaint(),
			},
			wantTaintUpdate: true,
			wantSubmitted:   true,
		},
		{
			name:         "returns submitted after compute class metadata clear succeeds",
			computeClass: "",
			currentLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "stale-priority",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
				testGKEComputeClassNodeTaint("stale-priority"),
			},
			wantLabels: map[string]string{
				"existing": "keep",
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
			},
			wantCombinedUpdate: true,
			wantSubmitted:      true,
		},
		{
			name:         "returns submitted after compute class taint clear succeeds",
			computeClass: "",
			currentLabels: map[string]string{
				"existing": "keep",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
				testGKEComputeClassNodeTaint("stale-priority"),
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEHeterogeneousIsolationNodeTaint(),
			},
			wantLabels: map[string]string{
				"existing": "keep",
			},
			wantCombinedUpdate: true,
			wantSubmitted:      true,
		},
		{
			name:         "returns 30s retriable error after update fails",
			computeClass: "tikv-priority",
			currentLabels: map[string]string{
				"existing": "keep",
			},
			currentTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			wantLabels: map[string]string{
				"existing":                     "keep",
				consts.GKEComputeClassLabelKey: "tikv-priority",
			},
			wantTaints: []*gcpcontainer.NodeTaint{
				testGKEComputeClassNodeTaint("tikv-priority"),
			},
			updateErr:          fmt.Errorf("quota exceeded"),
			wantCombinedUpdate: true,
			wantRequeueOnError: true,
			wantErrorContains:  "quota exceeded",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			gcpCli := gcp.NewMockClient(ctrl)
			gcpCli.EXPECT().
				GetNodePoolNodeLabels(gomock.Any(), nodePoolResourceName).
				Return(tt.currentLabels, nil)
			gcpCli.EXPECT().
				GetNodePoolNodeTaints(gomock.Any(), nodePoolResourceName).
				Return(tt.currentTaints, nil)
			if tt.wantCombinedUpdate {
				gcpCli.EXPECT().
					UpdateNodePoolNodeLabelsAndTaints(gomock.Any(), nodePoolResourceName, gomock.Any(), gomock.Any()).
					DoAndReturn(func(_ context.Context, _ string, labels map[string]string, taints []*gcpcontainer.NodeTaint) (*gcpcontainer.Operation, error) {
						assert.Equal(t, tt.wantLabels, labels)
						assert.ElementsMatch(t, tt.wantTaints, taints)
						return nil, tt.updateErr
					})
			}
			if tt.wantTaintUpdate {
				gcpCli.EXPECT().
					UpdateNodePoolNodeTaints(gomock.Any(), nodePoolResourceName, gomock.Any()).
					DoAndReturn(func(_ context.Context, _ string, taints []*gcpcontainer.NodeTaint) (*gcpcontainer.Operation, error) {
						assert.Equal(t, tt.wantTaints, taints)
						return nil, tt.updateErr
					})
			}

			nodePool := existingTestNodePool("shared-pool", "n2-standard-8")
			if tt.apiManaged {
				if nodePool.Labels == nil {
					nodePool.Labels = map[string]string{}
				}
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
			}
			nodePool.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
			nodePool.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("shared-pool")
			np := &NodePool{gcpCli: gcpCli}

			synced, submitted, err := np.ensureGKENodePoolComputeClassNodeMetadataWithRequeue(context.Background(), nodePool, tt.computeClass)
			if tt.wantErrorContains != "" {
				assert.ErrorContains(t, err, tt.wantErrorContains)
			} else {
				assert.NoError(t, err)
			}
			if tt.wantRequeueOnError {
				result := k8s.ShouldRequeue(err)
				assert.True(t, result.Requeue)
				assert.Equal(t, gkeComputeClassNodeMetadataRequeueAfter, result.RequeueAfter)
			}
			assert.Equal(t, tt.wantSynced, synced)
			assert.Equal(t, tt.wantSubmitted, submitted)
		})
	}
}

func TestNodePool_ensureGKENodePoolComputeClassNodeMetadataWithRequeueSkipsUntilExternalNodePoolReady(t *testing.T) {
	tests := []struct {
		name     string
		status   string
		selfLink string
	}{
		{
			name:     "nodepool is not running",
			status:   gkev1beta1.NodePoolStateProvisioning,
			selfLink: testGKENodePoolSelfLink("shared-pool"),
		},
		{
			name:   "nodepool self link is not observed",
			status: gkev1beta1.NodePoolStateRunning,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			ctrl := gomock.NewController(t)
			gcpCli := gcp.NewMockClient(ctrl)
			nodePool := existingTestNodePool("shared-pool", "n2-standard-8")
			nodePool.Status.AtProvider.Status = tt.status
			nodePool.Status.AtProvider.SelfLink = tt.selfLink
			np := &NodePool{gcpCli: gcpCli}

			synced, submitted, err := np.ensureGKENodePoolComputeClassNodeMetadataWithRequeue(context.Background(), nodePool, "tikv-priority")
			assert.NoError(t, err)
			assert.False(t, synced)
			assert.False(t, submitted)
		})
	}
}

func TestNewGKEComputeClassNodeMetadataRequeueError(t *testing.T) {
	err := newGKEComputeClassNodeMetadataRequeueError(fmt.Errorf("update submitted"))

	result := k8s.ShouldRequeue(err)
	assert.True(t, result.Requeue)
	assert.Equal(t, gkeComputeClassNodeMetadataRequeueAfter, result.RequeueAfter)
}

func TestNodePool_cleanupGKENodePoolComputeClassMetadataRequeuesAfterGKENodeMetadataUpdateSubmitted(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("shared-pool")

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataUpdate(t, gcpCli, "shared-pool",
		map[string]string{
			"existing":                     "keep",
			consts.GKEComputeClassLabelKey: "stale-priority",
		},
		[]*gcpcontainer.NodeTaint{
			testGKEComputeClassNodeTaint("stale-priority"),
		},
		map[string]string{"existing": "keep"},
		[]*gcpcontainer.NodeTaint{},
		nil,
	)

	local := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build()
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: &task.Context{Local: local}},
		gcpCli:   gcpCli,
	}

	err := np.cleanupGKENodePoolComputeClassMetadata(context.Background(), existing)
	assert.Error(t, err)
	result := k8s.ShouldRequeue(err)
	assert.True(t, result.Requeue)
	assert.Equal(t, gkeComputeClassNodeMetadataRequeueAfter, result.RequeueAfter)
}

func TestNodePool_cleanupGKENodePoolComputeClassMetadataRestoresMissingSpecGuardBeforeRequeue(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	delete(existing.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
	existing.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied] = "stale-priority"
	existing.Spec.ForProvider.Config.Labels = map[string]string{
		"managed": "keep",
	}
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("shared-pool")

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataUpdate(t, gcpCli, "shared-pool",
		map[string]string{
			"existing":                     "keep",
			consts.GKEComputeClassLabelKey: "stale-priority",
		},
		[]*gcpcontainer.NodeTaint{
			testGKEComputeClassNodeTaint("stale-priority"),
		},
		map[string]string{"existing": "keep"},
		[]*gcpcontainer.NodeTaint{},
		&gcpcontainer.Operation{Name: "operation-1"},
	)

	local := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build()
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: &task.Context{Local: local}},
		gcpCli:   gcpCli,
	}

	err := np.cleanupGKENodePoolComputeClassMetadata(context.Background(), existing)
	assert.ErrorContains(t, err, "remove compute class node metadata for nodepool shared-pool")
	result := k8s.ShouldRequeue(err)
	assert.True(t, result.Requeue)
	assert.Equal(t, gkeComputeClassNodeMetadataRequeueAfter, result.RequeueAfter)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
	assert.Equal(t, "stale-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, "stale-priority", got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
	assert.Equal(t, "keep", got.Spec.ForProvider.Config.Labels["managed"])
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataSkipsOrdinarySharedNodePool(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)

	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
	assert.Equal(t, "stale-priority", nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.Equal(t, "stale-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, "stale-priority", got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
	candidates, err := gkeheterogeneous.ComputeClassCleanupCandidates(cluster)
	require.NoError(t, err)
	require.Empty(t, candidates)
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataHandlesDeployIntentDeletionByNodePoolOwnership(t *testing.T) {
	tests := []struct {
		name                         string
		nodePoolName                 string
		apiManaged                   bool
		wantComputeClass             string
		expectGKENodeLabelRemoval    bool
		expectLocalComputeClassClear bool
	}{
		{
			name:             "ordinary shared nodepool keeps stale compute class metadata",
			nodePoolName:     "base-pool",
			wantComputeClass: "stale-priority",
		},
		{
			name:                         "api managed extra nodepool clears stale compute class metadata",
			nodePoolName:                 "extra-pool",
			apiManaged:                   true,
			expectGKENodeLabelRemoval:    true,
			expectLocalComputeClassClear: true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "cluster-a",
					Namespace: "ns",
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
					},
				},
				Spec: cloudv1.ClusterSpec{
					NetworkName: "shared-network",
					Cluster: cloudv1.ClusterTopology{
						TiKV: cloudv1.TiKV{
							Replicas:     3,
							InstanceType: "n2-standard-8",
						},
					},
				},
			}
			cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
			setTestGKEAPPBoxLabels(cluster)

			existing := existingSharedComputeClassTestNodePool(cluster, tt.nodePoolName, "n2-standard-8", "stale-priority")
			existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(tt.nodePoolName)
			if tt.apiManaged {
				markTestAPIManagedGKEHeterogeneousNodePool(existing)
				markTestGKEHeterogeneousNodePoolEverBound(existing)
				existing.Spec.ForProvider.Autoscaling = &gkev1beta1.NodePoolAutoscaling{
					MinNodeCount: ptr.To(int64(2)),
					MaxNodeCount: ptr.To(int64(8)),
				}
			}

			ctrl := gomock.NewController(t)
			gcpCli := gcp.NewMockClient(ctrl)
			if tt.expectGKENodeLabelRemoval {
				gcpCli.EXPECT().
					GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName(tt.nodePoolName)).
					Return(map[string]string{}, nil)
				gcpCli.EXPECT().
					GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName(tt.nodePoolName)).
					Return([]*gcpcontainer.NodeTaint{
						testGKEHeterogeneousIsolationNodeTaint(),
					}, nil)
			}

			ctx := &task.Context{
				Cluster: cluster,
				Namer: namer.NewNamer().
					WithClusterName(cluster.Name).
					WithTiDBNamespace(cluster.PrefixedId()),
				Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
				EnvProvider: &testEnvProvider{},
				CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
					NodeVersion: "1.35",
				},
			}
			setContextZones(ctx, []string{"us-central1-a"})
			np := &NodePool{
				BaseTask: &task.BaseTask{
					Context: ctx,
				},
				gcpCli: gcpCli,
			}

			err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
			assert.NoError(t, err)

			got := &gkev1beta1.NodePool{}
			assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: tt.nodePoolName}, got))
			assert.Equal(t, tt.wantComputeClass, nodepoolutil.GKEComputeClassNameForNodePool(got))
			if tt.apiManaged {
				assert.Equal(t, consts.GKEHeterogeneousNodePoolManagedByLabelValue, got.Labels[consts.GKEHeterogeneousNodePoolManagedByLabelKey])
				assert.Equal(t, "true", got.Annotations[consts.AnnoKeyGKEHeterogeneousNodePoolEverBound])
				assert.Equal(t, int64(2), *got.Spec.ForProvider.Autoscaling.MinNodeCount)
				assert.Equal(t, int64(8), *got.Spec.ForProvider.Autoscaling.MaxNodeCount)
			}
			if tt.expectLocalComputeClassClear {
				assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
				assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClassApplied)
				assert.NotContains(t, got.Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
			} else {
				assert.Equal(t, "stale-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
				assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClassApplied)
				assert.Equal(t, "stale-priority", got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
			}
		})
	}
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataSkipsGKENodeLabelWithoutLocalMetadata(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)

	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "")
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
	assert.NotContains(t, got.Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataKeepsAppliedMetadataWhenGKENodeLabelOperationSubmitted(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)

	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataUpdate(t, gcpCli, "shared-pool",
		map[string]string{
			"existing":                     "keep",
			consts.GKEComputeClassLabelKey: "stale-priority",
		},
		[]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
			testGKEComputeClassNodeTaint("stale-priority"),
		},
		map[string]string{"existing": "keep"},
		[]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
		},
		&gcpcontainer.Operation{Name: "operation-1"},
	)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "remove compute class node metadata for nodepool shared-pool")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
	assert.Equal(t, "stale-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
	assert.Equal(t, "stale-priority", got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataKeepsLocalMetadataUntilGKENodeMetadataSyncable(t *testing.T) {
	tests := []struct {
		name     string
		status   string
		selfLink string
	}{
		{
			name:     "nodepool is not running",
			status:   gkev1beta1.NodePoolStateProvisioning,
			selfLink: "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool",
		},
		{
			name:   "nodepool self link is not observed",
			status: gkev1beta1.NodePoolStateRunning,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "cluster-a",
					Namespace: "ns",
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
					},
				},
				Spec: cloudv1.ClusterSpec{
					NetworkName: "shared-network",
					Cluster: cloudv1.ClusterTopology{
						TiKV: cloudv1.TiKV{
							Replicas:     3,
							InstanceType: "n2-standard-8",
						},
					},
				},
			}
			cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
			setTestGKEAPPBoxLabels(cluster)

			existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
			markTestAPIManagedGKEHeterogeneousNodePool(existing)
			existing.Status.AtProvider.Status = tt.status
			existing.Status.AtProvider.SelfLink = tt.selfLink
			ctrl := gomock.NewController(t)
			gcpCli := gcp.NewMockClient(ctrl)
			ctx := &task.Context{
				Cluster:     cluster,
				Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
				EnvProvider: &testEnvProvider{},
			}
			np := &NodePool{
				BaseTask: &task.BaseTask{Context: ctx},
				gcpCli:   gcpCli,
			}

			err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
			assert.ErrorContains(t, err, "not ready to sync node metadata")

			got := &gkev1beta1.NodePool{}
			assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
			assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
			assert.Equal(t, "stale-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
			assert.Equal(t, "stale-priority", got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
		})
	}
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataClearsSpecOnlyMetadata(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)

	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey] = "stale-priority"
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	nodePoolResourceName := "projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool"
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), nodePoolResourceName).
		Return(map[string]string{}, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), nodePoolResourceName).
		Return([]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
		}, nil)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
	assert.NotContains(t, got.Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
}

func TestNodePool_cleanupUnusedAPIManagedGKEHeterogeneousNodePoolsDeletesOnlySafeEverBoundPool(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	nodePoolName := "extra-pool"
	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	markTestGKEHeterogeneousNodePoolEverBound(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	existing.Status.AtProvider.InstanceGroupUrls = []string{testGKEInstanceGroupURL(nodePoolName + "-ig")}

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
		Return(map[string]string{}, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
		Return([]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
		}, nil)
	gcpCli.EXPECT().
		GetInstanceGroupManagerManagedInstanceCount(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
		Return(int64(0), nil)
	gcpCli.EXPECT().
		GetInstanceGroupSize(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
		Return(int64(0), nil)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	deleted, err := np.cleanupUnusedAPIManagedGKEHeterogeneousNodePools(context.Background())
	assert.NoError(t, err)
	assert.True(t, deleted)

	got := &gkev1beta1.NodePool{}
	assert.Error(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
}

func TestNodePool_cleanupUnusedAPIManagedGKEHeterogeneousNodePoolsKeepsUnsafeCandidates(t *testing.T) {
	nodePoolName := "extra-pool"
	tests := []struct {
		name      string
		mutate    func(*testing.T, *cloudv1.Cluster, *gkev1beta1.NodePool)
		expectGCP func(*testing.T, *gcp.MockClient)
		wantErr   string
	}{
		{
			name: "keeps pure prepared pool that was never bound",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
			},
		},
		{
			name: "keeps ordinary shared pool even with historical marker",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
			},
		},
		{
			name: "keeps pool still referenced by current intent",
			mutate: func(t *testing.T, cluster *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
				updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority",
					[]string{"n2-standard-8", "n2-standard-64"},
					[]string{gkeheterogeneous.NodePoolName(cluster, "n2-standard-8"), nodePoolName},
				)
			},
		},
		{
			name: "keeps pool before instance groups are observed",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
				nodePool.Status.AtProvider.InstanceGroupUrls = nil
			},
		},
		{
			name: "keeps pool before gke node label can be synced",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
				nodePool.Status.AtProvider.SelfLink = ""
			},
		},
		{
			name: "keeps pool from different gcp cluster",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
				nodePool.Spec.ForProvider.Cluster = "projects/test-project/locations/us-central1/clusters/other-gke"
			},
		},
		{
			name: "keeps pool from different kubernetes cluster scope",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
				nodePool.Labels[cloudv1.ProjectTagKey] = "project-2"
			},
		},
		{
			name: "keeps pool with existing backing nodes",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
			},
			expectGCP: func(t *testing.T, gcpCli *gcp.MockClient) {
				t.Helper()
				gcpCli.EXPECT().
					GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
					Return(map[string]string{}, nil)
				gcpCli.EXPECT().
					GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
					Return([]*gcpcontainer.NodeTaint{
						testGKEHeterogeneousIsolationNodeTaint(),
					}, nil)
				gcpCli.EXPECT().
					GetInstanceGroupManagerManagedInstanceCount(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
					Return(int64(1), nil)
			},
		},
		{
			name: "keeps pool when instance group still reports backing nodes",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
			},
			expectGCP: func(t *testing.T, gcpCli *gcp.MockClient) {
				t.Helper()
				gcpCli.EXPECT().
					GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
					Return(map[string]string{}, nil)
				gcpCli.EXPECT().
					GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
					Return([]*gcpcontainer.NodeTaint{
						testGKEHeterogeneousIsolationNodeTaint(),
					}, nil)
				gcpCli.EXPECT().
					GetInstanceGroupManagerManagedInstanceCount(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
					Return(int64(0), nil)
				gcpCli.EXPECT().
					GetInstanceGroupSize(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
					Return(int64(1), nil)
			},
		},
		{
			name: "removes stale gke node label before delete",
			mutate: func(t *testing.T, _ *cloudv1.Cluster, nodePool *gkev1beta1.NodePool) {
				t.Helper()
				markTestAPIManagedGKEHeterogeneousNodePool(nodePool)
				markTestGKEHeterogeneousNodePoolEverBound(nodePool)
			},
			expectGCP: func(t *testing.T, gcpCli *gcp.MockClient) {
				t.Helper()
				expectGKENodePoolComputeClassNodeMetadataUpdate(t, gcpCli, nodePoolName,
					map[string]string{consts.GKEComputeClassLabelKey: "stale-priority"},
					[]*gcpcontainer.NodeTaint{
						testGKEHeterogeneousIsolationNodeTaint(),
					},
					map[string]string{},
					[]*gcpcontainer.NodeTaint{
						testGKEHeterogeneousIsolationNodeTaint(),
					},
					nil,
				)
			},
			wantErr: "remove compute class node metadata for nodepool extra-pool",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := testGKESharedScopeCluster("cluster-a")
			existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
			existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
			existing.Status.AtProvider.InstanceGroupUrls = []string{testGKEInstanceGroupURL(nodePoolName + "-ig")}
			tt.mutate(t, cluster, existing)

			ctrl := gomock.NewController(t)
			gcpCli := gcp.NewMockClient(ctrl)
			if tt.expectGCP != nil {
				tt.expectGCP(t, gcpCli)
			}

			ctx := &task.Context{
				Cluster:     cluster,
				Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
				EnvProvider: &testEnvProvider{},
			}
			np := &NodePool{
				BaseTask: &task.BaseTask{Context: ctx},
				gcpCli:   gcpCli,
			}

			deleted, err := np.cleanupUnusedAPIManagedGKEHeterogeneousNodePools(context.Background())
			assert.False(t, deleted)
			if tt.wantErr != "" {
				assert.ErrorContains(t, err, tt.wantErr)
			} else {
				assert.NoError(t, err)
			}

			got := &gkev1beta1.NodePool{}
			assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
		})
	}
}

func TestNodePool_cleanupUnusedAPIManagedGKEHeterogeneousNodePoolsSkipsWhenPeerViewIncomplete(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	nodePoolName := "extra-pool"
	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	markTestGKEHeterogeneousNodePoolEverBound(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	existing.Status.AtProvider.InstanceGroupUrls = []string{testGKEInstanceGroupURL(nodePoolName + "-ig")}

	peer := testGKESharedScopeCluster("cluster-b")
	peer.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, peer).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	deleted, err := np.cleanupUnusedAPIManagedGKEHeterogeneousNodePools(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataBuildsSharedPlanOnce(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, nodePoolName, "n2-standard-8")
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)
	local := &recordingClusterListClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster, base, extra).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)
	assert.Len(t, local.clusterListSelectors, 1)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataSkipsPeerPlanWithoutLocalBindingOrCandidate(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	local := &recordingClusterListClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster, sibling).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)
	assert.Empty(t, local.clusterListSelectors)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataSkipsOrdinaryGKEWithoutBindingOrAPIManagedNodePool(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	ordinary := existingTestNodePool("ordinary-pool", "n2-standard-8")
	ordinary.Labels = sharedTestNodePoolLabels(cluster)
	ordinary.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	ordinary.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("ordinary-pool")

	local := &recordingClusterListClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster, ordinary).Build(),
	}
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)
	assert.Empty(t, local.clusterListSelectors)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: ordinary.Name}, got))
	assert.False(t, nodepoolutil.HasGKEHeterogeneousNodePoolIsolationTaint(got))
	assert.NotContains(t, got.Labels, consts.GKEHeterogeneousNodePoolManagedByLabelKey)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataSkipsPeerPlanForPurePreparedAPIManagedPool(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)
	prepared := existingSharedComputeClassTestNodePool(cluster, "prepared-pool", "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(prepared)
	prepared.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("prepared-pool")

	local := &recordingClusterListClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster, sibling, prepared).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)
	assert.Empty(t, local.clusterListSelectors)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataRequeuesAPIManagedListErrorWithoutLocalBinding(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	local := &nodePoolListFailingClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.Error(t, err)
	assert.True(t, k8s.RequeueOnly(err))
	assert.False(t, deleted)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataReturnsAPIManagedListErrorWithLocalBinding(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})
	local := &nodePoolListFailingClient{
		Client: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster).Build(),
	}
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       local,
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "nodepool list failed")
	assert.False(t, deleted)
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataSyncsGKENodeLabelWhenLocalAnnotationAlreadyMatches(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/" + nodePoolName
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ordinaryNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	ordinary := existingTestNodePool(ordinaryNodePoolName, "n2-standard-8")
	ordinary.Annotations = map[string]string{annoKeyResourceLabels: "tidb_cluster:" + cluster.Name}
	ordinary.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	ordinary.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(ordinaryNodePoolName)

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, extra, ordinary).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied])
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataReturnsRetryWhenGKENodeLabelNotSyncable(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.Status = gkev1beta1.NodePoolStateProvisioning
	existing.Status.AtProvider.SelfLink = ""

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "non-active nodepool")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClassApplied)
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataPatchesDesiredBeforeExtraResourceLabelBackfill(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})

	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	extra := existingTestNodePool("extra-pool", "n2-standard-64")
	extra.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(extra)
	extra.Status.AtProvider.Status = gkev1beta1.NodePoolStateProvisioning
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, extra).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "non-active nodepool extra-pool")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClassApplied)
	assert.NotContains(t, got.Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
}

func TestNodePool_reconcileCurrentGKEComputeClassBindingNodePoolMetadataReturnsRetryWhenBoundPoolMissing(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureGCP3LayerNodePool)
	setTestGKEAPPBoxLabels(cluster)
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "missing-extra-pool"})

	base := existingTestNodePool(baseName, "n2-standard-8")
	base.Labels = sharedTestNodePoolLabels(cluster)
	markTestAPIManagedGKEHeterogeneousNodePool(base)
	base.Annotations = map[string]string{
		annoKeyResourceLabels: "tidb_cluster:" + cluster.Name,
	}
	base.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	base.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(baseName)
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "gke compute class binding nodepool")
	assert.ErrorContains(t, err, "missing-extra-pool")
	assert.ErrorContains(t, err, "component tikv")
	assert.ErrorContains(t, err, "instance type n2-standard-64")
	assert.ErrorContains(t, err, "is missing")
	result := k8s.ShouldRequeue(err)
	assert.True(t, result.Requeue)
	assert.Equal(t, gkeComputeClassNodeMetadataRequeueAfter, result.RequeueAfter)

	got := &gkev1beta1.NodePool{}
	err = ctx.Local.Get(context.Background(), client.ObjectKey{Name: "missing-extra-pool"}, got)
	assert.Error(t, err)
}

func TestNodePool_reconcileGKEComputeClassNodePoolMetadataDoesNotProjectPeerOnlyIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	updateTestGKEComputeClassBinding(t, sibling, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "peer-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(sibling, "n2-standard-8"), "peer-extra"})

	peerExtra := existingSharedComputeClassTestNodePool(cluster, "peer-extra", "n2-standard-64", "")
	ctx := &task.Context{
		Cluster: cluster,
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			sibling,
			peerExtra,
		).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "peer-extra"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_desiredNodePoolsIgnoresSiblingComputeClassNameConflict(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster1",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-compute-class", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(cluster, "n2-standard-8"), "current-extra"})
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster2"
	updateTestGKEComputeClassBinding(t, sibling, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-compute-class", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(sibling, "n2-standard-8"), "shared-extra"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			sibling,
			existingTestNodePool("shared-pool", "n2-standard-8"),
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a", "us-central1-b", "us-central1-c"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
	assert.NotContains(t, nodePoolsByName, "current-extra")
	assert.NotContains(t, nodePoolsByName, "shared-extra")
}

func TestNodePool_desiredNodePoolsDoesNotUseComputeClassForDefaultTiDBGroupID(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default-group",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiDBBindings, "tidb-priority-v2", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "custom-tidb-extra"})
	cluster.Spec.ExtraGroups = []cloudv1.ExtraGroupSpec{
		{
			Name: "default-group",
			TiDB: cloudv1.TiDB{
				Replicas:     2,
				InstanceType: "n2-standard-64",
			},
		},
	}

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			existingTestNodePool(baseName, "n2-standard-8"),
			existingTestNodePool("custom-tidb-extra", "n2-standard-64"),
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)

	defaultNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	if assert.Contains(t, nodePoolsByName, defaultNodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[defaultNodePoolName]))
	}
	extraGroupNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-64", "1.35")
	if assert.Contains(t, nodePoolsByName, extraGroupNodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[extraGroupNodePoolName]))
	}
	assert.NotContains(t, nodePoolsByName, baseName)
	assert.NotContains(t, nodePoolsByName, "custom-tidb-extra")
}

func TestNodePool_desiredNodePoolsClearsExistingNodePoolComputeClassWithoutActiveIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	existing := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: nodePoolName,
			Annotations: map[string]string{
				consts.AnnoKeyGKENodePoolComputeClass: "old-priority",
			},
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Config: &gkev1beta1.NodeConfig{
					Labels: map[string]string{
						consts.GKEComputeClassLabelKey: "old-priority",
					},
				},
			},
		},
	}
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.NotNil(t, nodePoolsByName[nodePoolName].Annotations)
		assert.Empty(t, nodePoolsByName[nodePoolName].Annotations)
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
		assert.NotContains(t, nodePoolsByName[nodePoolName].Annotations, consts.AnnoKeyGKENodePoolComputeClass)
		assert.NotContains(t, nodePoolsByName[nodePoolName].Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
	}
}

func TestNodePool_desiredNodePoolsIgnoresSiblingSharedComputeClassWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")
	updateTestGKEComputeClassBinding(t, sibling, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(sibling).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
}

func TestNodePool_desiredNodePoolsIgnoresSiblingSharedComputeClassForDifferentComponentWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.Spec.Cluster.TiKV = cloudv1.TiKV{
		Replicas:     3,
		InstanceType: "n2-standard-8",
	}
	sibling.Spec.Cluster.TiDB = cloudv1.TiDB{}
	sibling.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")
	updateTestGKEComputeClassBinding(t, sibling, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(sibling).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
}

func TestNodePool_desiredNodePoolsDoesNotValidateLocalHeterogeneousIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	cluster.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `{"computeClassName":"shared-priority","nodePools":[{"role":"base","instanceType":"n2-standard-8"},{"role":"extra","instanceType":"n2-standard-64","nodePoolName":"extra-pool"}]}`)

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
}

func TestNodePool_desiredNodePoolsIgnoresMalformedSiblingSharedComputeClassIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	currentNodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{currentNodePoolName})
	siblingA := cluster.DeepCopy()
	siblingA.Name = "cluster-b"
	siblingA.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")
	siblingA.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(siblingA).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	ordinaryNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	if assert.Contains(t, nodePoolsByName, ordinaryNodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[ordinaryNodePoolName]))
	}
	assert.NotContains(t, nodePoolsByName, currentNodePoolName)
}

func TestNodePool_desiredNodePoolsClearsExistingComputeClassWhenPeerViewIncomplete(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")
	sibling.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "existing-priority")
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(sibling, existing).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
}

func TestNodePool_buildSharedGKEComputeClassPlanUsesHeterogeneousNodePoolScopeBoundedPeerLookup(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)

	peer := cluster.DeepCopy()
	peer.Name = "cluster-b"
	peer.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	peerBaseNodePoolName := gkeheterogeneous.NodePoolName(peer, "n2-standard-16")
	updateTestGKEComputeClassBinding(t, peer, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-16"}, []string{peerBaseNodePoolName})

	sameAPPBoxMalformed := cluster.DeepCopy()
	sameAPPBoxMalformed.Name = "cluster-c"
	sameAPPBoxMalformed.Spec.NetworkName = "isolated-network"
	sameAPPBoxMalformed.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	differentAPPBoxMalformed := cluster.DeepCopy()
	differentAPPBoxMalformed.Name = "cluster-d"
	differentAPPBoxMalformed.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)
	setTestGKEAPPBoxLabelsWithProject(differentAPPBoxMalformed, "project-2")

	baseClient := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster, peer, sameAPPBoxMalformed, differentAPPBoxMalformed).Build()
	recordingClient := &recordingClusterListClient{Client: baseClient}
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Cluster: cluster,
				Local:   recordingClient,
			},
		},
	}

	plan, err := np.buildSharedGKEComputeClassPlan(context.Background())
	assert.NoError(t, err)
	expectedSelector := k8slabels.SelectorFromSet(k8slabels.Set{
		cloudv1.TenantTagKey:  "tenant-1",
		cloudv1.ProjectTagKey: "project-1",
	}).String()
	assert.Equal(t, []string{expectedSelector}, recordingClient.clusterListSelectors)
	assert.False(t, plan.peerViewIncomplete)
	assert.Equal(t, peer.GKEComputeClassNameForTiKV(), plan.byNodePool[peerBaseNodePoolName].computeClass)
}

func TestNodePool_buildSharedGKEComputeClassPlanSkipsPeerLookupWhenScopeLabelsMissingWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
	}

	baseClient := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster).Build()
	recordingClient := &recordingClusterListClient{Client: baseClient}
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Cluster: cluster,
				Local:   recordingClient,
			},
		},
	}

	plan, err := np.buildSharedGKEComputeClassPlan(context.Background())
	assert.NoError(t, err)
	assert.True(t, plan.peerViewIncomplete)
	assert.Empty(t, plan.byNodePool)
	assert.Empty(t, recordingClient.clusterListSelectors)
}

func TestNodePool_buildSharedGKEComputeClassPlanRequiresScopeLabelsForLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8"}, []string{gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")})

	baseClient := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(cluster).Build()
	recordingClient := &recordingClusterListClient{Client: baseClient}
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Cluster: cluster,
				Local:   recordingClient,
			},
		},
	}

	_, err := np.buildSharedGKEComputeClassPlan(context.Background())
	assert.ErrorContains(t, err, "missing required GKE heterogeneous nodepool scope labels")
	assert.Empty(t, recordingClient.clusterListSelectors)
}

func TestNodePool_desiredNodePoolsIgnoresExistingOrdinaryBaseNodePoolWhenPeerViewIncomplete(t *testing.T) {
	tests := []struct {
		name             string
		existing         func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool
		wantComputeClass string
		wantErrMsg       string
	}{
		{
			name: "ordinary nodepool without shared labels has no compute class",
			existing: func(_ *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				return existingTestNodePool(nodePoolName, "n2-standard-8")
			},
		},
		{
			name: "shared nodepool without scope labels and metadata has no compute class",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				nodePool := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
				scopeLabels, _ := nodepoolutil.GKEKubernetesClusterScopeLabels(cluster)
				for key := range scopeLabels {
					delete(nodePool.Labels, key)
				}
				return nodePool
			},
		},
		{
			name: "valid shared nodepool is allowed",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				return existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "existing-priority")
			},
			wantComputeClass: "existing-priority",
		},
		{
			name: "foreign GCP cluster ref fails closed",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				nodePool := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "existing-priority")
				nodePool.Spec.ForProvider.Cluster = "projects/other/locations/us-central1/clusters/foreign"
				return nodePool
			},
			wantErrMsg: "belongs to GCP cluster",
		},
		{
			name: "shared scope mismatch fails closed",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				nodePool := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "existing-priority")
				nodePool.Labels[cloudv1.TenantTagKey] = "other-scope"
				return nodePool
			},
			wantErrMsg: "does not belong to cluster",
		},
		{
			name: "compute class metadata without shared ownership fails closed",
			existing: func(_ *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				nodePool := existingTestNodePool(nodePoolName, "n2-standard-8")
				nodepoolutil.SetGKEComputeClassForNodePool(nodePool, "polluted-priority")
				return nodePool
			},
			wantErrMsg: "has compute class metadata",
		},
		{
			name: "compute class metadata without shared scope fails closed",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				nodePool := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "polluted-priority")
				scopeLabels, _ := nodepoolutil.GKEKubernetesClusterScopeLabels(cluster)
				for key := range scopeLabels {
					delete(nodePool.Labels, key)
				}
				return nodePool
			},
			wantErrMsg: "does not belong to cluster",
		},
		{
			name: "applied-only compute class metadata fails closed",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				nodePool := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
				nodePool.Annotations[consts.AnnoKeyGKENodePoolComputeClassApplied] = "existing-priority"
				return nodePool
			},
			wantErrMsg: "has incomplete compute-class metadata",
		},
		{
			name: "machine type mismatch fails closed",
			existing: func(cluster *cloudv1.Cluster, nodePoolName string) *gkev1beta1.NodePool {
				return existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-16", "existing-priority")
			},
			wantErrMsg: "has machine type",
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := &cloudv1.Cluster{
				ObjectMeta: metav1.ObjectMeta{
					Name:      "cluster-a",
					Namespace: "ns",
					Labels: map[string]string{
						cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
					},
				},
				Spec: cloudv1.ClusterSpec{
					NetworkName: "shared-network",
					Cluster: cloudv1.ClusterTopology{
						TiKV: cloudv1.TiKV{
							Replicas:     3,
							InstanceType: "n2-standard-8",
						},
					},
				},
			}
			setTestGKEAPPBoxLabels(cluster)
			sibling := cluster.DeepCopy()
			sibling.Name = "cluster-b"
			sibling.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")
			sibling.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

			nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
			existing := tt.existing(cluster, nodePoolName)
			ctx := &task.Context{
				Cluster: cluster,
				Namer: namer.NewNamer().
					WithClusterName(cluster.Name).
					WithTiDBNamespace(cluster.PrefixedId()),
				Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(sibling, existing).Build(),
				EnvProvider: &testEnvProvider{},
				CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
					NodeVersion: "1.35",
				},
			}
			setContextZones(ctx, []string{"us-central1-a"})
			np := &NodePool{
				BaseTask: &task.BaseTask{
					Context: ctx,
				},
			}

			nps, err := np.desiredNodePools(context.Background())
			assert.NoError(t, err)
			assert.NotNil(t, nps)
			nodePoolsByName := testGKENodePoolsByName(nps)
			if assert.Contains(t, nodePoolsByName, nodePoolName) {
				assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
			}
		})
	}
}

func TestNodePool_desiredNodePoolsIgnoresSiblingSharedInstanceTypeConflictWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	siblingA := cluster.DeepCopy()
	siblingA.Name = "cluster-b"
	updateTestGKEComputeClassBinding(t, siblingA, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(siblingA, "n2-standard-8"), "shared-pool"})
	siblingB := cluster.DeepCopy()
	siblingB.Name = "cluster-c"
	siblingB.Spec.Cluster.TiKV.InstanceType = "n2-standard-16"
	updateTestGKEComputeClassBinding(t, siblingB, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-16", "n2-standard-128"}, []string{gkeheterogeneous.NodePoolName(siblingB, "n2-standard-16"), "shared-pool"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(siblingA, siblingB).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
	assert.NotContains(t, nodePoolsByName, "shared-pool")
}

func TestNodePool_desiredNodePoolsAllowsSiblingSharedComputeClassWithSameBindings(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	siblingA := cluster.DeepCopy()
	siblingA.Name = "cluster-b"
	updateTestGKEComputeClassBinding(t, siblingA, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority-a", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(siblingA, "n2-standard-8"), "shared-pool"})
	siblingB := cluster.DeepCopy()
	siblingB.Name = "cluster-c"
	updateTestGKEComputeClassBinding(t, siblingB, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority-b", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(siblingB, "n2-standard-8"), "shared-pool"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(siblingA, siblingB).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)
	assert.NotNil(t, nps)
}

func TestNodePool_desiredNodePoolsFallsBackToDefaultNodePoolWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")
	updateTestGKEComputeClassBinding(t, sibling, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{gkeheterogeneous.NodePoolName(sibling, "n2-standard-8"), "shared-pool"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(sibling).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	nodePoolsByName := testGKENodePoolsByName(nps)
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
	}
	assert.NotContains(t, nodePoolsByName, "shared-pool")
}

func TestNodePool_CreateSyncsActiveComputeClassWithNodePoolSpecLabel(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "tikv-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/" + nodePoolName
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ordinaryNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	ordinary := existingTestNodePool(ordinaryNodePoolName, "n2-standard-8")
	ordinary.Annotations = map[string]string{annoKeyResourceLabels: "tidb_cluster:" + cluster.Name}
	ordinary.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	ordinary.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(ordinaryNodePoolName)

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, extra, ordinary).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
	assert.Equal(t, computeClassName, got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
	assert.Equal(t, computeClassName, got.Spec.ForProvider.Config.Labels[consts.GKEComputeClassLabelKey])
}

func TestNodePool_CreateRequeuesWithoutTaskErrorAfterUnusedAPIManagedGKEHeterogeneousNodePoolDeleted(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	cluster.Spec.Cluster.TiKV = cloudv1.TiKV{}
	nodePoolName := "extra-pool"
	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	markTestGKEHeterogeneousNodePoolEverBound(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	existing.Status.AtProvider.InstanceGroupUrls = []string{testGKEInstanceGroupURL(nodePoolName + "-ig")}

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	gcpCli.EXPECT().
		GetNodePoolNodeLabels(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
		Return(map[string]string{}, nil)
	gcpCli.EXPECT().
		GetNodePoolNodeTaints(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
		Return([]*gcpcontainer.NodeTaint{
			testGKEHeterogeneousIsolationNodeTaint(),
		}, nil)
	gcpCli.EXPECT().
		GetInstanceGroupManagerManagedInstanceCount(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
		Return(int64(0), nil)
	gcpCli.EXPECT().
		GetInstanceGroupSize(gomock.Any(), "test-project", "us-central1-a", nodePoolName+"-ig").
		Return(int64(0), nil)

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())
	assert.NoError(t, result.Error())
	assert.True(t, k8s.RequeueOnly(result.RawError()))
	requeue := k8s.ShouldRequeue(result.RawError())
	assert.True(t, requeue.Requeue)
	assert.Equal(t, defaultNodePoolRequeue.RequeueAfter, requeue.RequeueAfter)

	got := &gkev1beta1.NodePool{}
	assert.Error(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got))
}

func TestNodePool_CreateRequeuesWithoutTaskErrorWhenInactiveAPIManagedGKEHeterogeneousRefreshFails(t *testing.T) {
	cluster := testGKESharedScopeCluster("cluster-a")
	cluster.Spec.Cluster.TiKV = cloudv1.TiKV{}
	nodePoolName := "prepared-pool"
	prepared := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(prepared)
	delete(prepared.Annotations, annoKeyResourceLabels)
	prepared.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	gcpCli.EXPECT().
		GetNodePoolResourceLabels(gomock.Any(), testGKENodePoolResourceName(nodePoolName)).
		Return(nil, fmt.Errorf("gcp resource labels unavailable"))

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(prepared).Build(),
		EnvProvider: &testEnvProvider{},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())
	assert.NoError(t, result.Error())
	assert.True(t, k8s.RequeueOnly(result.RawError()))
	requeue := k8s.ShouldRequeue(result.RawError())
	assert.True(t, requeue.Requeue)
	assert.Equal(t, gkeComputeClassNodeMetadataRequeueAfter, requeue.RequeueAfter)
}

func TestNodePool_CreateIgnoresSharedComputeClassWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(consts.AnnoKeyControlPlaneVersion, "1.35")

	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	computeClassName := updateTestGKEComputeClassBinding(t, sibling, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})
	existing := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   nodePoolName,
			Labels: sharedTestNodePoolLabels(cluster),
			Annotations: map[string]string{
				annoKeyResourceLabels:                 "tidb_cluster:cluster-a",
				consts.AnnoKeyGKENodePoolComputeClass: computeClassName,
			},
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: testGCPClusterRef(),
				Config: &gkev1beta1.NodeConfig{
					MachineType: ptr.To("n2-standard-8"),
					Labels: map[string]string{
						consts.GKEComputeClassLabelKey: computeClassName,
					},
				},
			},
		},
		Status: gkev1beta1.NodePoolStatus{
			AtProvider: gkev1beta1.NodePoolObservation{
				Status:   gkev1beta1.NodePoolStateRunning,
				SelfLink: "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/" + nodePoolName,
			},
		},
	}
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			sibling,
			existing,
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
}

func TestNodePool_CreatePreservesSharedComputeClassWhenTiKVPaused(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:      3,
					InstanceType:  "n2-standard-8",
					SuspendAction: &cloudv1.SuspendAction{SuspendCompute: true},
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, extra).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Equal(t, computeClassName, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_CreatePreservesSharedComputeClassWhenTiKVReplicasAreZero(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     0,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, extra).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Equal(t, computeClassName, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_CreateClearsStaleOrdinaryComputeClassWithoutReadingIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)

	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	existing := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name:   nodePoolName,
			Labels: sharedTestNodePoolLabels(cluster),
			Annotations: map[string]string{
				annoKeyResourceLabels:                 "tidb_cluster:cluster-a",
				consts.AnnoKeyGKENodePoolComputeClass: "stale-priority",
			},
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: testGCPClusterRef(),
				Config: &gkev1beta1.NodeConfig{
					MachineType: ptr.To("n2-standard-8"),
					Labels: map[string]string{
						consts.GKEComputeClassLabelKey: "stale-priority",
					},
				},
			},
		},
		Status: gkev1beta1.NodePoolStatus{
			AtProvider: gkev1beta1.NodePoolObservation{
				Status:   gkev1beta1.NodePoolStateRunning,
				SelfLink: "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/" + nodePoolName,
			},
		},
	}
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			existing,
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.NotContains(t, got.Annotations, consts.AnnoKeyGKENodePoolComputeClass)
	if got.Spec.ForProvider.Config != nil && got.Spec.ForProvider.Config.Labels != nil {
		assert.NotContains(t, got.Spec.ForProvider.Config.Labels, consts.GKEComputeClassLabelKey)
	}
}

func TestNodePool_cleanupStaleSharedGKEComputeClassNodePoolMetadataSkipsDifferentGCPCluster(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)

	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	existing.Spec.ForProvider.Cluster = "projects/test-project/locations/us-central1/clusters/other-gke"
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}

	err := np.cleanupStaleSharedGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got))
	assert.Equal(t, "stale-priority", nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_CreatePreservesSharedComputeClassWhenPeerViewIncomplete(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	sibling.UpdateAnnotation(cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, `not-json`)

	localNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	localNodePool := existingSharedComputeClassTestNodePool(cluster, localNodePoolName, "n2-standard-8", "")
	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			sibling,
			localNodePool,
			existing,
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got)
	assert.NoError(t, err)
	assert.Equal(t, "stale-priority", nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_CreateAppliesSharedComputeClassWhenPeerViewIncomplete(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})
	sibling := cluster.DeepCopy()
	sibling.Name = "cluster-b"
	delete(sibling.Annotations, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings)

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "wrong-priority")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ordinaryNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	ordinary := existingTestNodePool(ordinaryNodePoolName, "n2-standard-8")
	ordinary.Annotations = map[string]string{annoKeyResourceLabels: "tidb_cluster:" + cluster.Name}
	ordinary.Status.AtProvider.Status = gkev1beta1.NodePoolStateRunning
	ordinary.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(ordinaryNodePoolName)
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local: fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(
			sibling,
			existing,
			extra,
			ordinary,
		).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.Truef(t, result.IsDoneState(), "result err: %v", result.RawError())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Equal(t, computeClassName, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_ReconcileCurrentGKEComputeClassBindingNodePoolMetadataRejectsDifferentScope(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})
	foreign := cluster.DeepCopy()
	foreign.Spec.NetworkName = "foreign-network"
	setTestGKEAPPBoxLabelsWithProject(foreign, "project-2")
	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	existing := existingSharedComputeClassTestNodePool(foreign, "extra-pool", "n2-standard-64", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "does not belong to cluster")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_ReconcileCurrentGKEComputeClassBindingNodePoolMetadataRejectsWrongMachineType(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})
	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	existing := existingSharedComputeClassTestNodePool(cluster, "extra-pool", "n2-standard-32", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "want n2-standard-64")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_ReconcileCurrentGKEComputeClassBindingNodePoolMetadataRejectsDifferentGCPCluster(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	baseName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{baseName, "extra-pool"})
	base := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, baseName, "n2-standard-8")
	existing := existingSharedComputeClassTestNodePool(cluster, "extra-pool", "n2-standard-64", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Spec.ForProvider.Cluster = "projects/test-project/locations/us-central1/clusters/other-gke"

	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, baseName, computeClassName)
	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(base, existing).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
		gcpCli:   gcpCli,
	}

	err := np.reconcileCurrentGKEComputeClassBindingNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "belongs to GCP cluster")

	got := &gkev1beta1.NodePool{}
	assert.NoError(t, ctx.Local.Get(context.Background(), client.ObjectKey{Name: "extra-pool"}, got))
	assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_CreatePreservesSharedComputeClassForPausedTiDBExtraGroup(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster:     cloudv1.ClusterTopology{},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "group1",
					TiDB: cloudv1.TiDB{
						Replicas:      2,
						InstanceType:  "n2-standard-8",
						SuspendAction: &cloudv1.SuspendAction{SuspendCompute: true},
					},
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), "group1-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, extra).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.True(t, result.IsDoneState())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Equal(t, computeClassName, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_CreateIgnoresOrphanTiDBGroupIntentWhenReconcilingSharedComputeClass(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster:     cloudv1.ClusterTopology{},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("orphan"), "orphan-priority", []string{"n2-standard-8"}, []string{"shared-pool"})

	existing := existingSharedComputeClassTestNodePool(cluster, "shared-pool", "n2-standard-8", "stale-priority")
	existing.Status.AtProvider.SelfLink = "https://container.googleapis.com/v1/projects/test-project/locations/us-central1/clusters/gke-test/nodePools/shared-pool"
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.True(t, result.IsDoneState())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: "shared-pool"}, got)
	assert.NoError(t, err)
	assert.Equal(t, "stale-priority", nodepoolutil.GKEComputeClassNameForNodePool(got))
	assert.Equal(t, "stale-priority", got.Annotations[consts.AnnoKeyGKENodePoolComputeClass])
}

func TestExplicitGKEComputeClassNodePoolsFromAnnotationsForTiDBIsAnnotationDriven(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster-a",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
			Annotations: map[string]string{
				cloudv1.DefaultLoadBalancerIDAnnotationKey: "default-group",
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{InstanceType: "n2-standard-8"},
			},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "group1",
					TiDB: cloudv1.TiDB{InstanceType: "n2-standard-32"},
				},
			},
		},
	}
	np := testNodePoolForCluster(cluster)

	bindings, err := np.explicitGKEComputeClassNodePoolsFromAnnotations()
	assert.NoError(t, err)
	assert.Empty(t, bindings)

	cluster.UpdateAnnotation(cluster.TiDBGKEComputeClassBindingsAnnoKey("orphan"), `not-json`)
	bindings, err = np.explicitGKEComputeClassNodePoolsFromAnnotations()
	assert.NoError(t, err)
	assert.Empty(t, bindings)

	cluster.UpdateAnnotation(cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), `not-json`)
	_, err = np.explicitGKEComputeClassNodePoolsFromAnnotations()
	assert.ErrorContains(t, err, "decode gke compute class binding annotation")

	group1NodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-32")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), "group1-priority", []string{"n2-standard-32"}, []string{group1NodePoolName})
	bindings, err = np.explicitGKEComputeClassNodePoolsFromAnnotations()
	assert.NoError(t, err)
	if assert.Len(t, bindings, 2) {
		assert.Equal(t, group1NodePoolName, bindings[0].nodePoolName)
		assert.Equal(t, cloudv1.ComponentTiDB, bindings[0].componentType)
		assert.Equal(t, computeClassName, bindings[0].computeClass)
		assert.Equal(t, cloudv1.ComponentTiDB, bindings[1].componentType)
		assert.Equal(t, computeClassName, bindings[1].computeClass)
	}
}

func TestNodePool_CreateRetainsSharedComputeClassForZeroReplicaTiDBExtraGroupIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster:     cloudv1.ClusterTopology{},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "group1",
					TiDB: cloudv1.TiDB{
						Replicas:     0,
						InstanceType: "n2-standard-8",
					},
				},
			},
		},
	}
	cluster.EnableFeature(consts.ClusterFeatureUseTiDBOperatorV2)
	setTestGKEAPPBoxLabels(cluster)
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	computeClassName := updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), "group1-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", computeClassName)
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	extraNodePoolName := gkeheterogeneous.NodePoolName(cluster, testGKEExtraInstanceType("n2-standard-8"))
	extra := syncedAPIManagedGKEHeterogeneousTestNodePool(cluster, extraNodePoolName, testGKEExtraInstanceType("n2-standard-8"))
	ctrl := gomock.NewController(t)
	gcpCli := gcp.NewMockClient(ctrl)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, nodePoolName, computeClassName)
	expectGKENodePoolComputeClassNodeMetadataSync(t, gcpCli, extraNodePoolName, computeClassName)
	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, extra).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
		gcpCli: gcpCli,
	}

	result := np.Create(context.Background())
	assert.True(t, result.IsDoneState())

	got := &gkev1beta1.NodePool{}
	err := ctx.Local.Get(context.Background(), client.ObjectKey{Name: nodePoolName}, got)
	assert.NoError(t, err)
	assert.Equal(t, computeClassName, nodepoolutil.GKEComputeClassNameForNodePool(got))
}

func TestNodePool_desiredNodePoolsIncludesActiveTiDBExtraGroupWhenDefaultGroupPaused(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:      1,
					InstanceType:  "n2-standard-4",
					SuspendAction: &cloudv1.SuspendAction{SuspendCompute: true},
				},
			},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "group1",
					TiDB: cloudv1.TiDB{
						Replicas:     2,
						InstanceType: "n2-standard-8",
					},
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	heterogeneousNodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), "group1-priority", []string{"n2-standard-8"}, []string{heterogeneousNodePoolName})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	if assert.Contains(t, nodePoolsByName, nodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[nodePoolName]))
		assert.Equal(t, "n2-standard-8", *nodePoolsByName[nodePoolName].Spec.ForProvider.Config.MachineType)
	}
	assert.NotContains(t, nodePoolsByName, heterogeneousNodePoolName)
}

func TestNodePool_desiredNodePoolsDoesNotUseComputeClassForDefaultTiDBWhenOnlyExtraGroupEnabled(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "n2-standard-4",
				},
			},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "group1",
					TiDB: cloudv1.TiDB{
						Replicas:     2,
						InstanceType: "n2-standard-8",
					},
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	heterogeneousExtraGroupNodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), "group1-priority", []string{"n2-standard-8", "n2-standard-64"}, []string{heterogeneousExtraGroupNodePoolName, "group1-extra"})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	nodePoolsByName := testGKENodePoolsByName(nps)
	defaultNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-4", "1.35")
	if assert.Contains(t, nodePoolsByName, defaultNodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[defaultNodePoolName]))
		assert.Equal(t, "n2-standard-4", *nodePoolsByName[defaultNodePoolName].Spec.ForProvider.Config.MachineType)
	}
	extraGroupNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	if assert.Contains(t, nodePoolsByName, extraGroupNodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[extraGroupNodePoolName]))
		assert.Equal(t, "n2-standard-8", *nodePoolsByName[extraGroupNodePoolName].Spec.ForProvider.Config.MachineType)
	}
	assert.NotContains(t, nodePoolsByName, heterogeneousExtraGroupNodePoolName)
	assert.NotContains(t, nodePoolsByName, "group1-extra")
}

func TestNodePool_desiredNodePoolsAllowsSharedNodePoolAcrossComponents(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "cluster-a",
			Namespace: "ns",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			NetworkName: "shared-network",
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     3,
					InstanceType: "n2-standard-8",
				},
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	setTestGKEAPPBoxLabels(cluster)
	sharedNodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{sharedNodePoolName})
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiDBBindings, "shared-priority", []string{"n2-standard-8"}, []string{sharedNodePoolName})

	ctx := &task.Context{
		Cluster: cluster,
		Namer: namer.NewNamer().
			WithClusterName(cluster.Name).
			WithTiDBNamespace(cluster.PrefixedId()),
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existingTestNodePool(sharedNodePoolName, "n2-standard-8")).Build(),
		EnvProvider: &testEnvProvider{},
		CommonK8sClusterStatus: &cloudv1.CommonK8sClusterStatus{
			NodeVersion: "1.35",
		},
	}
	setContextZones(ctx, []string{"us-central1-a"})
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: ctx,
		},
	}

	nps, err := np.desiredNodePools(context.Background())
	assert.NoError(t, err)

	defaultNodePoolName := nodepoolutil.GetGKENodePoolName(cluster, "n2-standard-8", "1.35")
	nodePoolsByName := testGKENodePoolsByName(nps)
	assert.Len(t, nodePoolsByName, 1)
	if assert.Contains(t, nodePoolsByName, defaultNodePoolName) {
		assert.Empty(t, nodepoolutil.GKEComputeClassNameForNodePool(nodePoolsByName[defaultNodePoolName]))
	}
	assert.NotContains(t, nodePoolsByName, sharedNodePoolName)
}

func TestNonInstanceTypeScalingComponentsAllowlistedTiKV(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					Replicas:     2,
					InstanceType: "n2-standard-8",
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			DB: &cloudv1.DBStatus{
				TiKV: &cloudv1.TiKVStatus{
					Members: []cloudv1.StoreMember{{InstanceType: "n2-standard-64"}},
				},
			},
		},
	}

	assert.NotContains(t, NonInstanceTypeScalingComponents(cluster), cloudv1.ComponentTiKV)

	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, cloudv1.DefaultGKEComputeClassForTiKV, []string{"n2-standard-8", "n2-standard-64"}, testGKEComputeClassNodePoolNames(cluster, "n2-standard-8", "n2-standard-64"))
	assert.Contains(t, NonInstanceTypeScalingComponents(cluster), cloudv1.ComponentTiKV)
}

func TestNonInstanceTypeScalingComponentsAllowlistedTiDB(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "n2-standard-8",
				},
			},
		},
		Status: cloudv1.ClusterStatus{
			DB: &cloudv1.DBStatus{
				TiDB: &cloudv1.TiDBStatus{
					Members: []cloudv1.TiDBMember{{InstanceType: "n2-standard-64"}},
				},
			},
		},
	}

	assert.NotContains(t, NonInstanceTypeScalingComponents(cluster), cloudv1.ComponentTiDB)

	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiDBBindings, cloudv1.DefaultGKEComputeClassForTiDB, []string{"n2-standard-8", "n2-standard-64"}, testGKEComputeClassNodePoolNames(cluster, "n2-standard-8", "n2-standard-64"))
	assert.Contains(t, NonInstanceTypeScalingComponents(cluster), cloudv1.ComponentTiDB)
}

func TestTiDBGroupNonInstanceTypeScalingComponentsAllowlistedTiDB(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "group1",
					TiDB: cloudv1.TiDB{
						Replicas:     2,
						InstanceType: "n2-standard-8",
					},
				},
			},
		},
	}
	extraGroup := &cloudv1.ExtraGroupCluster{
		Name: "group1",
		Spec: &cloudv1.ExtraGroupSpec{
			Name: "group1",
			TiDB: cloudv1.TiDB{
				Replicas:     2,
				InstanceType: "n2-standard-8",
			},
		},
		Status: &cloudv1.ExtraGroupStatus{
			TiDB: &cloudv1.TiDBStatus{
				Members: []cloudv1.TiDBMember{{InstanceType: "n2-standard-64"}},
			},
		},
	}
	task := &TiDBGroup{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Cluster: cluster,
			},
		},
	}

	assert.NotContains(t, task.NonInstanceTypeScalingComponents(extraGroup), cloudv1.ComponentTiDB)

	updateTestGKEComputeClassBinding(t, cluster, cluster.TiDBGKEComputeClassBindingsAnnoKey("group1"), cloudv1.DefaultGKEComputeClassNameForTiDBGroup(cluster.Name, "group1"), []string{"n2-standard-8", "n2-standard-64"}, testGKEComputeClassNodePoolNames(cluster, "n2-standard-8", "n2-standard-64"))
	assert.Contains(t, task.NonInstanceTypeScalingComponents(extraGroup), cloudv1.ComponentTiDB)
}

func TestValidateGKEComputeClassNodePoolPatchTargetRequiresScopeLabels(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster-a",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
	}
	nodePool := &gkev1beta1.NodePool{
		ObjectMeta: metav1.ObjectMeta{
			Name: "shared-pool",
			Labels: map[string]string{
				namer.ComponentSelectorKey: shared.ComponentShared,
				namer.ClusterSelectorKey:   shared.ClustersShared,
			},
		},
		Spec: gkev1beta1.NodePoolSpec{
			ForProvider: gkev1beta1.NodePoolParameters{
				Cluster: testGCPClusterRef(),
				Config: &gkev1beta1.NodeConfig{
					MachineType: ptr.To("n2-standard-8"),
				},
			},
		},
	}
	np := testNodePoolForCluster(cluster)

	err := np.validateGKEComputeClassNodePoolPatchTarget(nodePool, "shared-pool", "n2-standard-8", testGCPClusterRef())
	assert.ErrorContains(t, err, "missing required GKE heterogeneous nodepool scope labels")

	err = np.validateGKEComputeClassNodePoolPatchTarget(nodePool, "other-pool", "n2-standard-8", testGCPClusterRef())
	assert.ErrorContains(t, err, "does not match expected nodepool")
}

func TestNodePool_ReconcileGKEComputeClassNodePoolMetadataSkipsCleanupWhenScopeLabelsMissingWithoutLocalIntent(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster-a",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
	}
	ctx := &task.Context{
		Cluster: cluster,
		Local:   fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build(),
	}
	np := &NodePool{BaseTask: &task.BaseTask{Context: ctx}}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.NoError(t, err)
	assert.False(t, deleted)
}

func TestNodePool_ReconcileGKEComputeClassNodePoolMetadataRequiresScopeLabelsWithLocalBinding(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster-a",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiKV: cloudv1.TiKV{
					InstanceType: "n2-standard-8",
				},
			},
		},
	}
	nodePoolName := gkeheterogeneous.NodePoolName(cluster, "n2-standard-8")
	existing := existingSharedComputeClassTestNodePool(cluster, nodePoolName, "n2-standard-8", "")
	markTestAPIManagedGKEHeterogeneousNodePool(existing)
	existing.Status.AtProvider.SelfLink = testGKENodePoolSelfLink(nodePoolName)
	stale := existingSharedComputeClassTestNodePool(cluster, "stale-pool", "n2-standard-64", "stale-priority")
	stale.Status.AtProvider.SelfLink = testGKENodePoolSelfLink("stale-pool")
	updateTestGKEComputeClassBinding(t, cluster, cloudv1.AnnoKeyGKEComputeClassForTiKVBindings, "shared-priority", []string{"n2-standard-8"}, []string{nodePoolName})

	ctx := &task.Context{
		Cluster:     cluster,
		Local:       fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(existing, stale).Build(),
		EnvProvider: &testEnvProvider{},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{Context: ctx},
	}

	deleted, err := np.reconcileGKEComputeClassNodePoolMetadata(context.Background())
	assert.ErrorContains(t, err, "missing required GKE heterogeneous nodepool scope labels")
	assert.False(t, deleted)
}

func Test_buildOpDeployment(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Annotations: map[string]string{
				consts.DefaultLoadBalancerIDAnnotationKey: "default-group",
			},
			Name: "cluster1",
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "c5g.large",
				},
				TiKV: cloudv1.TiKV{
					Replicas:     1,
					InstanceType: "c5.2xlarge",
				},
			},
			ExtraGroups: []cloudv1.ExtraGroupSpec{
				{
					Name: "extra-group1",
					TiDB: cloudv1.TiDB{
						Replicas:     2,
						InstanceType: "c5.large",
					},
				},
			},
		},
		TypeMeta: metav1.TypeMeta{},
	}

	deployment := &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Namespace: "tidbcluster1",
			Labels: map[string]string{
				consts.InstanceLabelKey:  "default-group",
				consts.ComponentLabelKey: consts.OverProvisioningComponentLabelValue,
				// Set to "tidb" by default, will be overridden in deploymentModifier for tikv case
				consts.OverProvisioningComponentLabelKey: string(cloudv1.ComponentTiDB),
				// Set to "default-group" by default, will be overridden for extra group
				consts.LabelKeyTiDBGroupId:             "default-group",
				consts.OverProvisioningUsedForLabelKey: "tidbcluster",
			},
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: ptr.To(int32(2)),
			Selector: &metav1.LabelSelector{
				MatchLabels: map[string]string{
					consts.InstanceLabelKey:                  "default-group",
					consts.ComponentLabelKey:                 consts.OverProvisioningComponentLabelValue,
					consts.OverProvisioningComponentLabelKey: string(cloudv1.ComponentTiDB),
					consts.LabelKeyTiDBGroupId:               "default-group",
					consts.OverProvisioningUsedForLabelKey:   "tidbcluster",
				},
			},
			Template: corev1.PodTemplateSpec{
				ObjectMeta: metav1.ObjectMeta{
					Labels: map[string]string{
						consts.InstanceLabelKey:                  "default-group",
						consts.ComponentLabelKey:                 consts.OverProvisioningComponentLabelValue,
						consts.OverProvisioningComponentLabelKey: string(cloudv1.ComponentTiDB),
						consts.LabelKeyTiDBGroupId:               "default-group",
						consts.OverProvisioningUsedForLabelKey:   "tidbcluster",
					},
				},
				Spec: corev1.PodSpec{
					Containers: []corev1.Container{
						{
							Name:  overProvisioningContainerName,
							Image: "gcr.io/pingcap-public/third-party/pause:3.9",
							Resources: corev1.ResourceRequirements{
								Requests: corev1.ResourceList{
									corev1.ResourceCPU: resource.MustParse("5m"),
								},
							},
						},
					},
					NodeSelector: map[string]string{
						"cloud.pingcap.com/node-group-type": "data-plane",
						"foo":                               "bar",
					},
					TopologySpreadConstraints: []corev1.TopologySpreadConstraint{
						{
							MaxSkew:           1,
							TopologyKey:       "topology.kubernetes.io/zone",
							WhenUnsatisfiable: corev1.DoNotSchedule,
							LabelSelector: &metav1.LabelSelector{
								MatchLabels: map[string]string{
									consts.InstanceLabelKey:                  "default-group",
									consts.ComponentLabelKey:                 consts.OverProvisioningComponentLabelValue,
									consts.OverProvisioningComponentLabelKey: string(cloudv1.ComponentTiDB),
									consts.LabelKeyTiDBGroupId:               "default-group",
									consts.OverProvisioningUsedForLabelKey:   "tidbcluster",
								},
							},
							MinDomains: ptr.To(int32(3)),
						},
					},
					Affinity: &corev1.Affinity{
						PodAntiAffinity: &corev1.PodAntiAffinity{
							RequiredDuringSchedulingIgnoredDuringExecution: []corev1.PodAffinityTerm{
								{
									TopologyKey: "kubernetes.io/hostname",
									LabelSelector: &metav1.LabelSelector{
										MatchLabels: map[string]string{
											consts.InstanceLabelKey:  "default-group",
											consts.ComponentLabelKey: consts.OverProvisioningComponentLabelValue,
										},
									},
									NamespaceSelector: &metav1.LabelSelector{},
								},
								{
									TopologyKey: "kubernetes.io/hostname",
									LabelSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{{
											Key:      consts.ComponentLabelKey,
											Operator: metav1.LabelSelectorOpIn,
											Values: []string{
												string(cloudv1.ComponentPD),
												string(cloudv1.ComponentTiKV),
												string(cloudv1.ComponentTiDB),
												string(cloudv1.ComponentTiFlash),
												string(cloudv1.ComponentTiProxy),
											},
										}},
									},
									NamespaceSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{{
											Key:      corev1.LabelMetadataName,
											Operator: metav1.LabelSelectorOpNotIn,
											Values:   []string{"tidbcluster1"},
										}},
									},
								},
								{
									TopologyKey: "kubernetes.io/hostname",
									LabelSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{{
											Key:      consts.ComponentLabelKey,
											Operator: metav1.LabelSelectorOpIn,
											Values:   []string{"pd", "tikv", "tiflash", "tiproxy"},
										}},
									},
								},
								{
									// anti-affinity with other tidb node groups
									LabelSelector: &metav1.LabelSelector{
										MatchExpressions: []metav1.LabelSelectorRequirement{
											{
												Key:      consts.InstanceLabelKey,
												Operator: metav1.LabelSelectorOpNotIn,
												Values:   []string{"default-group"},
											},
										},
									},
									TopologyKey: corev1.LabelHostname,
								},
							},
						},
					},
					Tolerations: []corev1.Toleration{{
						Key:      "dedicated",
						Operator: corev1.TolerationOpEqual,
						Value:    "tidb-shared",
						Effect:   corev1.TaintEffectNoSchedule,
					}},
					TerminationGracePeriodSeconds: ptr.To(int64(0)),
				},
			},
		},
	}
	tests := map[string]struct {
		groupName          string
		compType           cloudv1.ComponentType
		paramsModifier     func(*buildOpDeploymentParams)
		deploymentModifier func(*appsv1.Deployment)
	}{
		"tidb-default-group": {
			groupName:      "default-group",
			compType:       cloudv1.ComponentTiDB,
			paramsModifier: func(p *buildOpDeploymentParams) {},
			deploymentModifier: func(d *appsv1.Deployment) {
				d.Name = "op-tidb-default"
				// Set all relevant label maps for tidb-default-group
				d.ObjectMeta.Labels[consts.LabelKeyTiDBGroupId] = "default-group"
				d.ObjectMeta.Labels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiDB)
				if d.Spec.Selector != nil {
					d.Spec.Selector.MatchLabels[consts.LabelKeyTiDBGroupId] = "default-group"
					d.Spec.Selector.MatchLabels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiDB)
				}
				if d.Spec.Template.ObjectMeta.Labels != nil {
					d.Spec.Template.ObjectMeta.Labels[consts.LabelKeyTiDBGroupId] = "default-group"
					d.Spec.Template.ObjectMeta.Labels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiDB)
				}
				d.Spec.Template.Spec.NodeSelector = map[string]string{
					"cloud.pingcap.com/node-group-type": "data-plane",
					"foo":                               "bar",
				}
			},
		},
		"tidb-extra-group": {
			groupName: "extra-group1",
			compType:  cloudv1.ComponentTiDB,
			paramsModifier: func(p *buildOpDeploymentParams) {
				p.OpName = "op-tidb-extra"
				p.OpReplicas = 3
				p.NodeSelector = map[string]string{"zone": "us-central1-a"}
			},
			deploymentModifier: func(d *appsv1.Deployment) {
				d.Name = "op-tidb-extra"
				d.ObjectMeta.Labels[consts.LabelKeyTiDBGroupId] = "extra-group1"
				d.ObjectMeta.Labels[consts.InstanceLabelKey] = "extra-group1"
				d.ObjectMeta.Labels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiDB)
				d.Spec.Replicas = ptr.To(int32(3))
				if d.Spec.Selector != nil {
					d.Spec.Selector.MatchLabels[consts.LabelKeyTiDBGroupId] = "extra-group1"
					d.Spec.Selector.MatchLabels[consts.InstanceLabelKey] = "extra-group1"
					d.Spec.Selector.MatchLabels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiDB)
				}
				if d.Spec.Template.ObjectMeta.Labels != nil {
					d.Spec.Template.ObjectMeta.Labels[consts.LabelKeyTiDBGroupId] = "extra-group1"
					d.Spec.Template.ObjectMeta.Labels[consts.InstanceLabelKey] = "extra-group1"
					d.Spec.Template.ObjectMeta.Labels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiDB)
				}
				d.Spec.Template.Spec.NodeSelector = map[string]string{
					"cloud.pingcap.com/node-group-type": "data-plane",
					"zone":                              "us-central1-a",
				}
				d.Spec.Template.Spec.TopologySpreadConstraints[0].LabelSelector.MatchLabels[consts.LabelKeyTiDBGroupId] = "extra-group1"
				d.Spec.Template.Spec.TopologySpreadConstraints[0].LabelSelector.MatchLabels[consts.InstanceLabelKey] = "extra-group1"
				d.Spec.Template.Spec.Affinity.PodAntiAffinity.RequiredDuringSchedulingIgnoredDuringExecution[0].LabelSelector.MatchLabels[consts.InstanceLabelKey] = "extra-group1"
				d.Spec.Template.Spec.Affinity.PodAntiAffinity.RequiredDuringSchedulingIgnoredDuringExecution[3].LabelSelector.MatchExpressions[0].Values = []string{"extra-group1"}
			},
		},
		"tikv-default-group": {
			groupName: "default-group",
			compType:  cloudv1.ComponentTiKV,
			paramsModifier: func(p *buildOpDeploymentParams) {
				p.OpName = "op-tikv-default"
				p.OpReplicas = 1
				p.NodeSelector = map[string]string{"zone": "us-central1-a"}
			},
			deploymentModifier: func(d *appsv1.Deployment) {
				d.Name = "op-tikv-default"
				d.ObjectMeta.Labels[consts.LabelKeyTiDBGroupId] = "default-group"
				d.ObjectMeta.Labels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiKV)
				d.Spec.Replicas = ptr.To(int32(1))
				if d.Spec.Selector != nil {
					d.Spec.Selector.MatchLabels[consts.LabelKeyTiDBGroupId] = "default-group"
					d.Spec.Selector.MatchLabels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiKV)
				}
				if d.Spec.Template.ObjectMeta.Labels != nil {
					d.Spec.Template.ObjectMeta.Labels[consts.LabelKeyTiDBGroupId] = "default-group"
					d.Spec.Template.ObjectMeta.Labels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiKV)
				}
				d.Spec.Template.Spec.NodeSelector = map[string]string{
					"cloud.pingcap.com/node-group-type": "data-plane",
					"zone":                              "us-central1-a",
				}
				d.Spec.Template.Spec.TopologySpreadConstraints[0].LabelSelector.MatchLabels[consts.OverProvisioningComponentLabelKey] = string(cloudv1.ComponentTiKV)
				d.Spec.Template.Spec.Affinity.PodAntiAffinity.RequiredDuringSchedulingIgnoredDuringExecution[2].LabelSelector.MatchExpressions = []metav1.LabelSelectorRequirement{{
					Key:      consts.ComponentLabelKey,
					Operator: metav1.LabelSelectorOpIn,
					Values:   []string{"pd", "tidb", "tiflash", "tiproxy"}, // exclude other than tikv
				}}
			},
		},
	}

	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			n := namer.NewNamer().
				WithClusterName(cluster.Name).
				WithTiDBNamespace(cluster.PrefixedId()).
				WithTiDBName(tt.groupName) // use group name as tidb name for simplicity in test
			params := &buildOpDeploymentParams{
				OpName:       "op-tidb-default",
				OpUsedFor:    "tidbcluster",
				OpReplicas:   2,
				Comp:         lo.Must1(cluster.GetComponentSpecInGroup(tt.groupName, tt.compType)),
				NodeSelector: map[string]string{"foo": "bar"},
				Namer:        n,
				Cluster:      cluster,
				Zones:        []string{"us-central1-a", "us-central1-b", "us-central1-c"},
			}
			if tt.paramsModifier != nil {
				tt.paramsModifier(params)
			}
			actual := buildOpDeployment(params)

			expected := deployment.DeepCopy()
			if tt.deploymentModifier != nil {
				tt.deploymentModifier(expected)
			}
			diff := cmp.Diff(expected, actual, cmpopts.EquateEmpty())
			assert.Emptyf(t, diff, fmt.Sprintf("%s ensure diff empty", name))
		})
	}
}

func TestNodePool_createStandbyNodeOPDeploymentsChecksInstanceTypeBeforeStandbyNodes(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Name: "cluster1",
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				PD: cloudv1.PD{
					InstanceType: "n2-standard-4",
				},
				TiKV: cloudv1.TiKV{
					InstanceType: "n2-standard-8",
				},
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "missing-instance-type",
				},
			},
		},
	}
	np := &NodePool{
		BaseTask: &task.BaseTask{
			Context: &task.Context{
				Cluster: cluster,
				Provider: fakeCloudConfigProvider{
					instanceTypes: map[string]cloudv1.InstanceType{
						"n2-standard-4": {Name: "n2-standard-4"},
						"n2-standard-8": {Name: "n2-standard-8"},
					},
				},
				TCGetter: tidbcluster.NewV1Getter(&opv1alpha1.TidbCluster{}),
			},
		},
	}
	remoteCli := fake.NewClientBuilder().WithScheme(scheme2.Scheme).Build()

	err := np.createStandbyNodeOPDeployments(context.Background(), remoteCli, []string{"us-central1-a"})
	require.Error(t, err)
	assert.Contains(t, err.Error(), "failed query arch of instance type(missing-instance-type)")
}

func Test_buildOpDeploymentAddsArchTolerationOnlyWhenArchProvided(t *testing.T) {
	cluster := &cloudv1.Cluster{
		ObjectMeta: metav1.ObjectMeta{
			Annotations: map[string]string{
				consts.DefaultLoadBalancerIDAnnotationKey: "default-group",
			},
			Labels: map[string]string{
				cloudv1.ProviderTypeLabel: cloudv1.ProviderTypeGKE,
			},
			Name: "cluster1",
		},
		Spec: cloudv1.ClusterSpec{
			Cluster: cloudv1.ClusterTopology{
				TiDB: cloudv1.TiDB{
					Replicas:     2,
					InstanceType: "c5g.large",
				},
			},
		},
	}
	comp := lo.Must1(cluster.GetComponentSpecInGroup("default-group", cloudv1.ComponentTiDB))
	n := namer.NewNamer().
		WithClusterName(cluster.Name).
		WithTiDBNamespace(cluster.PrefixedId()).
		WithTiDBName("default-group")

	tests := map[string]struct {
		arch                    string
		wantArchTolerationFound bool
	}{
		"empty arch keeps origin main tolerations": {
			arch:                    "",
			wantArchTolerationFound: false,
		},
		"arm arch adds arch toleration": {
			arch:                    consts.ARM64ArchName,
			wantArchTolerationFound: true,
		},
	}

	for name, tt := range tests {
		t.Run(name, func(t *testing.T) {
			actual := buildOpDeployment(&buildOpDeploymentParams{
				OpName:       "op-tidb-default",
				OpUsedFor:    consts.OverProvisioningUsedForStandbyNode,
				OpReplicas:   2,
				Comp:         comp,
				NodeSelector: map[string]string{"foo": "bar"},
				Namer:        n,
				Cluster:      cluster,
				Zones:        []string{"us-central1-a", "us-central1-b", "us-central1-c"},
				Arch:         tt.arch,
			})

			archToleration := *utils.TolerationWithArch(consts.ARM64ArchName)
			assert.Equal(t, tt.wantArchTolerationFound, lo.Contains(actual.Spec.Template.Spec.Tolerations, archToleration))
		})
	}
}

func TestNodePool_limitNodePoolsForZoneBootstrap(t *testing.T) {
	const (
		instanceType = "n2-standard-8"
		k8sVersion   = "1.35"
	)
	zones := []string{"us-central1-a", "us-central1-b", "us-central1-c"}

	tests := []struct {
		name             string
		sizes            map[string]int64
		missingURL       string
		consumers        func() []client.Object
		wantLimited      bool
		wantMaxNodeCount int64
	}{
		{
			name:  "all zones already have capacity",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 1, "us-central1-c": 1},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolTCConsumer(instanceType, k8sVersion, 3, false)}
			},
			wantMaxNodeCount: maxNodeCountPerZone,
		},
		{
			name:  "consumer without minDomains needs bootstrap",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolTCConsumer(instanceType, k8sVersion, 3, false)}
			},
			wantLimited:      true,
			wantMaxNodeCount: 1,
		},
		{
			name:  "effective minDomains keeps normal max",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolTCConsumer(instanceType, k8sVersion, 3, true)}
			},
			wantMaxNodeCount: maxNodeCountPerZone,
		},
		{
			name:  "inactive consumer does not need bootstrap",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolTCConsumer(instanceType, k8sVersion, 0, false)}
			},
			wantMaxNodeCount: maxNodeCountPerZone,
		},
		{
			name:  "consumer for another nodepool does not need bootstrap",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolTCConsumer("n2-standard-16", k8sVersion, 3, false)}
			},
			wantMaxNodeCount: maxNodeCountPerZone,
		},
		{
			name:  "over-provisioning consumer without minDomains needs bootstrap",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolDeploymentConsumer(instanceType, k8sVersion, false)}
			},
			wantLimited:      true,
			wantMaxNodeCount: 1,
		},
		{
			name:  "over-provisioning consumer with minDomains keeps normal max",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				return []client.Object{testNodePoolDeploymentConsumer(instanceType, k8sVersion, true)}
			},
			wantMaxNodeCount: maxNodeCountPerZone,
		},
		{
			name:  "over-provisioning ScheduleAnyway constraint needs bootstrap",
			sizes: map[string]int64{"us-central1-a": 1, "us-central1-b": 0},
			consumers: func() []client.Object {
				consumer := testNodePoolDeploymentConsumer(instanceType, k8sVersion, true)
				consumer.Spec.Template.Spec.TopologySpreadConstraints[0].WhenUnsatisfiable = corev1.ScheduleAnyway
				return []client.Object{consumer}
			},
			wantLimited:      true,
			wantMaxNodeCount: 1,
		},
		{
			name:       "missing instance group URL counts as a missing zone",
			sizes:      map[string]int64{"us-central1-a": 1, "us-central1-b": 1},
			missingURL: "us-central1-c",
			consumers: func() []client.Object {
				return []client.Object{testNodePoolTCConsumer(instanceType, k8sVersion, 3, false)}
			},
			wantLimited:      true,
			wantMaxNodeCount: 1,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			cluster := testGKEV1MinDomainsCluster()
			cluster.Annotations[consts.AnnoKeyControlPlaneVersion] = k8sVersion
			nodePoolName := nodepoolutil.GetGKENodePoolName(cluster, instanceType, k8sVersion)
			desired := existingTestNodePool(nodePoolName, instanceType)
			desired.Spec.ForProvider.Locations = zones
			desired.Spec.ForProvider.Autoscaling = &gkev1beta1.NodePoolAutoscaling{
				Enabled:      ptr.To(true),
				MinNodeCount: ptr.To(int64(0)),
				MaxNodeCount: ptr.To(maxNodeCountPerZone),
			}

			current := desired.DeepCopy()
			for _, zone := range zones {
				if zone != tt.missingURL {
					current.Status.AtProvider.InstanceGroupUrls = append(current.Status.AtProvider.InstanceGroupUrls,
						fmt.Sprintf("https://www.googleapis.com/compute/v1/projects/test-project/zones/%s/instanceGroupManagers/%s-%s", zone, nodePoolName, zone),
					)
				}
			}

			ctrl := gomock.NewController(t)
			gcpCli := gcp.NewMockClient(ctrl)
			for _, zone := range zones {
				if zone == tt.missingURL {
					break
				}
				size := tt.sizes[zone]
				gcpCli.EXPECT().
					GetInstanceGroupSize(gomock.Any(), "test-project", zone, nodePoolName+"-"+zone).
					Return(size, nil)
				if size < 1 {
					break
				}
			}

			nodePoolTask := &NodePool{
				BaseTask: &task.BaseTask{Context: &task.Context{
					Cluster: cluster,
					Local:   fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(current).Build(),
				}},
				gcpCli: gcpCli,
			}
			remoteCli := fake.NewClientBuilder().WithScheme(scheme2.Scheme).WithObjects(tt.consumers()...).Build()

			limited, err := nodePoolTask.limitNodePoolsForZoneBootstrap(context.Background(), remoteCli, []*gkev1beta1.NodePool{desired})

			require.NoError(t, err)
			assert.Equal(t, tt.wantLimited, limited)
			require.NotNil(t, desired.Spec.ForProvider.Autoscaling.MaxNodeCount)
			assert.Equal(t, tt.wantMaxNodeCount, *desired.Spec.ForProvider.Autoscaling.MaxNodeCount)
		})
	}
}

func testNodePoolTCConsumer(instanceType, k8sVersion string, replicas int32, minDomains bool) *opv1alpha1.TidbCluster {
	tc := &opv1alpha1.TidbCluster{
		ObjectMeta: metav1.ObjectMeta{Name: "db", Namespace: "cluster"},
		Spec: opv1alpha1.TidbClusterSpec{TiKV: &opv1alpha1.TiKVSpec{
			Replicas: replicas,
			ComponentSpec: opv1alpha1.ComponentSpec{NodeSelector: map[string]string{
				consts.InstanceTypeLabelKey: instanceType,
				consts.K8sVersionLabelKey:   k8sVersion,
			}},
		}},
	}
	if minDomains {
		gkemindomains.EnsureMinDomainsConstraints(tc, cloudv1.ComponentTiKV)
	}
	return tc
}

func testNodePoolDeploymentConsumer(instanceType, k8sVersion string, minDomains bool) *appsv1.Deployment {
	constraint := corev1.TopologySpreadConstraint{
		TopologyKey:       corev1.LabelTopologyZone,
		MaxSkew:           1,
		WhenUnsatisfiable: corev1.DoNotSchedule,
	}
	if minDomains {
		constraint.MinDomains = ptr.To(int32(gkemindomains.RequiredZoneCount))
	}
	return &appsv1.Deployment{
		ObjectMeta: metav1.ObjectMeta{
			Name:      "over-provisioning",
			Namespace: "cluster",
			Labels: map[string]string{
				consts.ComponentLabelKey: consts.OverProvisioningComponentLabelValue,
			},
		},
		Spec: appsv1.DeploymentSpec{
			Replicas: ptr.To(int32(3)),
			Template: corev1.PodTemplateSpec{Spec: corev1.PodSpec{
				NodeSelector: map[string]string{
					consts.InstanceTypeLabelKey: instanceType,
					consts.K8sVersionLabelKey:   k8sVersion,
				},
				TopologySpreadConstraints: []corev1.TopologySpreadConstraint{constraint},
			}},
		},
	}
}