package server

import (
	"context"
	"errors"
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	specpkg "github.com/tidbcloud/cluster-service-ng/pkg/autoscale/spec"
	dataplanemocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplane/mocks"
	kubemocks "github.com/tidbcloud/cluster-service-ng/pkg/externalmocks/mocks"
	schemav1 "github.com/tidbcloud/cluster-service-ng/pkg/storage/etcd/schema/v1"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/utils"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	autoscalingv2 "k8s.io/api/autoscaling/v2"
	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
	"k8s.io/apimachinery/pkg/labels"
	"sigs.k8s.io/controller-runtime/pkg/client"
)

// ---------------------------------------------------------------------------
// Pure-function tests for the HPA → Spec translation
// ---------------------------------------------------------------------------

func TestHpaToSpecs_NilHPA(t *testing.T) {
	require.Nil(t, hpaToSpecs(apiv1.ComponentType_TIDB, nil))
}

func TestHpaToSpecs_NilBehavior(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			// Behavior intentionally nil.
		},
	}
	require.Nil(t, hpaToSpecs(apiv1.ComponentType_TIDB, hpa),
		"HPA with no Behavior block → 0 specs (implicit-allow)")
}

func TestHpaToSpecs_TIKVExcluded(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			Behavior: hpaBehaviorEnabled(),
		},
	}
	require.Nil(t, hpaToSpecs(apiv1.ComponentType_TIKV, hpa),
		"TIKV has no HPA in this codebase — hpaToSpecs must emit 0 rows for it")
}

func TestHpaToSpecs_UnsupportedComponents(t *testing.T) {
	for _, ct := range []apiv1.ComponentType{
		apiv1.ComponentType_COMPONENT_TYPE_UNSPECIFIED,
	} {
		hpa := &autoscalingv2.HorizontalPodAutoscaler{
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				Behavior: hpaBehaviorEnabled(),
			},
		}
		require.Nil(t, hpaToSpecs(ct, hpa),
			"%s is not gated by GetRcuScaleComponents — 0 rows", ct.String())
	}
}

// TestHpaToSpecs_EmitsTiproxyBothDirections pins the new behavior:
// TIPROXY HPA emits 2 rows — 1 SCALE_OUT and 1 SCALE_IN, both at
// (NODE_COUNT, CPU), mirroring TIDB. This is the readback-side fix
// for the CLUSTER write/read key mismatch (CLUSTER scope accepts
// TIPROXY at the write path; the readback must emit the same
// (component, direction, capability, trigger) tuples).
func TestHpaToSpecs_EmitsTiproxyBothDirections(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			Behavior: hpaBehaviorEnabled(),
		},
	}
	specs := hpaToSpecs(apiv1.ComponentType_TIPROXY, hpa)
	require.Len(t, specs, 2, "TIPROXY should emit exactly 2 rows (SCALE_OUT + SCALE_IN)")
	byDir := map[apiv1.AutoscalePolicy_Direction]*entity.AutoscalePolicySpec{}
	for _, s := range specs {
		byDir[s.GetDirection()] = s
	}
	require.NotNil(t, byDir[apiv1.AutoscalePolicy_SCALE_OUT])
	require.NotNil(t, byDir[apiv1.AutoscalePolicy_SCALE_IN])
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIPROXY, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetTargetType())
	assert.Equal(t, apiv1.AutoscalePolicy_NODE_COUNT, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetCapability())
	assert.Equal(t, apiv1.AutoscalePolicy_CPU, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetTrigger())
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetState())
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIPROXY, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetTargetType())
	assert.Equal(t, apiv1.AutoscalePolicy_NODE_COUNT, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetCapability())
	assert.Equal(t, apiv1.AutoscalePolicy_CPU, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetTrigger())
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetState())
}

func TestHpaToSpecs_EmitsBothDirectionsForWorkerAndTIDB(t *testing.T) {
	for _, ct := range []apiv1.ComponentType{
		apiv1.ComponentType_TIDB_WORKER,
		apiv1.ComponentType_TIKV_WORKER,
		apiv1.ComponentType_COPROCESSOR_WORKER,
	} {
		hpa := &autoscalingv2.HorizontalPodAutoscaler{
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				Behavior: hpaBehaviorEnabled(),
			},
		}
		specs := hpaToSpecs(ct, hpa)
		require.Len(t, specs, 2, "%s should emit 2 rows (SCALE_OUT + SCALE_IN)", ct.String())
		hasScaleOut, hasScaleIn := false, false
		for _, s := range specs {
			assert.Equal(t, specpkg.ComponentTypeToAutoscaleTargetType(ct), s.GetTargetType())
			assert.Equal(t, apiv1.AutoscalePolicy_NODE_COUNT, s.GetCapability())
			assert.Equal(t, apiv1.AutoscalePolicy_CPU, s.GetTrigger())
			assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, s.GetState())
			switch s.GetDirection() {
			case apiv1.AutoscalePolicy_SCALE_OUT:
				hasScaleOut = true
			case apiv1.AutoscalePolicy_SCALE_IN:
				hasScaleIn = true
			}
		}
		assert.True(t, hasScaleOut, "%s should emit SCALE_OUT", ct.String())
		assert.True(t, hasScaleIn, "%s should emit SCALE_IN", ct.String())
	}
}

// TestHpaToSpecs_TIDBEmitsBothDirections pins the new behavior: TIDB
// HPA emits 2 rows — 1 SCALE_OUT (NODE_COUNT, CPU) from ScaleUp and
// 1 SCALE_IN (NODE_COUNT, CPU) from ScaleDown. This is the
// readback-side fix for the CLUSTER write/read key mismatch.
func TestHpaToSpecs_TIDBEmitsBothDirections(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			Behavior: hpaBehaviorEnabled(),
		},
	}
	specs := hpaToSpecs(apiv1.ComponentType_TIDB, hpa)
	require.Len(t, specs, 2, "TIDB should emit exactly 2 rows (SCALE_OUT + SCALE_IN)")
	byDir := map[apiv1.AutoscalePolicy_Direction]*entity.AutoscalePolicySpec{}
	for _, s := range specs {
		byDir[s.GetDirection()] = s
	}
	require.NotNil(t, byDir[apiv1.AutoscalePolicy_SCALE_OUT])
	require.NotNil(t, byDir[apiv1.AutoscalePolicy_SCALE_IN])
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIDB, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetTargetType())
	assert.Equal(t, apiv1.AutoscalePolicy_NODE_COUNT, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetCapability())
	assert.Equal(t, apiv1.AutoscalePolicy_CPU, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetTrigger())
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, byDir[apiv1.AutoscalePolicy_SCALE_OUT].GetState())
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIDB, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetTargetType())
	assert.Equal(t, apiv1.AutoscalePolicy_NODE_COUNT, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetCapability())
	assert.Equal(t, apiv1.AutoscalePolicy_CPU, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetTrigger())
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, byDir[apiv1.AutoscalePolicy_SCALE_IN].GetState())
}

func TestHpaToSpecs_DisabledSelectPolicyFlipsState(t *testing.T) {
	hpa := &autoscalingv2.HorizontalPodAutoscaler{
		Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
			Behavior: hpaBehaviorScaleInDisabled(),
		},
	}
	specs := hpaToSpecs(apiv1.ComponentType_TIDB_WORKER, hpa)
	require.Len(t, specs, 2, "worker now emits both SCALE_OUT and SCALE_IN")
	// Find the SCALE_IN row and assert it is DISABLED.
	var foundScaleIn bool
	for _, s := range specs {
		if s.GetDirection() == apiv1.AutoscalePolicy_SCALE_IN {
			assert.Equal(t, apiv1.AutoscalePolicy_DISABLED, s.GetState(),
				"DisabledPolicySelect → DISABLED SCALE_IN row")
			foundScaleIn = true
		}
	}
	assert.True(t, foundScaleIn, "must find a SCALE_IN row")
}

