package spec

import (
	"testing"

	"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"
	"github.com/tidbcloud/cluster-service-ng/pkg/storage/repo"
)

func TestSpecToFeatureFlags(t *testing.T) {
	makeSpec := func(ct apiv1.AutoscaleTargetType, dir apiv1.AutoscalePolicy_Direction, cap apiv1.AutoscalePolicy_Capability, trig apiv1.AutoscalePolicy_Trigger, state apiv1.AutoscalePolicy_State) *entity.AutoscalePolicySpec {
		return &entity.AutoscalePolicySpec{
			TargetType: ct,
			Direction:  dir,
			Capability: cap,
			Trigger:    trig,
			State:      state,
		}
	}

	t.Run("all TIKV SCALE_OUT specs map to correct flags", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_DISK_SIZE, apiv1.AutoscalePolicy_DISK, apiv1.AutoscalePolicy_ENABLED),
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_IOPS, apiv1.AutoscalePolicy_IO, apiv1.AutoscalePolicy_ENABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.True(t, patch.HasAnyChange())

		require.NotNil(t, patch.EnableAutoscaleTikvCpuScaleOut)
		require.True(t, *patch.EnableAutoscaleTikvCpuScaleOut)

		require.NotNil(t, patch.EnableAutoscaleDiskExpand)
		require.True(t, *patch.EnableAutoscaleDiskExpand)

		require.NotNil(t, patch.EnableAutoscaleIopsThroughputExpand)
		require.True(t, *patch.EnableAutoscaleIopsThroughputExpand)
		// SCALE_IN for TIKV and any SCALE_IN/SCALE_OUT for
		// non-TIKV must NOT touch the RCU flags.
		require.Nil(t, patch.EnableScaleSharedPoolByRu)
		require.Nil(t, patch.EnableScaleTidbGroupByRu)
	})

	t.Run("disabled state maps to false flags", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_DISABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.NotNil(t, patch.EnableAutoscaleTikvCpuScaleOut)
		require.False(t, *patch.EnableAutoscaleTikvCpuScaleOut)
		// RCU flags must NOT be touched by TIKV SCALE_OUT specs.
		require.Nil(t, patch.EnableScaleSharedPoolByRu)
	})

	t.Run("unspecified state is silently dropped", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_STATE_UNSPECIFIED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.False(t, patch.HasAnyChange())
		require.Nil(t, patch.EnableAutoscaleTikvCpuScaleOut)
	})

	t.Run("nil spec is skipped", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{nil, makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED)}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.NotNil(t, patch.EnableAutoscaleTikvCpuScaleOut)
		require.True(t, *patch.EnableAutoscaleTikvCpuScaleOut)
	})

	t.Run("unrecognized specs are reported", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIDB, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Len(t, unrecognized, 1)
		require.False(t, patch.HasAnyChange())
	})

	t.Run("worker SCALE_IN is unrecognized (HPA-controlled)", func(t *testing.T) {
		workers := []apiv1.AutoscaleTargetType{
			apiv1.AutoscaleTargetType_TARGET_TIKV_WORKER,
			apiv1.AutoscaleTargetType_TARGET_TIDB_WORKER,
			apiv1.AutoscaleTargetType_TARGET_COPROCESSOR_WORKER,
		}
		for _, ct := range workers {
			specs := []*entity.AutoscalePolicySpec{
				makeSpec(ct, apiv1.AutoscalePolicy_SCALE_IN, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
			}

			patch, unrecognized := SpecToFeatureFlags(specs)
			require.Len(t, unrecognized, 1, "component %s SCALE_IN must be unrecognized", ct.String())
			require.False(t, patch.HasAnyChange(), "no flag should be set for %s SCALE_IN", ct.String())
		}
	})

	t.Run("TIPROXY SCALE_IN maps to enable_autoscale_tiproxy_scale_in", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIPROXY, apiv1.AutoscalePolicy_SCALE_IN, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.NotNil(t, patch.EnableAutoscaleTiproxyScaleIn)
		require.True(t, *patch.EnableAutoscaleTiproxyScaleIn)
		require.Nil(t, patch.EnableAutoscaleTikvCpuScaleOut)
	})

	t.Run("TIPROXY SCALE_IN maps regardless of trigger", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIPROXY, apiv1.AutoscalePolicy_SCALE_IN, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_IO, apiv1.AutoscalePolicy_ENABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.NotNil(t, patch.EnableAutoscaleTiproxyScaleIn)
		require.True(t, *patch.EnableAutoscaleTiproxyScaleIn)
	})

	t.Run("TIPROXY SCALE_OUT is unrecognized (HPA-controlled)", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIPROXY, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Len(t, unrecognized, 1)
		require.False(t, patch.HasAnyChange())
	})

	t.Run("throughput maps to iops throughput flag", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_THROUGHPUT, apiv1.AutoscalePolicy_DISK, apiv1.AutoscalePolicy_ENABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.NotNil(t, patch.EnableAutoscaleIopsThroughputExpand)
		require.True(t, *patch.EnableAutoscaleIopsThroughputExpand)
	})

	t.Run("multiple specs same flag last-writer-wins", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_DISABLED),
		}

		patch, unrecognized := SpecToFeatureFlags(specs)
		require.Empty(t, unrecognized)
		require.NotNil(t, patch.EnableAutoscaleTikvCpuScaleOut)
		require.False(t, *patch.EnableAutoscaleTikvCpuScaleOut)
	})

	t.Run("empty input produces empty patch", func(t *testing.T) {
		patch, unrecognized := SpecToFeatureFlags(nil)
		require.Empty(t, unrecognized)
		require.False(t, patch.HasAnyChange())
	})
}