func TestPolicyStateFromScalingRules(t *testing.T) {
	up := autoscalingv2.MaxChangePolicySelect
	down := autoscalingv2.DisabledPolicySelect
	maxChange := autoscalingv2.MaxChangePolicySelect
	minChange := autoscalingv2.MinChangePolicySelect

	tests := []struct {
		name  string
		rules *autoscalingv2.HPAScalingRules
		want  apiv1.AutoscalePolicy_State
	}{
		{"nil rules → ENABLED (K8s default)", nil, apiv1.AutoscalePolicy_ENABLED},
		{"nil SelectPolicy → ENABLED (MaxChange default)", &autoscalingv2.HPAScalingRules{}, apiv1.AutoscalePolicy_ENABLED},
		{"MaxChangePolicySelect → ENABLED", &autoscalingv2.HPAScalingRules{SelectPolicy: &maxChange}, apiv1.AutoscalePolicy_ENABLED},
		{"MinChangePolicySelect → ENABLED", &autoscalingv2.HPAScalingRules{SelectPolicy: &minChange}, apiv1.AutoscalePolicy_ENABLED},
		{"DisabledPolicySelect → DISABLED", &autoscalingv2.HPAScalingRules{SelectPolicy: &down}, apiv1.AutoscalePolicy_DISABLED},
	}
	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			assert.Equal(t, tt.want, policyStateFromScalingRules(tt.rules))
		})
	}

	// Sanity: the K8s default types are reachable.
	_ = up
}

// ---------------------------------------------------------------------------
// readRuntimeSpecs scope routing: DEFAULT / TENANT short-circuit to (nil, nil)
// without touching any backing store.
// ---------------------------------------------------------------------------

// TestReadRuntimeSpecs_DefaultAndTenantScope pins the contract that
// DEFAULT and TENANT scopes have no runtime state — readRuntimeSpecs
// must return (nil, nil) without consulting any repo.
func TestReadRuntimeSpecs_DefaultAndTenantScope(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}
	for _, scope := range []apiv1.AutoscalePolicyScope{
		apiv1.AutoscalePolicyScope_SCOPE_DEFAULT,
		apiv1.AutoscalePolicyScope_SCOPE_TENANT,
	} {
		t.Run(scope.String(), func(t *testing.T) {
			specs, err := srv.readRuntimeSpecs(ctx, scope, "any-id")
			require.NoError(t, err)
			require.Nil(t, specs, "%s scope must return nil specs (no runtime state)", scope.String())
		})
	}
}

// ---------------------------------------------------------------------------
// readRuntimeSpecsForCluster: defensive / error / filter paths.
// ---------------------------------------------------------------------------

// TestReadRuntimeSpecsForCluster_NilRepoReturnsNil pins the contract
// that a nil tidbGroupRepo short-circuits to (nil, nil) — this is
// the implicit-allow path used by tests / harnesses that don't
// wire the cluster-level dependency.
func TestReadRuntimeSpecsForCluster_NilRepoReturnsNil(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{} // tidbGroupRepo intentionally nil
	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-1")
	require.NoError(t, err)
	require.Nil(t, specs)
}

// TestReadRuntimeSpecsForCluster_LoadErrorPropagates verifies that
// a tidbGroupRepo.Load error is wrapped with codes.Internal and
// surfaced to the caller — failures must not be silently swallowed
// at CLUSTER scope.
func TestReadRuntimeSpecsForCluster_LoadErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	srv := &serverImpl{tidbGroupRepo: mockTG}
	mockTG.On("Load", mock.Anything, "grp-bad", false).
		Return(nil, errors.New("etcd timeout")).Once()

	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-bad")
	require.Error(t, err)
	require.Nil(t, specs)
	assert.Equal(t, codes.Internal, status.Code(err),
		"CLUSTER Load error must surface as codes.Internal")
	assert.Contains(t, err.Error(), "grp-bad",
		"wrapped error must mention the failing group ID")
}

// TestReadRuntimeSpecsForCluster_NilTiDBGroupReturnsNil covers the
// "group not found" case — the repo returns (nil, nil) (no row
// in etcd). CLUSTER scope is effectively empty until the cluster
// is created.
func TestReadRuntimeSpecsForCluster_NilTiDBGroupReturnsNil(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	srv := &serverImpl{tidbGroupRepo: mockTG}
	mockTG.On("Load", mock.Anything, "grp-missing", false).
		Return(nil, nil).Once()

	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-missing")
	require.NoError(t, err)
	require.Nil(t, specs)
}

// TestReadRuntimeSpecsForCluster_EmptySharedPoolIDReturnsNil covers
// the "group loaded but not yet joined to a shared pool" case.
// CLUSTER scope is empty until shared_pool_id is wired up.
func TestReadRuntimeSpecsForCluster_EmptySharedPoolIDReturnsNil(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	srv := &serverImpl{tidbGroupRepo: mockTG}
	mockTG.On("Load", mock.Anything, "grp-no-sp", false).
		Return(&entity.TiDBGroup{
			TiDBGroup: apiv1.TiDBGroup{Id: "grp-no-sp" /* SharedPoolId: "" */},
		}, nil).Once()

	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-no-sp")
	require.NoError(t, err)
	require.Nil(t, specs)
}

// TestReadRuntimeSpecsForCluster_FiltersToTIDBAndTIPROXY pins the
// documented CLUSTER-scope contract: the function reads the
// shared-pool-level runtime specs and filters down to cluster-
// specific rows only. Per the hpaToSpecs comment, TIDB and TIPROXY
// both emit SCALE_IN rows (1 each) — they're the per-group
// components CLUSTER scope governs. Workers (TIDB_WORKER,
// TIKV_WORKER, COPROCESSOR_WORKER) and TIKV SCALE_OUT rows are
// dropped here because they belong to the shared-pool scope and
// resolve via the priority chain.
func TestReadRuntimeSpecsForCluster_FiltersToTIDBAndTIPROXY(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	mockRC := repomnocks.NewRuConfigRepo(t)
	srv := &serverImpl{
		tidbGroupRepo: mockTG,
		ruConfigRepo:  mockRC,
	}
	mockTG.On("Load", mock.Anything, "grp-1", false).
		Return(&entity.TiDBGroup{
			TiDBGroup: apiv1.TiDBGroup{Id: "grp-1", ClusterId: "cluster-1", SharedPoolId: "sp-1"},
		}, nil).Once()
	// HPA direct-read returns TIDB + TIDB_WORKER + TIKV_WORKER +
	// COPROCESSOR_WORKER + TIPROXY. CLUSTER scope's filter
	// strips workers; hpaToSpecs itself emits TIDB and TIPROXY
	// (both fall into the SCALE_IN group).
	//
	// Component identity model:
	//   TIDB HPA    → group label `db-<tidbGroupID>`   (= grpLabel)
	//   TIPROXY HPA → group label `db-<tidbClusterID>` (= clusterLabel)
	// Mirrors what generatedTiDBGroup / generatedTiProxyGroup write.
	grpLabel := hpaGroupLabel("grp-1", apiv1.ComponentType_TIDB)
	clusterLabel := hpaGroupLabel("cluster-1", apiv1.ComponentType_TIPROXY)
	installHPARig(t, srv, "sp-1", []hpaFixture{
		{ComponentLabel: "tidb", GroupLabel: grpLabel, Behavior: hpaBehaviorEnabled()},
		{ComponentLabel: "tiproxy", GroupLabel: clusterLabel, Behavior: hpaBehaviorEnabled()},
		{ComponentLabel: entity.TiDBWorkerType, Behavior: hpaBehaviorEnabled()},
		{ComponentLabel: entity.TiKVWorkerType, Behavior: hpaBehaviorEnabled()},
		{ComponentLabel: entity.CoprocessorWorkerType, Behavior: hpaBehaviorEnabled()},
	})
	// CLUSTER scope is HPA-only — no ru_configs read (CLUSTER has no
	// ru_configs row; shared-pool FeatureFlags are not relevant to
	// per-group TIDB/TIPROXY reads).

	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-1")
	require.NoError(t, err)
	seen := map[apiv1.AutoscaleTargetType]int{}
	for _, s := range specs {
		if s == nil {
			continue
		}
		seen[s.GetTargetType()]++
	}
	assert.Equal(t, 2, seen[apiv1.AutoscaleTargetType_TARGET_TIDB],
		"TIDB HPA emits 2 rows: SCALE_OUT + SCALE_IN (per-group component CLUSTER scope governs)")
	assert.Equal(t, 2, seen[apiv1.AutoscaleTargetType_TARGET_TIPROXY],
		"TIPROXY HPA emits 2 rows: SCALE_OUT + SCALE_IN (per-group component CLUSTER scope governs)")
	assert.Equal(t, 0, seen[apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER], "TIDB_WORKER must be filtered at CLUSTER scope")
	assert.Equal(t, 0, seen[apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER], "TIKV_WORKER must be filtered at CLUSTER scope")
	assert.Equal(t, 0, seen[apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER], "COPROCESSOR_WORKER must be filtered at CLUSTER scope")
	assert.Equal(t, 0, seen[apiv1.AutoscaleTargetType_TARGET_TIKV], "TIKV SCALE_OUT must not be emitted at CLUSTER scope")
}

// TestReadRuntimeSpecsForCluster_SharedPoolChainErrorPropagates
// pins that an error coming out of the shared-pool chain (HPA
// load, K8s client, etc.) propagates through readRuntimeSpecsForCluster
// to the caller — failures inside the cluster's underlying pool
// must surface, not be swallowed.
func TestReadRuntimeSpecsForCluster_SharedPoolChainErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	mockBundle := repomnocks.NewAutoscalePolicyBundleRepo(t)
	srv := &serverImpl{
		tidbGroupRepo:             mockTG,
		autoscalePolicyBundleRepo: mockBundle,
	}
	mockTG.On("Load", mock.Anything, "grp-1", false).
		Return(&entity.TiDBGroup{
			TiDBGroup: apiv1.TiDBGroup{Id: "grp-1", SharedPoolId: "sp-broken"},
		}, nil).Once()

	// sharedPoolRepo + dataplaneManager rigged so the K8s List call
	// returns an error. installHPARig with `.Maybe()` mocks doesn't
	// help here — we need a one-shot erroring list. Hand-roll the rig.
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-broken", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-broken", DataplaneName: "dp-sp-broken"},
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-broken").Return(kubeclient, nil).Once()
	// CLUSTER scope issues one List call per component type; the
	// first matching call propagates the error before any
	// subsequent calls run. `.Maybe()` keeps the mock from
	// panicking on later calls that never reach K8s once the
	// first error surfaces.
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Return(errors.New("kube apiserver down")).Maybe()
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager
	// Bundle repo not needed for this path but referenced via
	// autoscalePolicyBundleRepo field — set nil is fine here because
	// readRuntimeSpecsForCluster does not touch it.
	_ = mockBundle

	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-1")
	require.Error(t, err)
	require.Nil(t, specs)
	assert.Equal(t, codes.Internal, status.Code(err),
		"HPA chain errors must surface as codes.Internal")
	assert.Contains(t, err.Error(), "sp-broken",
		"wrapped error must identify the failing shared pool")
}

// ---------------------------------------------------------------------------
// listSharedPoolHPASpecs error paths + unknown-label filter.
// ---------------------------------------------------------------------------

// TestListSharedPoolHPASpecs_EmptySharedPoolIDReturnsNil pins the
// defensive guard at the top of the function.
func TestListSharedPoolHPASpecs_EmptySharedPoolIDReturnsNil(t *testing.T) {
	srv := &serverImpl{}
	specs, err := srv.listSharedPoolHPASpecs(context.Background(), "", "", "")
	require.NoError(t, err)
	require.Nil(t, specs)
}

// TestListSharedPoolHPASpecs_SharedPoolLoadErrorPropagates verifies
// sharedPoolRepo.Load errors are wrapped with %w and surfaced.
func TestListSharedPoolHPASpecs_SharedPoolLoadErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockSP := repomnocks.NewSharedPoolRepo(t)
	srv := &serverImpl{sharedPoolRepo: mockSP}
	mockSP.On("Load", mock.Anything, "sp-x", false).
		Return(nil, errors.New("etcd down")).Once()

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-x", "", "")
	require.Error(t, err)
	require.Nil(t, specs)
	assert.Contains(t, err.Error(), "etcd down",
		"underlying error must be wrapped with %w so callers can errors.Is")
}

// TestListSharedPoolHPASpecs_GetClientErrorPropagates verifies
// dataplaneManager.GetClient errors are wrapped with %w.
func TestListSharedPoolHPASpecs_GetClientErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockSP := repomnocks.NewSharedPoolRepo(t)
	mockDM := dataplanemocks.NewManager(t)
	srv := &serverImpl{
		sharedPoolRepo:   mockSP,
		dataplaneManager: mockDM,
	}
	mockSP.On("Load", mock.Anything, "sp-y", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-y", DataplaneName: "dp-y"},
	}, nil).Once()
	mockDM.On("GetClient", "dp-y").Return(nil, errors.New("no dataplane")).Once()

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-y", "", "")
	require.Error(t, err)
	require.Nil(t, specs)
	assert.Contains(t, err.Error(), "no dataplane")
}

// TestListSharedPoolHPASpecs_KubeListErrorPropagates verifies the
// K8s List error path is wrapped with %w.
func TestListSharedPoolHPASpecs_KubeListErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockSP := repomnocks.NewSharedPoolRepo(t)
	mockDM := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	srv := &serverImpl{
		sharedPoolRepo:   mockSP,
		dataplaneManager: mockDM,
	}
	mockSP.On("Load", mock.Anything, "sp-z", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-z", DataplaneName: "dp-z"},
	}, nil).Once()
	mockDM.On("GetClient", "dp-z").Return(kubeclient, nil).Once()
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Return(errors.New("list timeout")).Once()

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-z", "", "")
	require.Error(t, err)
	require.Nil(t, specs)
	assert.Contains(t, err.Error(), "list timeout")
}

// TestListSharedPoolHPASpecs_SkipsUnknownComponentLabel pins that an
// HPA whose LabelComponentKey is not in componentTypeLookup is
// silently skipped — it does NOT produce a 0-row spec or panic.
// This protects against future label additions breaking the read
// path until hpaToSpecs learns about them.
func TestListSharedPoolHPASpecs_SkipsUnknownComponentLabel(t *testing.T) {
	ctx := context.Background()
	mockSP := repomnocks.NewSharedPoolRepo(t)
	mockDM := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	srv := &serverImpl{
		sharedPoolRepo:   mockSP,
		dataplaneManager: mockDM,
	}
	mockSP.On("Load", mock.Anything, "sp-mix", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-mix", DataplaneName: "dp-mix"},
	}, nil).Once()
	mockDM.On("GetClient", "dp-mix").Return(kubeclient, nil).Once()
	// SHARED_POOL scope now issues 3 worker List calls. Inject the
	// known TIDB_WORKER HPA + the unknown-label HPA on the first
	// worker call (TIDB_WORKER); the other two return empty.
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
				{
					ObjectMeta: metav1.ObjectMeta{
						Labels: map[string]string{
							schemav1.LabelComponentKey: entity.TiDBWorkerType,
						},
					},
					Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
						Behavior: hpaBehaviorEnabled(),
					},
				},
				{
					ObjectMeta: metav1.ObjectMeta{
						Labels: map[string]string{
							schemav1.LabelComponentKey: "future-component-not-yet-mapped",
						},
					},
					Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
						Behavior: hpaBehaviorEnabled(),
					},
				},
			}
		}).
		Return(nil).Once()
	// TIKV_WORKER and COPROCESSOR_WORKER calls: empty.
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Return(nil).Times(2)

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-mix", "", "")
	require.NoError(t, err)
	require.Len(t, specs, 2, "unknown label HPA must be skipped — only TIDB_WORKER produces a spec")
	require.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, specs[0].GetTargetType())
}

// TestListSharedPoolHPASpecs_EmptyNamespaceReturnsEmptySlice pins
// the contract: namespace with zero HPAs returns an empty,
// NON-nil slice (the implicit-allow case for the 3 scale
// entrypoints).
func TestListSharedPoolHPASpecs_EmptyNamespaceReturnsEmptySlice(t *testing.T) {
	ctx := context.Background()
	mockSP := repomnocks.NewSharedPoolRepo(t)
	mockDM := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	srv := &serverImpl{
		sharedPoolRepo:   mockSP,
		dataplaneManager: mockDM,
	}
	mockSP.On("Load", mock.Anything, "sp-empty", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-empty", DataplaneName: "dp-empty"},
	}, nil).Once()
	mockDM.On("GetClient", "dp-empty").Return(kubeclient, nil).Once()
	// SHARED_POOL scope issues 3 worker List calls; all return empty.
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Return(nil).Times(3)

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-empty", "", "")
	require.NoError(t, err)
	require.NotNil(t, specs, "empty namespace must return non-nil empty slice (implicit-allow contract)")
	require.Empty(t, specs)
}

// ---------------------------------------------------------------------------
// readRuntimeSpecsForSharedPool silent-skip paths (graceful degradation)
// ---------------------------------------------------------------------------