func TestMergeSpecs(t *testing.T) {
	enabledSpec := func(ct apiv1.AutoscaleTargetType) *entity.AutoscalePolicySpec {
		return &entity.AutoscalePolicySpec{
			TargetType: ct,
			Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
			State:      apiv1.AutoscalePolicy_ENABLED,
		}
	}
	disabledSpec := func(ct apiv1.AutoscaleTargetType) *entity.AutoscalePolicySpec {
		return &entity.AutoscalePolicySpec{
			TargetType: ct,
			Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
			State:      apiv1.AutoscalePolicy_DISABLED,
		}
	}
	makeDiskSpec := func() *entity.AutoscalePolicySpec {
		return &entity.AutoscalePolicySpec{
			TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
			Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
			Capability: apiv1.AutoscalePolicy_DISK_SIZE,
			Trigger:    apiv1.AutoscalePolicy_DISK,
			State:      apiv1.AutoscalePolicy_ENABLED,
		}
	}

	t.Run("ReplaceMode with nil base returns patch", func(t *testing.T) {
		patch := []*entity.AutoscalePolicySpec{enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		result := MergeSpecs(nil, patch, MergeModeReplace)
		require.Len(t, result, 1)
		require.Equal(t, apiv1.AutoscalePolicy_ENABLED, result[0].GetState())
	})

	t.Run("ReplaceMode drops unspecified state specs", func(t *testing.T) {
		patch := []*entity.AutoscalePolicySpec{
			enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV),
			{
				TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
				Direction:  apiv1.AutoscalePolicy_SCALE_IN,
				Capability: apiv1.AutoscalePolicy_NODE_COUNT,
				Trigger:    apiv1.AutoscalePolicy_RCU,
				State:      apiv1.AutoscalePolicy_STATE_UNSPECIFIED,
			},
		}
		result := MergeSpecs(nil, patch, MergeModeReplace)
		require.Len(t, result, 1)
	})

	t.Run("ReplaceMode replaces base entirely", func(t *testing.T) {
		base := []*entity.AutoscalePolicySpec{enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		patch := []*entity.AutoscalePolicySpec{disabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		result := MergeSpecs(base, patch, MergeModeReplace)
		require.Len(t, result, 1)
		require.Equal(t, apiv1.AutoscalePolicy_DISABLED, result[0].GetState())
	})

	t.Run("MergeMode merges per-key", func(t *testing.T) {
		base := []*entity.AutoscalePolicySpec{enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		patch := []*entity.AutoscalePolicySpec{disabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		result := MergeSpecs(base, patch, MergeModeMerge)
		require.Len(t, result, 1)
		require.Equal(t, apiv1.AutoscalePolicy_DISABLED, result[0].GetState())
	})

	t.Run("MergeMode adds new keys from patch", func(t *testing.T) {
		base := []*entity.AutoscalePolicySpec{enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		patch := []*entity.AutoscalePolicySpec{makeDiskSpec()}
		result := MergeSpecs(base, patch, MergeModeMerge)
		require.Len(t, result, 2)
	})

	t.Run("MergeMode preserves base keys not in patch", func(t *testing.T) {
		base := []*entity.AutoscalePolicySpec{
			enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV),
			makeDiskSpec(),
		}
		patch := []*entity.AutoscalePolicySpec{disabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		result := MergeSpecs(base, patch, MergeModeMerge)
		require.Len(t, result, 2)

		stateMap := make(map[string]apiv1.AutoscalePolicy_State)
		for _, s := range result {
			stateMap[specKey(s)] = s.GetState()
		}
		require.Equal(t, apiv1.AutoscalePolicy_DISABLED, stateMap[specKey(enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV))])
		require.Equal(t, apiv1.AutoscalePolicy_ENABLED, stateMap[specKey(makeDiskSpec())])
	})

	t.Run("MergeMode drops nil and unspecified from patch", func(t *testing.T) {
		base := []*entity.AutoscalePolicySpec{enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		patch := []*entity.AutoscalePolicySpec{
			nil,
			{
				TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
				Direction:  apiv1.AutoscalePolicy_SCALE_IN,
				Capability: apiv1.AutoscalePolicy_NODE_COUNT,
				Trigger:    apiv1.AutoscalePolicy_RCU,
				State:      apiv1.AutoscalePolicy_STATE_UNSPECIFIED,
			},
		}
		result := MergeSpecs(base, patch, MergeModeMerge)
		require.Len(t, result, 1)
	})

	t.Run("empty base MergeMode returns compacted patch", func(t *testing.T) {
		patch := []*entity.AutoscalePolicySpec{enabledSpec(apiv1.AutoscaleTargetType_TARGET_TIKV)}
		result := MergeSpecs(nil, patch, MergeModeMerge)
		require.Len(t, result, 1)
	})
}

func TestMergeSpecsSortedOutput(t *testing.T) {
	specs := []*entity.AutoscalePolicySpec{
		{
			TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
			Direction:  apiv1.AutoscalePolicy_SCALE_IN,
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_RCU,
			State:      apiv1.AutoscalePolicy_ENABLED,
		},
		{
			TargetType: apiv1.AutoscaleTargetType_TARGET_TIKV,
			Direction:  apiv1.AutoscalePolicy_SCALE_OUT,
			Capability: apiv1.AutoscalePolicy_NODE_COUNT,
			Trigger:    apiv1.AutoscalePolicy_CPU,
			State:      apiv1.AutoscalePolicy_ENABLED,
		},
	}

	for i := 0; i < 10; i++ {
		result := MergeSpecs(nil, specs, MergeModeReplace)
		require.Len(t, result, 2)
		require.Equal(t, apiv1.AutoscalePolicy_SCALE_IN, result[0].GetDirection())
		require.Equal(t, apiv1.AutoscalePolicy_SCALE_OUT, result[1].GetDirection())
	}
}

func TestFeatureFlagsPatch(t *testing.T) {
	t.Run("HasAnyChange on nil patch", func(t *testing.T) {
		var p *repo.FeatureFlagsPatch
		require.False(t, p.HasAnyChange())
	})

	t.Run("HasAnyChange on empty patch", func(t *testing.T) {
		p := repo.FeatureFlagsPatch{}
		require.False(t, p.HasAnyChange())
	})

	t.Run("HasAnyChange with one field set", func(t *testing.T) {
		b := true
		p := repo.FeatureFlagsPatch{EnableAutoscaleTikvCpuScaleOut: &b}
		require.True(t, p.HasAnyChange())
	})

	t.Run("HasAnyChange with false field is still a change", func(t *testing.T) {
		b := false
		p := repo.FeatureFlagsPatch{EnableScaleSharedPoolByRu: &b}
		require.True(t, p.HasAnyChange())
	})
}

func TestSpecsToFeatureFlagsBlock(t *testing.T) {
	makeSpec := func(ct apiv1.AutoscaleTargetType, dir apiv1.AutoscalePolicy_Direction, cap apiv1.AutoscalePolicy_Capability, trig apiv1.AutoscalePolicy_Trigger, state apiv1.AutoscalePolicy_State) *entity.AutoscalePolicySpec {
		return &entity.AutoscalePolicySpec{
			TargetType: ct,
			Direction:  dir,
			Capability: cap,
			Trigger:    trig,
			State:      state,
		}
	}

	t.Run("empty input produces an all-nil block (wipe contract)", func(t *testing.T) {
		block := SpecsToFeatureFlagsBlock(nil)
		require.NotNil(t, block)
		require.Nil(t, block.EnableScaleSharedPoolByRu)
		require.Nil(t, block.EnableScaleTidbGroupByRu)
		require.Nil(t, block.EnableAutoscaleTikvCpuScaleOut)
		require.Nil(t, block.EnableAutoscaleDiskExpand)
		require.Nil(t, block.EnableAutoscaleIopsThroughputExpand)
		require.Nil(t, block.EnableAutoscaleTiproxyScaleIn)
	})

	t.Run("one spec sets exactly one flag; the other 5 stay nil", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableAutoscaleTikvCpuScaleOut)
		require.True(t, *block.EnableAutoscaleTikvCpuScaleOut)
		require.Nil(t, block.EnableScaleSharedPoolByRu)
		require.Nil(t, block.EnableScaleTidbGroupByRu)
		require.Nil(t, block.EnableAutoscaleDiskExpand)
		require.Nil(t, block.EnableAutoscaleIopsThroughputExpand)
		require.Nil(t, block.EnableAutoscaleTiproxyScaleIn)
	})

	t.Run("3 TIKV SCALE_OUT specs set the 3 TIKV flags", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_DISK_SIZE, apiv1.AutoscalePolicy_DISK, apiv1.AutoscalePolicy_ENABLED),
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_IOPS, apiv1.AutoscalePolicy_IO, apiv1.AutoscalePolicy_ENABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableAutoscaleTikvCpuScaleOut)
		require.NotNil(t, block.EnableAutoscaleDiskExpand)
		require.NotNil(t, block.EnableAutoscaleIopsThroughputExpand)
		// SCALE_IN for TIKV and any SCALE_IN/SCALE_OUT for
		// non-TIKV must NOT touch the RCU flags — those flags
		// are RCU quota controls, not HPA scaleOut / scaleIn
		// master switches.
		require.Nil(t, block.EnableScaleSharedPoolByRu)
		require.Nil(t, block.EnableScaleTidbGroupByRu)
	})

	t.Run("disabled state maps to explicit false", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_DISABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableAutoscaleTikvCpuScaleOut)
		require.False(t, *block.EnableAutoscaleTikvCpuScaleOut)
	})

	t.Run("unspecified state is dropped (flag stays nil)", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_STATE_UNSPECIFIED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.Nil(t, block.EnableAutoscaleTikvCpuScaleOut)
		// All 6 flags are nil — the block is the "wipe" shape.
		require.Nil(t, block.EnableScaleSharedPoolByRu)
		require.Nil(t, block.EnableScaleTidbGroupByRu)
		require.Nil(t, block.EnableAutoscaleDiskExpand)
		require.Nil(t, block.EnableAutoscaleIopsThroughputExpand)
		require.Nil(t, block.EnableAutoscaleTiproxyScaleIn)
	})

	t.Run("nil spec is skipped", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{nil, makeSpec(apiv1.AutoscaleTargetType_TARGET_TIKV, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED)}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableAutoscaleTikvCpuScaleOut)
		require.True(t, *block.EnableAutoscaleTikvCpuScaleOut)
	})

	t.Run("SHARED_POOL virtual component maps to enable_scale_shared_pool_by_ru", func(t *testing.T) {
		// SCALE_OUT direction
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_SHARED_POOL, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, apiv1.AutoscalePolicy_ENABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableScaleSharedPoolByRu)
		require.True(t, *block.EnableScaleSharedPoolByRu)
		require.Nil(t, block.EnableScaleTidbGroupByRu)
		require.Nil(t, block.EnableAutoscaleTikvCpuScaleOut)
		require.Nil(t, block.EnableAutoscaleDiskExpand)
		require.Nil(t, block.EnableAutoscaleIopsThroughputExpand)
	})

	t.Run("SHARED_POOL SCALE_IN also maps to enable_scale_shared_pool_by_ru", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_SHARED_POOL, apiv1.AutoscalePolicy_SCALE_IN, apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, apiv1.AutoscalePolicy_DISABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableScaleSharedPoolByRu)
		require.False(t, *block.EnableScaleSharedPoolByRu)
	})

	t.Run("TIDB_GROUP virtual component maps to enable_scale_tidb_group_by_ru", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_TIDB_GROUP, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_SPEC, apiv1.AutoscalePolicy_RCU, apiv1.AutoscalePolicy_ENABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.NotNil(t, block.EnableScaleTidbGroupByRu)
		require.True(t, *block.EnableScaleTidbGroupByRu)
		require.Nil(t, block.EnableScaleSharedPoolByRu)
	})

	t.Run("SHARED_POOL with wrong capability/trigger is unrecognized", func(t *testing.T) {
		specs := []*entity.AutoscalePolicySpec{
			makeSpec(apiv1.AutoscaleTargetType_SHARED_POOL, apiv1.AutoscalePolicy_SCALE_OUT, apiv1.AutoscalePolicy_NODE_COUNT, apiv1.AutoscalePolicy_CPU, apiv1.AutoscalePolicy_ENABLED),
		}
		block := SpecsToFeatureFlagsBlock(specs)
		require.Nil(t, block.EnableScaleSharedPoolByRu)
		require.Nil(t, block.EnableScaleTidbGroupByRu)
	})
}