// TestReadRuntimeSpecsForSharedPool_RuConfigErrorPropagates pins the
// contract: GetRuConfig returns the server default (non-nil, nil
// error) when a row is absent, so a non-nil error is a real DB /
// unmarshal failure and MUST surface — treating it as implicit-allow
// would silently flip a scale decision from deny to allow.
func TestReadRuntimeSpecsForSharedPool_RuConfigErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockRC := repomnocks.NewRuConfigRepo(t)
	srv := &serverImpl{ruConfigRepo: mockRC}
	installHPARig(t, srv, "sp-rc", []hpaFixture{
		{ComponentLabel: entity.TiDBWorkerType, Behavior: hpaBehaviorEnabled()},
	})
	mockRC.On("GetRuConfig", mock.Anything, "sp-rc", apiv1.RuConfig_SHARED_POOL).
		Return(nil, errors.New("ru_config read failed")).Once()

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-rc")
	require.Error(t, err, "a real ru_config error must surface, not be swallowed")
	require.Nil(t, specs)
	assert.Equal(t, codes.Internal, status.Code(err),
		"ru_config read error must surface as codes.Internal")
}

// TestReadRuntimeSpecsForSharedPool_RuConfigNilRowSwallowed verifies
// that a (nil, nil) return from ru_config repo is treated the
// same as the error case — implicit-allow.
func TestReadRuntimeSpecsForSharedPool_RuConfigNilRowSwallowed(t *testing.T) {
	ctx := context.Background()
	mockRC := repomnocks.NewRuConfigRepo(t)
	srv := &serverImpl{ruConfigRepo: mockRC}
	installHPARig(t, srv, "sp-nil-rc", []hpaFixture{
		{ComponentLabel: entity.TiDBWorkerType, Behavior: hpaBehaviorEnabled()},
	})
	mockRC.On("GetRuConfig", mock.Anything, "sp-nil-rc", apiv1.RuConfig_SHARED_POOL).
		Return(nil, nil).Once()

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-nil-rc")
	require.NoError(t, err)
	require.Len(t, specs, 2, "worker emits SCALE_IN + SCALE_OUT")
}

// TestReadRuntimeSpecsForSharedPool_RuConfigNilFeatureFlagsSwallowed
// verifies that a row with FeatureFlags == nil still contributes the
// full 9 DISABLED specs (TIKV × 4 axes + SHARED_POOL × 2 + TIDB_GROUP × 2
// + TIPROXY SCALE_IN),
// not zero — every key shows up in the bundle so the operator sees the
// full policy surface. The FeatureFlags path must not panic on a nil
// FeatureFlags struct.
func TestReadRuntimeSpecsForSharedPool_RuConfigNilFeatureFlagsSwallowed(t *testing.T) {
	ctx := context.Background()
	mockRC := repomnocks.NewRuConfigRepo(t)
	srv := &serverImpl{ruConfigRepo: mockRC}
	installHPARig(t, srv, "sp-empty-ff", []hpaFixture{
		{ComponentLabel: entity.TiDBWorkerType, Behavior: hpaBehaviorEnabled()},
	})
	mockRC.On("GetRuConfig", mock.Anything, "sp-empty-ff", apiv1.RuConfig_SHARED_POOL).
		Return(&entity.RuConfig{RuConfig: apiv1.RuConfig{}}, nil).Once()

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-empty-ff")
	require.NoError(t, err)
	// 2 worker HPA specs (SCALE_OUT + SCALE_IN) + 9 FeatureFlags fallback
	// specs (TIKV × 4 axes + SHARED_POOL × 2 + TIDB_GROUP × 2 + TIPROXY SCALE_IN).
	require.Len(t, specs, 11,
		"FeatureFlags==nil must contribute 9 DISABLED fallback specs; HPA-derived row still present")
	// First two are the HPA-derived worker specs; the remaining 9
	// are the FeatureFlags fallback (all DISABLED).
	flagSpecs := specs[2:]
	disabledCount := 0
	for _, s := range flagSpecs {
		if s.GetState() == apiv1.AutoscalePolicy_DISABLED {
			disabledCount++
		}
	}
	require.Equal(t, 9, disabledCount,
		"all 9 FeatureFlags fallback specs must be DISABLED when ff is nil")
}

// TestReadRuntimeSpecsForSharedPool_NilSharedPoolRepoSkipsHPA pins
// the nil-guard: when sharedPoolRepo is nil, the function must
// skip the HPA direct-read path entirely and only consult
// ru_config.
func TestReadRuntimeSpecsForSharedPool_NilSharedPoolRepoSkipsHPA(t *testing.T) {
	ctx := context.Background()
	mockRC := repomnocks.NewRuConfigRepo(t)
	srv := &serverImpl{ruConfigRepo: mockRC} // sharedPoolRepo nil
	enabled := true
	mockRC.On("GetRuConfig", mock.Anything, "sp-no-spr", apiv1.RuConfig_SHARED_POOL).
		Return(&entity.RuConfig{RuConfig: apiv1.RuConfig{FeatureFlags: &apiv1.RuConfig_FeatureFlags{
			EnableAutoscaleTikvCpuScaleOut: &enabled,
		}}}, nil).Once()

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-no-spr")
	require.NoError(t, err)
	// Partial FeatureFlags (only EnableAutoscaleTikvCpuScaleOut set):
	// 1 ENABLED TIKV SCALE_OUT/NODE_COUNT/CPU + 8 DISABLED fallback
	// specs (TIKV × 3 remaining axes + SHARED_POOL × 2 + TIDB_GROUP × 2 + TIPROXY SCALE_IN).
	require.Len(t, specs, 9)
	enabledFound := false
	for _, s := range specs {
		if s.GetTargetType() == apiv1.AutoscaleTargetType_TARGET_TIKV &&
			s.GetCapability() == apiv1.AutoscalePolicy_NODE_COUNT &&
			s.GetTrigger() == apiv1.AutoscalePolicy_CPU &&
			s.GetState() == apiv1.AutoscalePolicy_ENABLED {
			enabledFound = true
			break
		}
	}
	require.True(t, enabledFound,
		"FeatureFlags.EnableAutoscaleTikvCpuScaleOut=true must surface as TIKV ENABLED")
}

// TestReadRuntimeSpecsForSharedPool_NilDataPlaneManagerSkipsHPA —
// same contract, but with dataplaneManager nil. Ru_config still
// contributes; HPA path is skipped.
func TestReadRuntimeSpecsForSharedPool_NilDataPlaneManagerSkipsHPA(t *testing.T) {
	ctx := context.Background()
	mockRC := repomnocks.NewRuConfigRepo(t)
	mockSP := repomnocks.NewSharedPoolRepo(t)
	srv := &serverImpl{
		sharedPoolRepo: mockSP,
		ruConfigRepo:   mockRC,
		// dataplaneManager intentionally nil
	}
	enabled := true
	mockRC.On("GetRuConfig", mock.Anything, "sp-no-dm", apiv1.RuConfig_SHARED_POOL).
		Return(&entity.RuConfig{RuConfig: apiv1.RuConfig{FeatureFlags: &apiv1.RuConfig_FeatureFlags{
			EnableAutoscaleTikvCpuScaleOut: &enabled,
		}}}, nil).Once()

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-no-dm")
	require.NoError(t, err)
	require.Len(t, specs, 9)
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIKV, specs[0].GetTargetType())
}

// TestReadRuntimeSpecsForSharedPool_HPAErrorPropagates pins that a
// K8s List error is wrapped with codes.Internal — HPA failures are
// NOT silently swallowed at SHARED_POOL scope (operator-facing
// RPC must surface failures rather than mask with stale data).
func TestReadRuntimeSpecsForSharedPool_HPAErrorPropagates(t *testing.T) {
	ctx := context.Background()
	mockSP := repomnocks.NewSharedPoolRepo(t)
	mockDM := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	srv := &serverImpl{
		sharedPoolRepo:   mockSP,
		dataplaneManager: mockDM,
	}
	mockSP.On("Load", mock.Anything, "sp-hpa-fail", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-hpa-fail", DataplaneName: "dp-hpa-fail"},
	}, nil).Once()
	mockDM.On("GetClient", "dp-hpa-fail").Return(kubeclient, nil).Once()
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Return(errors.New("apiserver 503")).Once()

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-hpa-fail")
	require.Error(t, err)
	require.Nil(t, specs)
	assert.Equal(t, codes.Internal, status.Code(err),
		"HPA read errors must surface as codes.Internal")
	assert.Contains(t, err.Error(), "apiserver 503",
		"underlying error must be wrapped so callers can errors.Is")
}

// TestReadRuntimeSpecsForSharedPool_ExcludesAllTIDB pins R1: with no
// HasSharedTiProxy, the SHARED_POOL read returns only workers (no
// TIDB at all). This is the post-fix guarantee against the leak
// where per-group TIDB HPAs used to be returned.
func TestReadRuntimeSpecsForSharedPool_ExcludesAllTIDB(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-pure", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-pure", DataplaneName: "dp-sp-pure"},
		DeployType: "", // !HasSharedTiProxy
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-pure").Return(kubeclient, nil).Once()

	// Three worker List calls: TIDB_WORKER, TIKV_WORKER, COPROCESSOR_WORKER.
	// Return a TIDB HPA on the TIDB_WORKER call to verify selector
	// exclusion (the worker selector must NOT return TIDB HPAs).
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			sel := opts[0].(*client.ListOptions).LabelSelector.String()
			if sel == "pingcap.com/component=worker-tidb" {
				hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
					{
						ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
							schemav1.LabelComponentKey: entity.TiDBWorkerType,
						}},
						Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()},
					},
				}
			}
		}).
		Return(nil).Times(3)
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-pure")
	require.NoError(t, err)
	for _, s := range specs {
		assert.NotEqual(t, apiv1.AutoscaleTargetType_TARGET_TIDB, s.GetTargetType(),
			"SHARED_POOL scope must NOT include TIDB (CLUSTER-only)")
	}
	require.Len(t, specs, 2, "TIDB_WORKER emits 2 rows (SCALE_OUT + SCALE_IN)")
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, specs[0].GetTargetType())
}

// TestReadRuntimeSpecsForSharedPool_ExcludesPerGroupTiProxy pins R1:
// per-group TIPROXY HPAs (CLUSTER-only) are excluded from SHARED_POOL
// reads.
func TestReadRuntimeSpecsForSharedPool_ExcludesPerGroupTiProxy(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-pg-tip", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-pg-tip", DataplaneName: "dp-sp-pg-tip"},
		DeployType: "", // !HasSharedTiProxy
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-pg-tip").Return(kubeclient, nil).Once()

	// 3 worker List calls. All return a TIDB_WORKER HPA (so the read
	// result has at least one non-empty spec).
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
				{
					ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
						schemav1.LabelComponentKey: entity.TiDBWorkerType,
					}},
					Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()},
				},
			}
		}).
		Return(nil).Times(3)
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-pg-tip")
	require.NoError(t, err)
	for _, s := range specs {
		assert.NotEqual(t, apiv1.AutoscaleTargetType_TARGET_TIPROXY, s.GetTargetType(),
			"SHARED_POOL scope must NOT include per-group TIPROXY (CLUSTER-only)")
	}
}

// TestReadRuntimeSpecsForSharedPool_IncludesPoolSharedTiProxy pins R1
// + design §Pool-shared TIPROXY: a HasSharedTiProxy pool surfaces the
// pool-shared TIPROXY HPA at SHARED_POOL scope, identified by the
// group label `db-<FnvHash(spID)>`.
func TestReadRuntimeSpecsForSharedPool_IncludesPoolSharedTiProxy(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}
	sharedPoolID := "sp-pooltip"
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, sharedPoolID, false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: sharedPoolID, DataplaneName: "dp-" + sharedPoolID},
		DeployType: "shared_tiproxy", // HasSharedTiProxy
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-"+sharedPoolID).Return(kubeclient, nil).Once()

	// 4 List calls (3 workers + 1 pool-shared TIPROXY). The
	// TIPROXY call carries both the component label and the
	// group label `db-<FnvHash(spID)>`; we inject a matching
	// HPA on that call. Other worker calls return empty.
	poolTipGroupLabel := hpaGroupLabel(utils.FnvHash(sharedPoolID), apiv1.ComponentType_TIPROXY)
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			sel := opts[0].(*client.ListOptions).LabelSelector.String()
			// Match the pool-shared TIPROXY call (carries both
			// component and group labels).
			if sel == "pingcap.com/component=tiproxy,pingcap.com/group="+poolTipGroupLabel {
				hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
					{
						ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
							schemav1.LabelComponentKey:    "tiproxy",
							schemav1.LabelPingcapGroupKey: poolTipGroupLabel,
						}},
						Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()},
					},
				}
			}
		}).
		Return(nil).Times(4)
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, sharedPoolID)
	require.NoError(t, err)
	foundTip := false
	hasScaleOut := false
	hasScaleIn := false
	for _, s := range specs {
		if s.GetTargetType() == apiv1.AutoscaleTargetType_TARGET_TIPROXY {
			foundTip = true
			switch s.GetDirection() {
			case apiv1.AutoscalePolicy_SCALE_OUT:
				hasScaleOut = true
			case apiv1.AutoscalePolicy_SCALE_IN:
				hasScaleIn = true
			}
		}
	}
	assert.True(t, foundTip, "pool-shared TIPROXY HPA must appear at SHARED_POOL scope when HasSharedTiProxy")
	assert.True(t, hasScaleOut, "pool-shared TIPROXY must emit a SCALE_OUT row")
	assert.True(t, hasScaleIn, "pool-shared TIPROXY must emit a SCALE_IN row")
}

// TestReadRuntimeSpecsForSharedPool_EmptySharedPoolIDReturnsNil
// pins the defensive guard at the top of the function.
func TestReadRuntimeSpecsForSharedPool_EmptySharedPoolIDReturnsNil(t *testing.T) {
	srv := &serverImpl{}
	specs, err := srv.readRuntimeSpecsForSharedPool(context.Background(), "")
	require.NoError(t, err)
	require.Nil(t, specs)
}

// ---------------------------------------------------------------------------
// Cross-group HPA isolation. Pins the bug fix for the P1: in a shared
// pool hosting multiple TiDB groups, listSharedPoolHPASpecs must scope
// its K8s-side selector by group for TIDB and TIPROXY (the 2
// group-scoped components) so readRuntimeSpecsForCluster cannot leak
// another group's HPA-derived rows.
//
// SHARED_POOL scope must continue to return HPAs from every group in
// the namespace — that is its documented behavior (workers live at the
// shared-pool level).
// ---------------------------------------------------------------------------

// hpaGroupLabel returns the value SetHPAScaleBehavior writes to
// LabelPingcapGroupKey for the given (groupID, componentType) pair.
// Matches entity.GetComponentGroupName + the write-side selector in
// domain/server/set_hpa_scale_behavior.go.
func hpaGroupLabel(groupID string, ct apiv1.ComponentType) string {
	return entity.GetComponentGroupName(&entity.Component{
		Id:            groupID,
		ComponentType: ct,
	})
}

// installSharedPoolWithMultipleGroups wires the shared-pool/dataplane/
// K8s mocks so the namespace contains HPAs from TWO TiDB groups
// ("grp-a" and "grp-b") sharing the same shared pool. The K8s List
// mock is selector-aware: it returns fixtures whose labels match the
// caller's LabelSelector (and an empty list for unmatched selectors),
// which is what a real K8s apiserver would do.
//
// The 3 worker HPAs are shared-pool-scoped (no group label) — each
// group fixture list contains one of each.
//
// TIDB and TIPROXY HPAs each carry LabelPingcapGroupKey set to
// hpaGroupLabel(...) — the same selector SetHPAScaleBehavior writes.
func installSharedPoolWithMultipleGroups(
	t *testing.T,
	srv *serverImpl,
	sharedPoolID string,
	groupAID, groupBID, clusterAID, clusterBID string,
	// stateA is the scale-down state for groupA's TIDB/TIPROXY HPAs.
	stateA apiv1.AutoscalePolicy_State,
	// stateB is the scale-down state for groupB's TIDB/TIPROXY HPAs.
	stateB apiv1.AutoscalePolicy_State,
) {
	t.Helper()
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)

	sharedPoolRepo.On("Load", mock.Anything, sharedPoolID, false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{
			Id:            sharedPoolID,
			DataplaneName: "dp-" + sharedPoolID,
		},
	}, nil).Maybe()
	dataplaneManager.On("GetClient", "dp-"+sharedPoolID).Return(kubeclient, nil).Maybe()

	buildHPA := func(name, componentLabel, groupLabel string, state apiv1.AutoscalePolicy_State) autoscalingv2.HorizontalPodAutoscaler {
		var behavior *autoscalingv2.HorizontalPodAutoscalerBehavior
		switch state {
		case apiv1.AutoscalePolicy_ENABLED:
			behavior = hpaBehaviorEnabled()
		case apiv1.AutoscalePolicy_DISABLED:
			behavior = hpaBehaviorScaleInDisabled()
		}
		labels := map[string]string{schemav1.LabelComponentKey: componentLabel}
		if groupLabel != "" {
			labels[schemav1.LabelPingcapGroupKey] = groupLabel
		}
		return autoscalingv2.HorizontalPodAutoscaler{
			ObjectMeta: metav1.ObjectMeta{Name: name, Labels: labels},
			Spec: autoscalingv2.HorizontalPodAutoscalerSpec{
				Behavior: behavior,
			},
		}
	}

	// Component identity model (must mirror what generatedTiDBGroup /
	// generatedTiProxyGroup write in production):
	//   TIDB HPA    → `db-<tidbGroupID>`   (= hpaGroupLabel(groupID, TIDB))
	//   TIPROXY HPA → `db-<tidbClusterID>` (= hpaGroupLabel(clusterID, TIPROXY))
	groupATidb := buildHPA("tidb-a", "tidb", hpaGroupLabel(groupAID, apiv1.ComponentType_TIDB), stateA)
	groupATiproxy := buildHPA("tiproxy-a", "tiproxy", hpaGroupLabel(clusterAID, apiv1.ComponentType_TIPROXY), stateA)
	groupBTidb := buildHPA("tidb-b", "tidb", hpaGroupLabel(groupBID, apiv1.ComponentType_TIDB), stateB)
	groupBTiproxy := buildHPA("tiproxy-b", "tiproxy", hpaGroupLabel(clusterBID, apiv1.ComponentType_TIPROXY), stateB)
	// Shared-pool worker HPAs (no group label — workers are
	// pool-scoped). One set shared between both groups.
	workerTidb := buildHPA("worker-tidb-sp", entity.TiDBWorkerType, "", apiv1.AutoscalePolicy_ENABLED)
	workerTikv := buildHPA("worker-tikv-sp", entity.TiKVWorkerType, "", apiv1.AutoscalePolicy_ENABLED)
	workerCop := buildHPA("worker-cop-sp", entity.CoprocessorWorkerType, "", apiv1.AutoscalePolicy_ENABLED)

	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			if len(opts) == 0 {
				// SHARED_POOL scope: no selector → return every
				// HPA in the namespace (all groups + workers).
				hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
					groupATidb, groupBTidb,
					groupATiproxy, groupBTiproxy,
					workerTidb, workerTikv, workerCop,
				}
				return
			}
			listOpts, ok := opts[0].(*client.ListOptions)
			if !ok || listOpts.LabelSelector == nil {
				hpaList.Items = nil
				return
			}
			sel := listOpts.LabelSelector
			// Pool of all HPAs in the namespace — the selector
			// decides which ones a real K8s apiserver would return.
			all := []*autoscalingv2.HorizontalPodAutoscaler{
				&groupATidb, &groupBTidb,
				&groupATiproxy, &groupBTiproxy,
				&workerTidb, &workerTikv, &workerCop,
			}
			for _, hpa := range all {
				if sel.Matches(labels.Set(hpa.GetLabels())) {
					hpaList.Items = append(hpaList.Items, *hpa)
				}
			}
		}).
		Return(nil).Maybe()

	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager
}

// TestReadRuntimeSpecsForCluster_GroupScopedSelectorIsolatesGroups is
// the regression test for the P1 bug: a shared pool hosting multiple
// TiDB groups must NOT bleed another group's TIDB/TIPROXY HPA-derived
// rows into a single group's CLUSTER-scope effective read.
//
// Setup: two TiDB groups (grp-a, grp-b) in the same shared pool, each
// with its own TIDB + TIPROXY HPA. groupA's HPA carries ScaleDown
// ENABLED while groupB's carries ScaleDown DISABLED.
//
// Expected: readRuntimeSpecsForCluster("grp-a") returns ONLY the
// grp-a rows (2 TIDB rows: SCALE_OUT + SCALE_IN, 2 TIPROXY rows:
// SCALE_OUT + SCALE_IN), all with state=ENABLED. No rows from
// groupB. The K8s selector must scope both group-scoped components
// correctly.
func TestReadRuntimeSpecsForCluster_GroupScopedSelectorIsolatesGroups(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	srv := &serverImpl{tidbGroupRepo: mockTG}
	mockTG.On("Load", mock.Anything, "grp-a", false).
		Return(&entity.TiDBGroup{
			TiDBGroup: apiv1.TiDBGroup{Id: "grp-a", ClusterId: "cluster-a", SharedPoolId: "sp-multi"},
		}, nil).Once()

	installSharedPoolWithMultipleGroups(t, srv, "sp-multi", "grp-a", "grp-b", "cluster-a", "cluster-b",
		apiv1.AutoscalePolicy_ENABLED,  // groupA TIDB/TIPROXY scale-down state
		apiv1.AutoscalePolicy_DISABLED, // groupB TIDB/TIPROXY scale-down state
	)

	specs, err := srv.readRuntimeSpecsForCluster(ctx, "grp-a")
	require.NoError(t, err)

	// groupA → 4 rows: 2 TIDB (SCALE_OUT + SCALE_IN, both ENABLED)
	// + 2 TIPROXY (SCALE_OUT + SCALE_IN, both ENABLED).
	// The K8s selector must scope both group-scoped components so
	// groupB's HPAs never reach us.
	require.Len(t, specs, 4,
		"CLUSTER scope for grp-a must return grp-a's TIDB + TIPROXY rows only (2 each, both directions = 4 total)")
	seen := map[apiv1.AutoscaleTargetType]map[apiv1.AutoscalePolicy_Direction]apiv1.AutoscalePolicy_State{}
	for _, s := range specs {
		if seen[s.GetTargetType()] == nil {
			seen[s.GetTargetType()] = map[apiv1.AutoscalePolicy_Direction]apiv1.AutoscalePolicy_State{}
		}
		seen[s.GetTargetType()][s.GetDirection()] = s.GetState()
	}
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, seen[apiv1.AutoscaleTargetType_TARGET_TIDB][apiv1.AutoscalePolicy_SCALE_OUT],
		"TIDB SCALE_OUT must reflect grp-a's HPA state, not grp-b's")
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, seen[apiv1.AutoscaleTargetType_TARGET_TIDB][apiv1.AutoscalePolicy_SCALE_IN],
		"TIDB SCALE_IN must reflect grp-a's HPA state, not grp-b's")
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, seen[apiv1.AutoscaleTargetType_TARGET_TIPROXY][apiv1.AutoscalePolicy_SCALE_OUT],
		"TIPROXY SCALE_OUT must reflect grp-a's HPA state, not grp-b's")
	assert.Equal(t, apiv1.AutoscalePolicy_ENABLED, seen[apiv1.AutoscaleTargetType_TARGET_TIPROXY][apiv1.AutoscalePolicy_SCALE_IN],
		"TIPROXY SCALE_IN must reflect grp-a's HPA state, not grp-b's")
}

// TestReadRuntimeSpecsForCluster_GroupScopedSelectorIsolatesGroups_TIPROXY
// is the TIPROXY counterpart of the regression test. TIPROXY HPAs are
// also group-scoped (LabelPingcapGroupKey) per SetHPAScaleBehavior's
// write path, so a selector regression would leak them just the same.
//
// TIPROXY is not currently emitted by hpaToSpecs (per the comment on
// hpaToSpecs: TIPROXY is not in GetRcuScaleComponents). The bug is
// still real if the CLUSTER filter is ever widened to include
// TIPROXY-emitted rows — the K8s-side selector must already be
// group-scoped. We verify the selector directly by inspecting the
// kubeclient mock's captured list options.
func TestReadRuntimeSpecsForCluster_GroupScopedSelectorIsolatesGroups_TIPROXY(t *testing.T) {
	ctx := context.Background()
	mockTG := repomnocks.NewTiDBGroupRepo(t)
	srv := &serverImpl{tidbGroupRepo: mockTG}
	mockTG.On("Load", mock.Anything, "grp-a", false).
		Return(&entity.TiDBGroup{
			TiDBGroup: apiv1.TiDBGroup{Id: "grp-a", ClusterId: "cluster-a", SharedPoolId: "sp-multi"},
		}, nil).Once()

	// Wire a capture-only mock that records every (componentLabel,
	// groupLabel) pair the K8s List is called with. This pins that
	// the per-component-type list calls include BOTH the group
	// label for the 2 group-scoped components (TIDB, TIPROXY) AND
	// only the component label for the 3 workers.
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-multi", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-multi", DataplaneName: "dp-sp-multi"},
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-multi").Return(kubeclient, nil).Once()

	type selectorCapture struct {
		component string
	}
	var seen []selectorCapture
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			if len(opts) == 0 {
				return
			}
			listOpts, ok := opts[0].(*client.ListOptions)
			if !ok || listOpts.LabelSelector == nil {
				return
			}
			seen = append(seen, selectorCapture{
				component: listOpts.LabelSelector.String(),
			})
			_ = hpaList
		}).
		Return(nil).Maybe()

	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	_, err := srv.readRuntimeSpecsForCluster(ctx, "grp-a")
	require.NoError(t, err)

	// 2 List calls expected: TIDB and TIPROXY only. Workers belong to
	// SHARED_POOL scope (their HPA is per-sharedpool, not per-cluster),
	// so CLUSTER scope does not query worker HPAs. readRuntimeSpecsForCluster
	// downstream-filters non-TIDB/non-TIPROXY anyway; narrowing the list
	// avoids wasted K8s API calls.
	require.Len(t, seen, 2,
		"CLUSTER scope must issue exactly 2 K8s list calls (TIDB + TIPROXY only)")

	// Aggregate the selector strings; the TIPROXY + TIDB selectors
	// must contain the group label.
	gotGroupScoped := map[string]string{} // componentLabel → full selector string
	for _, s := range seen {
		// Find "pingcap.com/component=<value>" within the selector
		// string and key by it. The selector's String() form is
		// deterministic: "a=b,c=d".
		compKey := schemav1.LabelComponentKey + "="
		idx := -1
		for i := 0; i+len(compKey) <= len(s.component); i++ {
			if s.component[i:i+len(compKey)] == compKey {
				idx = i + len(compKey)
				break
			}
		}
		require.GreaterOrEqual(t, idx, 0, "selector %q missing %s", s.component, compKey)
		// Component label ends at ',' or end-of-string.
		end := idx
		for end < len(s.component) && s.component[end] != ',' {
			end++
		}
		component := s.component[idx:end]
		gotGroupScoped[component] = s.component
	}

	// TIDB and TIPROXY must include the group label. The id used
	// differs by component (TIDB → tidbGroupID, TIPROXY → tidbClusterID).
	wantGroupSubstr := map[string]string{
		"tidb":    hpaGroupLabel("grp-a", apiv1.ComponentType_TIDB),
		"tiproxy": hpaGroupLabel("cluster-a", apiv1.ComponentType_TIPROXY),
	}
	for _, ct := range []string{"tidb", "tiproxy"} {
		selector, ok := gotGroupScoped[ct]
		require.True(t, ok, "no list call for component %q", ct)
		assert.Contains(t, selector, schemav1.LabelPingcapGroupKey+"=",
			"%s HPA selector must include the group label (got %q)", ct, selector)
		assert.Contains(t, selector, wantGroupSubstr[ct][:4],
			"%s HPA selector must include this group's identifier (got %q, want %q)",
			ct, selector, wantGroupSubstr[ct])
	}
	// Workers are SHARED_POOL-scoped, so CLUSTER scope must NOT query
	// them at all.
	for _, ct := range []string{entity.TiDBWorkerType, entity.TiKVWorkerType, entity.CoprocessorWorkerType} {
		_, ok := gotGroupScoped[ct]
		assert.False(t, ok,
			"CLUSTER scope must not issue a list call for worker component %q — workers are pool-scoped", ct)
	}
}

// TestReadRuntimeSpecsForSharedPool_ExcludesPerGroupTIDB pins the
// post-R1 contract: per-group TIDB HPAs are CLUSTER-only and must
// NOT appear in SHARED_POOL reads, even when the pool hosts multiple
// TiDB groups. SHARED_POOL now issues 3 per-component-type worker
// List calls (no TIDB call), so a mock that returns TIDB HPAs for
// every selector must produce no TIDB specs in the result.
func TestReadRuntimeSpecsForSharedPool_GroupScopingReturnsAllGroups(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}

	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-multi", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-multi", DataplaneName: "dp-sp-multi"},
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-multi").Return(kubeclient, nil).Once()

	// Return BOTH groups' TIDB HPAs (must be ignored by SHARED_POOL
	// scope under the new contract) plus a TIDB_WORKER HPA on the
	// TIDB_WORKER selector call.
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			require.Len(t, opts, 1)
			listOpts := opts[0].(*client.ListOptions)
			require.NotNil(t, listOpts.LabelSelector, "SHARED_POOL scope must use LabelSelector (per-component-type)")
			selStr := listOpts.LabelSelector.String()
			if selStr == "pingcap.com/component=worker-tidb" {
				hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
					{
						ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
							schemav1.LabelComponentKey: entity.TiDBWorkerType,
						}},
						Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()},
					},
				}
			}
			// TIKV_WORKER / COPROCESSOR_WORKER calls return empty.
			// If a stale selector returned TIDB HPAs (it can't, because
			// we never List for component=tidb at SHARED_POOL), they'd be
			// silently dropped by the per-component-type loop.
		}).
		Return(nil).Times(3)
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	specs, err := srv.readRuntimeSpecsForSharedPool(ctx, "sp-multi")
	require.NoError(t, err)
	// Only TIDB_WORKER is included; per-group TIDB HPAs from both
	// groups are CLUSTER-only and excluded by the SHARED_POOL
	// per-component-type selector.
	require.Len(t, specs, 2, "SHARED_POOL scope returns only TIDB_WORKER (SCALE_OUT + SCALE_IN); per-group TIDB excluded")
	for _, s := range specs {
		assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, s.GetTargetType(),
			"per-group TIDB must not leak into SHARED_POOL scope; got %s", s.GetTargetType().String())
	}
}

// TestListSharedPoolHPASpecs_EmptyGroupIDScopedList pins the
// SHARED_POOL-scope contract on the helper directly: empty groupID
// → per-component-type List with label selectors (mirrors the
// CLUSTER branch). Pool !HasSharedTiProxy → 3 List calls (3
// workers). Per-group TIDB/TIPROXY HPAs are CLUSTER scope only
// and must not appear here.
func TestListSharedPoolHPASpecs_EmptyGroupIDScopedList(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-pool", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-pool", DataplaneName: "dp-sp-pool"},
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-pool").Return(kubeclient, nil).Once()

	listCallCount := 0
	// Return a TIDB_WORKER HPA for every worker list call (the first
	// worker selector wins; subsequent worker calls return empty so we
	// don't double-count).
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			listCallCount++
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			require.Len(t, opts, 1, "ListOptions must always be passed (Namespace at minimum)")
			listOpts, ok := opts[0].(*client.ListOptions)
			require.True(t, ok)
			require.NotNil(t, listOpts.LabelSelector,
				"SHARED_POOL scope must use LabelSelector (per-component-type) so per-group TIDB/TIPROXY are excluded")
			sel, err := labels.Parse(listOpts.LabelSelector.String())
			require.NoError(t, err)
			// Only the 3 worker components are allowed; no group label.
			requirements, _ := sel.Requirements()
			hasComponent := false
			hasGroup := false
			for _, r := range requirements {
				switch r.Key() {
				case schemav1.LabelComponentKey:
					hasComponent = true
				case schemav1.LabelPingcapGroupKey:
					hasGroup = true
				}
			}
			require.True(t, hasComponent, "selector must include component label")
			require.False(t, hasGroup, "worker selector must NOT include group label (got %q)", listOpts.LabelSelector.String())
			if listCallCount == 1 {
				hpaList.Items = []autoscalingv2.HorizontalPodAutoscaler{
					{
						ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
							schemav1.LabelComponentKey: entity.TiDBWorkerType,
						}},
						Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()},
					},
				}
			}
		}).
		Return(nil).Times(3)
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-pool", "", "")
	require.NoError(t, err)
	assert.Equal(t, 3, listCallCount, "SHARED_POOL scope issues 3 worker List calls when pool !HasSharedTiProxy")
	require.Len(t, specs, 2, "single TIDB_WORKER HPA emits 2 rows (SCALE_OUT + SCALE_IN)")
	assert.Equal(t, apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER, specs[0].GetTargetType())
}

// TestListSharedPoolHPASpecs_NonEmptyGroupIDPerComponentSelector pins
// that the helper issues exactly one K8s List call per supported
// component type when groupID is set, with each call carrying the
// correct LabelSelector (group label for TIDB/TIPROXY, component
// label only for the 3 workers).
func TestListSharedPoolHPASpecs_NonEmptyGroupIDPerComponentSelector(t *testing.T) {
	ctx := context.Background()
	srv := &serverImpl{}
	sharedPoolRepo := repomnocks.NewSharedPoolRepo(t)
	dataplaneManager := dataplanemocks.NewManager(t)
	kubeclient := kubemocks.NewClient(t)
	sharedPoolRepo.On("Load", mock.Anything, "sp-grp", false).Return(&entity.SharedPool{
		SharedPool: apiv1.SharedPool{Id: "sp-grp", DataplaneName: "dp-sp-grp"},
	}, nil).Once()
	dataplaneManager.On("GetClient", "dp-sp-grp").Return(kubeclient, nil).Once()

	// Selector → HPAs that selector matches in the namespace.
	// Group-scoped selectors return that group's HPAs; component-only
	// selectors return pool-scoped workers.
	//
	// Component identity model:
	//   TIDB HPA    → `db-<tidbGroupID>`   (= grp-x label)
	//   TIPROXY HPA → `db-<tidbClusterID>` (= cluster-x label)
	matchedBySelector := func(sel labels.Selector) []autoscalingv2.HorizontalPodAutoscaler {
		grp := hpaGroupLabel("grp-x", apiv1.ComponentType_TIDB)
		cluster := hpaGroupLabel("cluster-x", apiv1.ComponentType_TIPROXY)
		candidates := []autoscalingv2.HorizontalPodAutoscaler{
			{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
				schemav1.LabelComponentKey:    "tidb",
				schemav1.LabelPingcapGroupKey: grp,
			}}, Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()}},
			{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
				schemav1.LabelComponentKey:    "tiproxy",
				schemav1.LabelPingcapGroupKey: cluster,
			}}, Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()}},
			{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
				schemav1.LabelComponentKey: entity.TiDBWorkerType,
			}}, Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()}},
			{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
				schemav1.LabelComponentKey: entity.TiKVWorkerType,
			}}, Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()}},
			{ObjectMeta: metav1.ObjectMeta{Labels: map[string]string{
				schemav1.LabelComponentKey: entity.CoprocessorWorkerType,
			}}, Spec: autoscalingv2.HorizontalPodAutoscalerSpec{Behavior: hpaBehaviorEnabled()}},
		}
		var matched []autoscalingv2.HorizontalPodAutoscaler
		for _, hpa := range candidates {
			if sel.Matches(labels.Set(hpa.GetLabels())) {
				matched = append(matched, hpa)
			}
		}
		return matched
	}

	seenSelectors := []string{}
	kubeclient.On("List", mock.Anything, mock.IsType(&autoscalingv2.HorizontalPodAutoscalerList{}), mock.Anything).
		Run(func(args mock.Arguments) {
			hpaList := args.Get(1).(*autoscalingv2.HorizontalPodAutoscalerList)
			opts := args.Get(2).([]client.ListOption)
			require.Len(t, opts, 1, "exactly one ListOption must be passed per call")
			listOpts, ok := opts[0].(*client.ListOptions)
			require.True(t, ok)
			require.NotNil(t, listOpts.LabelSelector, "every CLUSTER-scope List call must carry a LabelSelector")
			seenSelectors = append(seenSelectors, listOpts.LabelSelector.String())
			hpaList.Items = matchedBySelector(listOpts.LabelSelector)
		}).
		Return(nil).Maybe()
	srv.sharedPoolRepo = sharedPoolRepo
	srv.dataplaneManager = dataplaneManager

	specs, err := srv.listSharedPoolHPASpecs(ctx, "sp-grp", "cluster-x", "grp-x")
	require.NoError(t, err)

	// 2 List calls expected: TIDB and TIPROXY only. CLUSTER scope no
	// longer queries worker HPAs — workers belong to SHARED_POOL scope.
	require.Len(t, seenSelectors, 2,
		"non-empty groupID must issue exactly 2 List calls (TIDB + TIPROXY only)")

	// Every selector string must include pingcap.com/component=...
	for _, s := range seenSelectors {
		assert.Contains(t, s, schemav1.LabelComponentKey+"=",
			"every selector must filter on LabelComponentKey (got %q)", s)
	}

	// TIDB + TIPROXY selectors must also include the group label.
	groupSelectors := map[string]string{} // componentLabel → selector string
	for _, s := range seenSelectors {
		// Parse the selector string by splitting on ','. Each
		// segment is "k=v". Order isn't guaranteed so we index by
		// the pingcap.com/component value.
		var component string
		for _, seg := range splitSelector(s) {
			k, v, _ := splitKV(seg)
			if k == schemav1.LabelComponentKey {
				component = v
			}
		}
		groupSelectors[component] = s
	}

	for _, ct := range []string{"tidb", "tiproxy"} {
		sel, ok := groupSelectors[ct]
		require.True(t, ok, "no list call for component %q", ct)
		assert.Contains(t, sel, schemav1.LabelPingcapGroupKey+"=",
			"%s HPA selector must include the group label (got %q)", ct, sel)
	}
	// Workers are SHARED_POOL-scoped — CLUSTER scope must NOT issue
	// list calls for them.
	for _, ct := range []string{entity.TiDBWorkerType, entity.TiKVWorkerType, entity.CoprocessorWorkerType} {
		_, ok := groupSelectors[ct]
		assert.False(t, ok,
			"CLUSTER scope must not issue a list call for worker component %q — workers are pool-scoped", ct)
	}

	// Specs derived: 4 HPA-derived rows expected (TIDB emits 2 +
	// TIPROXY emits 2 = 4). Workers are SHARED_POOL-scoped and not
	// surfaced at CLUSTER scope.
	require.Len(t, specs, 4,
		"TIDB + TIPROXY HPAs each emit 2 rows (SCALE_OUT + SCALE_IN) at CLUSTER scope")
	seen := map[apiv1.AutoscaleTargetType]int{}
	for _, s := range specs {
		seen[s.GetTargetType()]++
	}
	assert.Equal(t, 2, seen[apiv1.AutoscaleTargetType_TARGET_TIDB],
		"TIDB HPA now emits 2 rows (SCALE_OUT + SCALE_IN) to match the CLUSTER write-side allowlist")
	assert.Equal(t, 2, seen[apiv1.AutoscaleTargetType_TARGET_TIPROXY],
		"TIPROXY HPA now emits 2 rows (SCALE_OUT + SCALE_IN) to match the CLUSTER write-side allowlist")
	assert.Zero(t, seen[apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER],
		"worker HPAs are SHARED_POOL-scoped; CLUSTER scope must not surface them")
	assert.Zero(t, seen[apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER],
		"worker HPAs are SHARED_POOL-scoped; CLUSTER scope must not surface them")
	assert.Zero(t, seen[apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER],
		"worker HPAs are SHARED_POOL-scoped; CLUSTER scope must not surface them")
}

// splitSelector splits a labels.Selector.String() output by ",".
// labels.Selector.String() format: "a=b,c=d,e=f". Items containing
// "==" / "!=" / " in " etc. are also comma-separated; splitting by
// "," is safe for the selectors this test inspects because the
// fixture selectors only use equality (k=v) — see componentLabelForHPA
// and groupLabel in SetHPAScaleBehavior's write path.
func splitSelector(s string) []string {
	out := []string{}
	cur := ""
	for i := 0; i < len(s); i++ {
		if s[i] == ',' {
			out = append(out, cur)
			cur = ""
			continue
		}
		cur += string(s[i])
	}
	if cur != "" {
		out = append(out, cur)
	}
	return out
}

// splitKV splits "k=v" — no escaping handling, sufficient for the
// label keys in these tests (k8s label keys never contain "=").
func splitKV(s string) (string, string, bool) {
	for i := 0; i < len(s); i++ {
		if s[i] == '=' {
			return s[:i], s[i+1:], true
		}
	}
	return "", "", false
}