package server

import (
	"context"
	"fmt"
	"strings"
	"testing"

	"github.com/stretchr/testify/assert"
	"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"
	mysqlschema "github.com/tidbcloud/cluster-service-ng/pkg/storage/mysql/schema"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/proto"
)

func TestGenTiKVNodeCount(t *testing.T) {
	cases := []struct {
		in   uint32
		want uint32
		name string
	}{
		{in: 1, want: 3, name: "round up below 3"},
		{in: 2, want: 3, name: "round up to 3"},
		{in: 3, want: 3, name: "exact multiple stays"},
		{in: 4, want: 3, name: "4 -> 3 (round down)"},
		{in: 5, want: 3, name: "5 -> 3 (round down)"},
		{in: 6, want: 6, name: "6 stays 6"},
		{in: 7, want: 6, name: "7 -> 6 (round down)"},
		{in: 8, want: 6, name: "8 -> 6 (round down)"},
		{in: 9, want: 9, name: "9 stays 9"},
	}

	for _, tc := range cases {
		if got := genTiKVNodeCount(tc.in); got != tc.want {
			t.Errorf("%s: genTiKVNodeCount(%d) = %d, want %d", tc.name, tc.in, got, tc.want)
		}
	}
}

func TestCompareNodeQuota(t *testing.T) {
	makeSpec := func(cpu, mem uint32) *entity.NodeSpec { return &entity.NodeSpec{Cpu: cpu, MemoryGi: mem} }
	newQuota := func(count, max, min uint32, spec *entity.NodeSpec) *entity.NodeQuota {
		return &entity.NodeQuota{NodeCount: count, MaxNodeCount: max, MinNodeCount: min, NodeSpec: spec}
	}

	// Case 1: curr is nil -> use new
	{
		curr := (*entity.NodeQuota)(nil)
		newQ := newQuota(3, 5, 1, makeSpec(8, 16))
		got := compareNodeQuota(curr, newQ)
		if got != newQ {
			t.Fatalf("expected new quota to be returned when curr is nil")
		}
	}

	// Case 2: new spec is larger -> return new with NodeCount >= curr.NodeCount
	{
		curr := newQuota(4, 4, 1, makeSpec(4, 8))
		newQ := newQuota(2, 3, 1, makeSpec(8, 16))
		got := compareNodeQuota(curr, newQ)
		if got.NodeSpec.Cpu != 8 || got.NodeSpec.MemoryGi != 16 {
			t.Fatalf("expected new spec to be used")
		}
		if got.NodeCount != 4 {
			t.Fatalf("expected node count to be max(curr,new) = 4, got %d", got.NodeCount)
		}
		if got.MaxNodeCount != 4 {
			t.Fatalf("expected max node count adjusted to %d, got %d", 4, got.MaxNodeCount)
		}
	}

	// Case 3: new spec not larger -> keep curr, but NodeCount = max(curr,new)
	{
		curr := newQuota(3, 5, 1, makeSpec(8, 16))
		newQ := newQuota(4, 2, 1, makeSpec(4, 8))
		got := compareNodeQuota(curr, newQ)
		if got.NodeSpec.Cpu != 8 || got.NodeSpec.MemoryGi != 16 {
			t.Fatalf("expected current spec to be kept")
		}
		if got.NodeCount != 4 {
			t.Fatalf("expected node count to be max(curr,new)=4, got %d", got.NodeCount)
		}
		if got.MaxNodeCount != 5 {
			t.Fatalf("max node count should remain from curr unless exceeded, got %d", got.MaxNodeCount)
		}
	}

	// Case 4: node count exceeds max -> adjust max
	{
		curr := newQuota(3, 5, 1, makeSpec(8, 16))
		newQ := newQuota(7, 6, 1, makeSpec(4, 8))
		got := compareNodeQuota(curr, newQ)
		if got.NodeCount != 7 {
			t.Fatalf("expected node count to be max(curr,new)=7, got %d", got.NodeCount)
		}
		if got.MaxNodeCount != 7 {
			t.Fatalf("expected max node count to be adjusted to 7, got %d", got.MaxNodeCount)
		}
	}
}

func TestGetNodeSpec(t *testing.T) {
	// Prepare desired key and candidate lists
	desired := &entity.NodeSpec{Cpu: 8, MemoryGi: 16}
	key := entity.GetComponentNodeSpec(desired)

	// Two variants: with and without local disk
	withLocal := &entity.NodeSpec{Cpu: 8, MemoryGi: 16, WithLocalDisk: true, LocalDiskSizeGi: 500, AvailableCpu: 8, AvailableMemoryMi: 16000}
	withoutLocal := &entity.NodeSpec{Cpu: 8, MemoryGi: 16, WithLocalDisk: false, AvailableCpu: 8, AvailableMemoryMi: 16000}

	// Map with both options
	nodeSpecMap := map[string][]*entity.NodeSpec{
		key: {withLocal, withoutLocal},
	}

	ctx := context.Background()
	// Expect choose without local when withLocalDisk=false
	if got, err := getNodeSpec(ctx, desired, nodeSpecMap, false); err != nil {
		t.Fatalf("unexpected error: %v", err)
	} else if got.WithLocalDisk {
		t.Fatalf("expected choose non-local disk spec when withLocalDisk=false")
	}

	// Expect choose with local when withLocalDisk=true
	if got, err := getNodeSpec(ctx, desired, nodeSpecMap, true); err != nil {
		t.Fatalf("unexpected error: %v", err)
	} else if !got.WithLocalDisk {
		t.Fatalf("expected choose local disk spec when withLocalDisk=true")
	}

	// Prefer local when requested even if non-local appears first
	nodeSpecMap[key] = []*entity.NodeSpec{withoutLocal, withLocal}
	if got, err := getNodeSpec(ctx, desired, nodeSpecMap, true); err != nil {
		t.Fatalf("unexpected error: %v", err)
	} else if !got.WithLocalDisk {
		t.Fatalf("expected local disk spec when withLocalDisk=true")
	}

	// Invalid: withLocalDisk=true but LocalDiskSizeGi=0
	badLocal := &entity.NodeSpec{Cpu: 8, MemoryGi: 16, WithLocalDisk: true, LocalDiskSizeGi: 0, AvailableCpu: 8, AvailableMemoryMi: 16000}
	nodeSpecMap[key] = []*entity.NodeSpec{badLocal}
	if _, err := getNodeSpec(ctx, desired, nodeSpecMap, true); err == nil {
		t.Fatalf("expected error for invalid local disk size")
	}

	// Invalid: withLocalDisk=true but only non-local candidates exist
	nodeSpecMap[key] = []*entity.NodeSpec{withoutLocal}
	if _, err := getNodeSpec(ctx, desired, nodeSpecMap, true); err == nil {
		t.Fatalf("expected error when requesting local but no local spec available")
	}

	// Invalid: no available CPU/memory
	invalid := &entity.NodeSpec{Cpu: 8, MemoryGi: 16, WithLocalDisk: false, AvailableCpu: 0, AvailableMemoryMi: 0}
	nodeSpecMap[key] = []*entity.NodeSpec{invalid}
	if _, err := getNodeSpec(ctx, desired, nodeSpecMap, false); err == nil {
		t.Fatalf("expected error when no available resources")
	}

	// Invalid: no available CPU/memory (local variant)
	invalidLocal := &entity.NodeSpec{Cpu: 8, MemoryGi: 16, WithLocalDisk: true, LocalDiskSizeGi: 500, AvailableCpu: 0, AvailableMemoryMi: 0}
	nodeSpecMap[key] = []*entity.NodeSpec{invalidLocal}
	if _, err := getNodeSpec(ctx, desired, nodeSpecMap, true); err == nil {
		t.Fatalf("expected error when no available resources for local spec")
	}

	// Not found key
	if _, err := getNodeSpec(ctx, desired, map[string][]*entity.NodeSpec{}, false); err == nil {
		t.Fatalf("expected error when node spec key missing")
	}

	// Nil desired
	if _, err := getNodeSpec(ctx, nil, nodeSpecMap, false); err == nil {
		t.Fatalf("expected error when desired node spec is nil")
	}
}

func TestGeneratedSharedPoolComponent_PDDeploymentMode(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	tests := []struct {
		name                  string
		pdDeploymentMode      apiv1.PDDeploymentMode
		enableResourceManager bool
		wantPDTso             bool
		wantPDScheduling      bool
		wantResourceManager   bool
	}{
		{
			name:             "unspecified defaults to microservice",
			pdDeploymentMode: apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_UNSPECIFIED,
			wantPDTso:        true,
			wantPDScheduling: true,
		},
		{
			name:                  "monolithic skips pd microservice components",
			pdDeploymentMode:      apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC,
			enableResourceManager: true,
			wantPDTso:             false,
			wantPDScheduling:      false,
			wantResourceManager:   false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:               "sp1",
				CapacityMode:     apiv1.CapacityMode_RCU_DRIVEN,
				PdDeploymentMode: tt.pdDeploymentMode,
				Provider:         &apiv1.Provider{Type: apiv1.ProviderType_AWS},
			}}
			ruConfig := buildGeneratedSharedPoolComponentRuConfig()
			ruConfig.FeatureFlags = &apiv1.RuConfig_FeatureFlags{
				EnableResourceManager: tt.enableResourceManager,
			}
			nodeSpecMap := map[string][]*entity.NodeSpec{
				"8C16G": {
					{
						Cpu:               8,
						MemoryGi:          16,
						AvailableCpu:      8,
						AvailableMemoryMi: 16384,
					},
				},
				"2C4G": {
					{
						Cpu:               2,
						MemoryGi:          4,
						AvailableCpu:      2,
						AvailableMemoryMi: 4096,
					},
				},
			}
			initConfig := &entity.InitSharedPoolConfig{
				TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiKVDataStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
				TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiKVRaftlogStorageSize, Iops: 5000, ThroughputMbps: 250,
				},
				TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiFlashCNStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
				TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiFlashWNStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
			}

			err := s.generatedSharedPoolComponent(
				ctx,
				sharedPool,
				&entity.RequestUnit{Min: 100000, Max: 200000},
				nodeSpecMap,
				ruConfig,
				initConfig,
				false,
			)
			if err != nil {
				t.Fatalf("generatedSharedPoolComponent() error = %v", err)
			}

			gotTypes := map[apiv1.ComponentType]bool{}
			for _, comp := range sharedPool.GetComponentGroups() {
				gotTypes[comp.GetComponentType()] = true
			}

			if !gotTypes[apiv1.ComponentType_PD] {
				t.Fatalf("expected PD component")
			}
			if gotTypes[apiv1.ComponentType_PD_TSO] != tt.wantPDTso {
				t.Fatalf("PD_TSO presence = %v, want %v", gotTypes[apiv1.ComponentType_PD_TSO], tt.wantPDTso)
			}
			if gotTypes[apiv1.ComponentType_PD_SCHEDULING] != tt.wantPDScheduling {
				t.Fatalf("PD_SCHEDULING presence = %v, want %v", gotTypes[apiv1.ComponentType_PD_SCHEDULING], tt.wantPDScheduling)
			}
			if gotTypes[apiv1.ComponentType_RESOURCE_MANAGER] != tt.wantResourceManager {
				t.Fatalf("RESOURCE_MANAGER presence = %v, want %v", gotTypes[apiv1.ComponentType_RESOURCE_MANAGER], tt.wantResourceManager)
			}
		})
	}
}

func TestGeneratedSharedPoolComponent_EnabledTiKVRemoteWorkers(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	tests := []struct {
		name                  string
		enabledWorkers        []apiv1.ComponentType
		wantCoprocessorWorker bool
	}{
		{
			name:                  "empty keeps coprocessor worker",
			wantCoprocessorWorker: true,
		},
		{
			name: "tikv and coprocessor keeps coprocessor worker",
			enabledWorkers: []apiv1.ComponentType{
				apiv1.ComponentType_COPROCESSOR_WORKER,
				apiv1.ComponentType_TIKV_WORKER,
			},
			wantCoprocessorWorker: true,
		},
		{
			name: "tikv only skips coprocessor worker",
			enabledWorkers: []apiv1.ComponentType{
				apiv1.ComponentType_TIKV_WORKER,
			},
			wantCoprocessorWorker: false,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:                       "sp1",
				CapacityMode:             apiv1.CapacityMode_RCU_DRIVEN,
				EnabledTikvRemoteWorkers: tt.enabledWorkers,
				PdDeploymentMode:         apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC,
				Provider:                 &apiv1.Provider{Type: apiv1.ProviderType_AWS},
			}}
			ruConfig := buildGeneratedSharedPoolComponentRuConfig()
			nodeSpecMap := map[string][]*entity.NodeSpec{
				"8C16G": {
					{
						Cpu:               8,
						MemoryGi:          16,
						AvailableCpu:      8,
						AvailableMemoryMi: 16384,
					},
				},
			}
			initConfig := &entity.InitSharedPoolConfig{
				TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiKVDataStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
				TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiKVRaftlogStorageSize, Iops: 5000, ThroughputMbps: 250,
				},
				TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiFlashCNStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
				TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiFlashWNStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
			}

			err := s.generatedSharedPoolComponent(
				ctx,
				sharedPool,
				&entity.RequestUnit{Min: 100000, Max: 200000},
				nodeSpecMap,
				ruConfig,
				initConfig,
				false,
			)
			if err != nil {
				t.Fatalf("generatedSharedPoolComponent() error = %v", err)
			}

			gotTypes := map[apiv1.ComponentType]bool{}
			for _, comp := range sharedPool.GetComponentGroups() {
				gotTypes[comp.GetComponentType()] = true
			}

			if !gotTypes[apiv1.ComponentType_TIKV_WORKER] {
				t.Fatalf("expected TIKV_WORKER component")
			}
			if gotTypes[apiv1.ComponentType_COPROCESSOR_WORKER] != tt.wantCoprocessorWorker {
				t.Fatalf("COPROCESSOR_WORKER presence = %v, want %v",
					gotTypes[apiv1.ComponentType_COPROCESSOR_WORKER],
					tt.wantCoprocessorWorker,
				)
			}
		})
	}
}

func TestGeneratedSharedPoolComponent_TiflashColumnar(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	tests := []struct {
		name                           string
		deployType                     string
		tiflashMode                    string
		tiflashComputeDefaultNodeCount *uint32
		wantTiflashComputeNode         uint32
		wantSchemaManagerNode          uint32
		wantTiflashWrite               bool
		wantSchemaManager              bool
	}{
		{
			name:                           "columnar pool starts schema manager at zero when tiflash compute is zero",
			tiflashMode:                    mysqlschema.SharedPoolTiflashModeColumnar,
			tiflashComputeDefaultNodeCount: uint32Ptr(0),
			wantTiflashComputeNode:         0,
			wantSchemaManagerNode:          0,
			wantSchemaManager:              true,
		},
		{
			name:                   "columnar pool starts schema manager when tiflash compute starts",
			tiflashMode:            mysqlschema.SharedPoolTiflashModeColumnar,
			wantTiflashComputeNode: 1,
			wantSchemaManagerNode:  1,
			wantSchemaManager:      true,
		},
		{
			name:                   "legacy pool keeps tiflash write",
			wantTiflashComputeNode: 1,
			wantTiflashWrite:       true,
		},
		{
			name:                   "shared tiproxy columnar pool follows the same on-demand policy",
			deployType:             mysqlschema.SharedPoolDeployTypeSharedTiProxy,
			tiflashMode:            mysqlschema.SharedPoolTiflashModeColumnar,
			wantTiflashComputeNode: 1,
			wantSchemaManagerNode:  1,
			wantSchemaManager:      true,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			sharedPool := &entity.SharedPool{SharedPool: apiv1.SharedPool{
				Id:               "sp1",
				CapacityMode:     apiv1.CapacityMode_RCU_DRIVEN,
				PdDeploymentMode: apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC,
				Provider:         &apiv1.Provider{Type: apiv1.ProviderType_AWS},
			}}
			sharedPool.DeployType = tt.deployType
			sharedPool.TiflashMode = tt.tiflashMode

			ruConfig := buildGeneratedSharedPoolComponentRuConfig()
			ruConfig.Components["tiflash_compute"].NodeSpecRuRange["8c16g"].DefaultNodeCount = tt.tiflashComputeDefaultNodeCount
			if tt.deployType != "" {
				ruConfig.Components["tiproxy"] = &apiv1.RuConfig_ComponentsMap{
					MaxNodeCount: 10,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"8c16g": {Min: 0, Max: 0},
					},
				}
			}
			nodeSpecMap := map[string][]*entity.NodeSpec{
				"8C16G": {{Cpu: 8, MemoryGi: 16, AvailableCpu: 8, AvailableMemoryMi: 16384}},
				"2C8G":  {{Cpu: 2, MemoryGi: 8, AvailableCpu: 2, AvailableMemoryMi: 8192}},
			}
			initConfig := &entity.InitSharedPoolConfig{
				TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiKVDataStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
				TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiKVRaftlogStorageSize, Iops: 5000, ThroughputMbps: 250,
				},
				TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiFlashCNStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
				TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3, SizeGi: entity.MinTiFlashWNStorageSize, Iops: 10000, ThroughputMbps: 500,
				},
			}

			require.NoError(t, s.generatedSharedPoolComponent(
				ctx,
				sharedPool,
				&entity.RequestUnit{Min: 100000, Max: 200000},
				nodeSpecMap,
				ruConfig,
				initConfig,
				false,
			))

			componentTypes := make(map[apiv1.ComponentType]bool)
			var tiflashComputeComp, schemaManagerComp *entity.Component
			for _, component := range sharedPool.GetComponentGroups() {
				componentTypes[component.GetComponentType()] = true
				if component.GetComponentType() == apiv1.ComponentType_TIFLASH_COMPUTE {
					tiflashComputeComp = component
				}
				if component.GetComponentType() == apiv1.ComponentType_SCHEMA_MANAGER {
					schemaManagerComp = component
				}
			}
			require.True(t, componentTypes[apiv1.ComponentType_TIFLASH_COMPUTE])
			require.Equal(t, tt.wantTiflashComputeNode, tiflashComputeComp.GetNodeQuota().GetNodeCount())
			require.True(t, componentTypes[apiv1.ComponentType_TIFLASH_LEARNER])
			require.True(t, componentTypes[apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER])
			require.Equal(t, tt.wantTiflashWrite, componentTypes[apiv1.ComponentType_TIFLASH_WRITE])
			require.Equal(t, tt.wantSchemaManager, componentTypes[apiv1.ComponentType_SCHEMA_MANAGER])
			if tt.wantSchemaManager {
				require.Equal(t, tt.wantSchemaManagerNode, schemaManagerComp.GetNodeQuota().GetNodeCount())
				require.Equal(t, uint32(2), schemaManagerComp.GetNodeQuota().GetNodeSpec().GetCpu())
				require.Equal(t, uint32(8), schemaManagerComp.GetNodeQuota().GetNodeSpec().GetMemoryGi())

				schemaManagerComp.NodeQuota.NodeCount = 1
				schemaManagerComp.Phase = apiv1.Component_AVAILABLE

				// Simulate RU-based regeneration after an explicit schema manager scale.
				require.NoError(t, s.generatedSharedPoolComponent(
					ctx,
					sharedPool,
					&entity.RequestUnit{Min: 200000, Max: 400000},
					nodeSpecMap,
					ruConfig,
					initConfig,
					false,
				))
				require.Equal(t, uint32(1), schemaManagerComp.GetNodeQuota().GetNodeCount(),
					"post-creation regeneration must preserve explicit schema manager scaling")
			}
		})
	}
}

func TestGenDefaultSchemaManagerQuota(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"2C8G": {{Cpu: 2, MemoryGi: 8, AvailableCpu: 2, AvailableMemoryMi: 8192}},
	}

	tests := []struct {
		name               string
		phase              apiv1.Component_Phase
		startSchemaManager bool
		wantNodeCount      uint32
	}{
		{
			name:               "creating schema manager stays at zero without tiflash compute",
			phase:              apiv1.Component_CREATING,
			startSchemaManager: false,
			wantNodeCount:      0,
		},
		{
			name:               "creating schema manager starts with tiflash compute",
			phase:              apiv1.Component_CREATING,
			startSchemaManager: true,
			wantNodeCount:      1,
		},
		{
			name:               "post-creation schema manager is not changed by tiflash compute",
			phase:              apiv1.Component_AVAILABLE,
			startSchemaManager: true,
			wantNodeCount:      0,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			quota, err := s.genDefaultSchemaManagerQuota(ctx, nodeSpecMap, tt.phase, tt.startSchemaManager)
			require.NoError(t, err)
			require.Equal(t, tt.wantNodeCount, quota.GetNodeCount())
			require.Equal(t, uint32(0), quota.GetMinNodeCount())
			require.Equal(t, uint32(1), quota.GetMaxNodeCount())
		})
	}
}

func buildGeneratedSharedPoolComponentRuConfig() *entity.RuConfig {
	componentNames := []string{
		"tikv",
		"tikv_worker",
		"coprocessor_worker",
		"tidb",
		"tidb_worker",
		"pd",
		"pd_tso",
		"pd_scheduling",
		"tiflash_write",
		"tiflash_compute",
		"resource_manager",
		"tiproxy",
	}

	components := make(map[string]*apiv1.RuConfig_ComponentsMap, len(componentNames))
	for _, name := range componentNames {
		components[name] = &apiv1.RuConfig_ComponentsMap{
			MaxNodeCount: 10,
			NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
				"8c16g": {Min: 0, Max: 0},
			},
		}
	}

	return &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: components,
		},
	}
}

func TestGeneratedSharedPoolComponentBootstrapOverlay(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	selectedTiKV := &entity.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableCpu: 15, AvailableMemoryMi: 30000}
	selectedCompute := &entity.NodeSpec{Cpu: 32, MemoryGi: 64, AvailableCpu: 31, AvailableMemoryMi: 62000}

	for _, tt := range []struct {
		name             string
		compute          *BootstrapComponentCapacity
		wantComputeCount uint32
		wantComputeSpec  *entity.NodeSpec
	}{
		{
			name:             "TiKV and Compute override",
			compute:          &BootstrapComponentCapacity{NodeSpec: selectedCompute, NodeCount: 1},
			wantComputeCount: 1,
			wantComputeSpec:  selectedCompute,
		},
		{
			name:             "omitted Compute becomes zero-node scaffold",
			wantComputeCount: 0,
		},
	} {
		t.Run(tt.name, func(t *testing.T) {
			ruConfig := buildGeneratedSharedPoolComponentRuConfig()
			ruConfig.Components["tikv"].NodeSpecRuRange = map[string]*apiv1.RuConfig_RuRange{
				"16c32g": {Min: 100_000, Max: 500_000},
			}
			ruConfigBefore := proto.Clone(&ruConfig.RuConfig)
			sharedPool := newGeneratedSharedPoolForBootstrapTest()
			nodeSpecs := generatedSharedPoolNodeSpecs()
			nodeSpecs["16C32G"] = []*entity.NodeSpec{proto.Clone(selectedTiKV).(*entity.NodeSpec)}
			err := s.generatedSharedPoolComponent(
				ctx,
				sharedPool,
				nil,
				nodeSpecs,
				ruConfig,
				generatedSharedPoolInitConfig(),
				false,
				&SharedPoolBootstrapCapacity{
					TiKV:           &BootstrapComponentCapacity{NodeSpec: selectedTiKV, NodeCount: 6},
					TiFlashCompute: tt.compute,
				},
			)
			require.NoError(t, err)
			assert.True(t, proto.Equal(ruConfigBefore, &ruConfig.RuConfig), "bootstrap overlay must not mutate the resolved RU config")

			tikv := requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIKV)
			compute := requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIFLASH_COMPUTE)
			write := requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIFLASH_WRITE)
			assert.True(t, proto.Equal(selectedTiKV, tikv.GetNodeQuota().GetNodeSpec()))
			assert.EqualValues(t, 6, tikv.GetNodeQuota().GetNodeCount())
			assert.Positive(t, tikv.GetNodeQuota().GetMinNodeCount())
			assert.Positive(t, tikv.GetNodeQuota().GetMaxNodeCount())
			assert.Equal(t, tt.wantComputeCount, compute.GetNodeQuota().GetNodeCount())
			if tt.wantComputeSpec != nil {
				assert.True(t, proto.Equal(tt.wantComputeSpec, compute.GetNodeQuota().GetNodeSpec()))
			} else {
				assert.EqualValues(t, 8, compute.GetNodeQuota().GetNodeSpec().GetCpu(), "omitted Compute preserves the RU-derived shape")
			}
			assert.Zero(t, compute.GetNodeQuota().GetMinNodeCount())
			assert.EqualValues(t, 10, compute.GetNodeQuota().GetMaxNodeCount())
			assert.NotZero(t, write.GetNodeQuota().GetNodeCount(), "TiFlash Write remains RU-derived")
			requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIFLASH_LEARNER)
			requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIFLASH_COMPUTE_LEARNER)
		})
	}
}

func TestGeneratedSharedPoolComponentBootstrapUsesSelectedLocalDiskForStorage(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	selected := &entity.NodeSpec{
		Cpu: 16, MemoryGi: 32, AvailableCpu: 15, AvailableMemoryMi: 30000,
		WithLocalDisk: true, LocalDiskSizeGi: 900,
	}
	initConfig := generatedSharedPoolInitConfig()
	initConfig.TikvDataStorage = &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{Type: entity.LocalSSD}
	nodeSpecs := generatedSharedPoolNodeSpecs()
	nodeSpecs["16C32G"] = []*entity.NodeSpec{proto.Clone(selected).(*entity.NodeSpec)}
	sharedPool := newGeneratedSharedPoolForBootstrapTest()
	ruConfig := buildGeneratedSharedPoolComponentRuConfig()
	ruConfig.Components["tikv"].NodeSpecRuRange = map[string]*apiv1.RuConfig_RuRange{
		"16c32g": {Min: 0, Max: 0},
	}

	err := s.generatedSharedPoolComponent(
		ctx,
		sharedPool,
		nil,
		nodeSpecs,
		ruConfig,
		initConfig,
		false,
		&SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: selected, NodeCount: 3}},
	)
	require.NoError(t, err)
	tikv := requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIKV)
	require.NotEmpty(t, tikv.GetStorages())
	assert.Equal(t, entity.LocalSSD, tikv.GetStorages()[0].GetType())
	assert.EqualValues(t, 900, tikv.GetStorages()[0].GetSizeGi())
}

func TestGeneratedSharedPoolComponentBootstrapUsesInferredRU(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	selectedTiKV := &entity.NodeSpec{Cpu: 16, MemoryGi: 32, AvailableCpu: 15, AvailableMemoryMi: 30000}
	ruConfig := buildGeneratedSharedPoolComponentRuConfig()
	ruConfig.Components["tikv"].NodeSpecRuRange = map[string]*apiv1.RuConfig_RuRange{
		"16c32g": {Min: 100_000, Max: 500_000},
	}
	for _, component := range []string{"tiproxy", "tidb_worker", "tikv_worker", "coprocessor_worker"} {
		ruConfig.Components[component].NodeSpecRuRange = map[string]*apiv1.RuConfig_RuRange{
			"4c8g":  {Min: 0, Max: 300_000},
			"8c16g": {Min: 300_001, Max: 0},
		}
	}
	nodeSpecs := generatedSharedPoolNodeSpecs()
	nodeSpecs["16C32G"] = []*entity.NodeSpec{proto.Clone(selectedTiKV).(*entity.NodeSpec)}
	nodeSpecs["4C8G"] = []*entity.NodeSpec{{Cpu: 4, MemoryGi: 8, AvailableCpu: 4, AvailableMemoryMi: 8192}}
	nodeSpecs["2C8G"] = []*entity.NodeSpec{{Cpu: 2, MemoryGi: 8, AvailableCpu: 2, AvailableMemoryMi: 8192}}
	sharedPool := newGeneratedSharedPoolForBootstrapTest()
	sharedPool.DeployType = mysqlschema.SharedPoolDeployTypeSharedTiProxy

	err := s.generatedSharedPoolComponent(
		ctx,
		sharedPool,
		nil,
		nodeSpecs,
		ruConfig,
		generatedSharedPoolInitConfig(),
		false,
		&SharedPoolBootstrapCapacity{TiKV: &BootstrapComponentCapacity{NodeSpec: selectedTiKV, NodeCount: 6}},
	)
	require.NoError(t, err)

	tiproxy := requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIPROXY)
	assert.EqualValues(t, 2, tiproxy.GetNodeQuota().GetNodeCount())
	assert.EqualValues(t, 2, tiproxy.GetNodeQuota().GetMinNodeCount())
	assert.EqualValues(t, 10, tiproxy.GetNodeQuota().GetMaxNodeCount(), "bootstrap inference does not replace the configured HPA cap")
	assert.EqualValues(t, 8, tiproxy.GetNodeQuota().GetNodeSpec().GetCpu(), "hidden spec is selected at the inferred upper bound")
	worker := requireGeneratedComponent(t, sharedPool, apiv1.ComponentType_TIKV_WORKER)
	assert.EqualValues(t, 1, worker.GetNodeQuota().GetMinNodeCount())
	assert.EqualValues(t, 10, worker.GetNodeQuota().GetMaxNodeCount(), "workers without an RU-sensitive count rule retain their configured HPA cap")
	assert.EqualValues(t, 8, worker.GetNodeQuota().GetNodeSpec().GetCpu())
}

func newGeneratedSharedPoolForBootstrapTest() *entity.SharedPool {
	return &entity.SharedPool{SharedPool: apiv1.SharedPool{
		Id:               "sp1",
		CapacityMode:     apiv1.CapacityMode_RCU_DRIVEN,
		PdDeploymentMode: apiv1.PDDeploymentMode_PD_DEPLOYMENT_MODE_MONOLITHIC,
		Provider:         &apiv1.Provider{Type: apiv1.ProviderType_AWS},
	}}
}

func generatedSharedPoolNodeSpecs() map[string][]*entity.NodeSpec {
	return map[string][]*entity.NodeSpec{
		"8C16G": {{Cpu: 8, MemoryGi: 16, AvailableCpu: 8, AvailableMemoryMi: 16384}},
	}
}

func generatedSharedPoolInitConfig() *entity.InitSharedPoolConfig {
	return &entity.InitSharedPoolConfig{
		TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type: entity.GP3, SizeGi: entity.MinTiKVDataStorageSize, Iops: 10000, ThroughputMbps: 500,
		},
		TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type: entity.GP3, SizeGi: entity.MinTiKVRaftlogStorageSize, Iops: 5000, ThroughputMbps: 250,
		},
		TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type: entity.GP3, SizeGi: entity.MinTiFlashCNStorageSize, Iops: 10000, ThroughputMbps: 500,
		},
		TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type: entity.GP3, SizeGi: entity.MinTiFlashWNStorageSize, Iops: 10000, ThroughputMbps: 500,
		},
	}
}

func requireGeneratedComponent(t *testing.T, sharedPool *entity.SharedPool, componentType apiv1.ComponentType) *entity.Component {
	t.Helper()
	for _, component := range sharedPool.GetComponentGroups() {
		if component.GetComponentType() == componentType {
			return component
		}
	}
	require.FailNowf(t, "component not generated", "component type %s", componentType)
	return nil
}

func TestGenDefaultStorages(t *testing.T) {
	s := &serverImpl{}

	// Test TiDB storages for AWS
	storages, err := s.genDefaultTiDBStorages(apiv1.ProviderType_AWS, &entity.RequestUnit{})
	if err != nil {
		t.Fatalf("failed to generate TiDB storages for AWS: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for TiDB, got %d", len(storages))
	}

	// Test TiDB storages for AliCloud
	storages, err = s.genDefaultTiDBStorages(apiv1.ProviderType_ALICLOUD, &entity.RequestUnit{})
	if err != nil {
		t.Fatalf("failed to generate TiDB storages for AliCloud: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for TiDB, got %d", len(storages))
	}

	// Test unsupported provider
	_, err = s.genDefaultTiDBStorages(apiv1.ProviderType_GCP, &entity.RequestUnit{})
	if err == nil {
		t.Fatalf("expected error for unsupported provider")
	}
	if st, ok := status.FromError(err); ok {
		if st.Code() != codes.FailedPrecondition {
			t.Fatalf("expected FailedPrecondition error, got %v", st.Code())
		}
	} else {
		t.Fatalf("expected gRPC status error, got %v", err)
	}

	// Test PD storages for AWS
	storages, err = s.genDefaultPDStorages(apiv1.ProviderType_AWS, &entity.RequestUnit{})
	if err != nil {
		t.Fatalf("failed to generate PD storages for AWS: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for PD, got %d", len(storages))
	}

	// Test PD storages for AliCloud
	storages, err = s.genDefaultPDStorages(apiv1.ProviderType_ALICLOUD, &entity.RequestUnit{})
	if err != nil {
		t.Fatalf("failed to generate PD storages for AliCloud: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for PD, got %d", len(storages))
	}
}

func TestGenDefaultTiFlashStorages(t *testing.T) {
	s := &serverImpl{}

	nodeSpec := &apiv1.NodeSpec{
		Cpu:               8,
		MemoryGi:          32,
		LocalDiskSizeGi:   1000,
		AvailableCpu:      8,
		AvailableMemoryMi: 32000,
	}

	initConfig := &entity.InitSharedPoolConfig{
		TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.GP3,
			SizeGi:         500,
			Iops:           3000,
			ThroughputMbps: 125,
		},
		TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.GP3,
			SizeGi:         500,
			Iops:           3000,
			ThroughputMbps: 125,
		},
	}

	// Test TiFlash CN storages for AWS
	storages, err := s.genDefaultTiFlashCNStorages(apiv1.ProviderType_AWS, nodeSpec, initConfig)
	if err != nil {
		t.Fatalf("failed to generate TiFlash CN storages for AWS: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for TiFlash CN, got %d", len(storages))
	}
	if storages[0].Type != entity.GP3 {
		t.Fatalf("expected GP3 storage type, got %s", storages[0].Type)
	}

	// Test TiFlash WN storages for AWS
	storages, err = s.genDefaultTiFlashWNStorages(apiv1.ProviderType_AWS, initConfig)
	if err != nil {
		t.Fatalf("failed to generate TiFlash WN storages for AWS: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for TiFlash WN, got %d", len(storages))
	}
	if storages[0].Type != entity.GP3 {
		t.Fatalf("expected GP3 storage type, got %s", storages[0].Type)
	}

	// Test TiFlash CN storages for AliCloud with correct storage type
	initConfigAli := &entity.InitSharedPoolConfig{
		TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.CloudAuto, // Use correct type for AliCloud
			SizeGi:         100,
			Iops:           1000,
			ThroughputMbps: 0,
		},
		TiflashWnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.CloudAuto, // Use correct type for AliCloud
			SizeGi:         100,
			Iops:           1000,
			ThroughputMbps: 0,
		},
	}

	storages, err = s.genDefaultTiFlashCNStorages(apiv1.ProviderType_ALICLOUD, nodeSpec, initConfigAli)
	if err != nil {
		t.Fatalf("failed to generate TiFlash CN storages for AliCloud: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for TiFlash CN, got %d", len(storages))
	}

	// Test TiFlash WN storages for AliCloud
	storages, err = s.genDefaultTiFlashWNStorages(apiv1.ProviderType_ALICLOUD, initConfigAli)
	if err != nil {
		t.Fatalf("failed to generate TiFlash WN storages for AliCloud: %v", err)
	}
	if len(storages) != 1 {
		t.Fatalf("expected 1 storage for TiFlash WN, got %d", len(storages))
	}

	// Test unsupported provider
	_, err = s.genDefaultTiFlashCNStorages(apiv1.ProviderType_GCP, nodeSpec, initConfig)
	if err == nil {
		t.Fatalf("expected error for unsupported provider")
	}
}

func TestGenDefaultTiKVStorages(t *testing.T) {
	s := &serverImpl{}

	nodeSpec := &entity.NodeSpec{
		Cpu:               8,
		MemoryGi:          32,
		LocalDiskSizeGi:   1000,
		AvailableCpu:      8,
		AvailableMemoryMi: 32000,
	}

	initConfig := &entity.InitSharedPoolConfig{
		TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.GP3,
			SizeGi:         1500,
			Iops:           8000,
			ThroughputMbps: 400,
		},
		TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:           entity.GP3,
			SizeGi:         150,
			Iops:           3000,
			ThroughputMbps: 125,
		},
	}

	// Test TiKV storages for AWS with GP3
	storages, err := s.genDefaultTiKVStorages(apiv1.ProviderType_AWS, nodeSpec, initConfig)
	if err != nil {
		t.Fatalf("failed to generate TiKV storages for AWS: %v", err)
	}
	if len(storages) != 2 {
		t.Fatalf("expected 2 storages for TiKV (data + raftlog), got %d", len(storages))
	}

	// Verify data and raftlog storages
	var dataStorage, raftlogStorage *entity.Storage
	for _, storage := range storages {
		switch storage.Name {
		case apiv1.Component_Storage_DATA:
			dataStorage = storage
		case apiv1.Component_Storage_RAFTLOG:
			raftlogStorage = storage
		}
	}
	if dataStorage == nil {
		t.Fatalf("missing data storage")
	}
	if raftlogStorage == nil {
		t.Fatalf("missing raftlog storage")
	}

	if dataStorage.Type != entity.GP3 {
		t.Fatalf("expected GP3 data storage type, got %s", dataStorage.Type)
	}
	if raftlogStorage.Type != entity.GP3 {
		t.Fatalf("expected GP3 raftlog storage type, got %s", raftlogStorage.Type)
	}

	// Test TiKV storages for AWS with LocalSSD
	initConfig.TikvDataStorage.Type = entity.LocalSSD
	initConfig.TikvDataStorage.SizeGi = 0         // LocalSSD should not have size set
	initConfig.TikvDataStorage.Iops = 0           // LocalSSD should not have IOPS
	initConfig.TikvDataStorage.ThroughputMbps = 0 // LocalSSD should not have throughput
	// Keep TikvRaftlogStorage as GP3 since LocalSSD is not supported for raftlog
	initConfig.TikvRaftlogStorage.Type = entity.GP3
	initConfig.TikvRaftlogStorage.SizeGi = 150
	initConfig.TikvRaftlogStorage.Iops = 3000
	initConfig.TikvRaftlogStorage.ThroughputMbps = 125
	storages, err = s.genDefaultTiKVStorages(apiv1.ProviderType_AWS, nodeSpec, initConfig)
	if err != nil {
		t.Fatalf("failed to generate TiKV storages for AWS with LocalSSD: %v", err)
	}
	if len(storages) != 2 {
		t.Fatalf("expected 2 storages for TiKV with LocalSSD (data + raftlog), got %d", len(storages))
	}

	// Reset variables to check LocalSSD storages
	dataStorage = nil
	raftlogStorage = nil
	for _, storage := range storages {
		switch storage.Name {
		case apiv1.Component_Storage_DATA:
			dataStorage = storage
		case apiv1.Component_Storage_RAFTLOG:
			raftlogStorage = storage
		}
	}
	if dataStorage == nil {
		t.Fatalf("missing data storage for LocalSSD")
	}
	if dataStorage.Type != entity.LocalSSD {
		t.Fatalf("expected LocalSSD data storage type, got %s", dataStorage.Type)
	}
	if raftlogStorage == nil {
		t.Fatalf("missing raftlog storage for LocalSSD")
	}

	// Test TiKV storages for AliCloud
	initConfigAli := &entity.InitSharedPoolConfig{
		TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:   entity.CloudAuto,
			SizeGi: 1500,
			Iops:   8000,
			// AliCloud cloud-essd does not support throughput
		},
		TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
			Type:   entity.CloudAuto,
			SizeGi: 150,
			Iops:   3000,
			// AliCloud cloud-essd does not support throughput
		},
	}
	storages, err = s.genDefaultTiKVStorages(apiv1.ProviderType_ALICLOUD, nodeSpec, initConfigAli)
	if err != nil {
		t.Fatalf("failed to generate TiKV storages for AliCloud: %v", err)
	}
	if len(storages) != 2 {
		t.Fatalf("expected 2 storages for TiKV (data + raftlog), got %d", len(storages))
	}
}

func TestComponentPhaseTransitions(t *testing.T) {
	makeSpec := func(cpu, mem uint32) *entity.NodeSpec {
		return &entity.NodeSpec{Cpu: cpu, MemoryGi: mem, AvailableCpu: cpu, AvailableMemoryMi: mem * 1000}
	}
	newQuota := func(count, max, min uint32, spec *entity.NodeSpec) *entity.NodeQuota {
		return &entity.NodeQuota{NodeCount: count, MaxNodeCount: max, MinNodeCount: min, NodeSpec: spec}
	}

	// Test case: component transitions from CREATING to CREATING (no change)
	{
		comp := &entity.Component{
			Phase:     apiv1.Component_CREATING,
			NodeQuota: newQuota(2, 5, 1, makeSpec(4, 8)),
		}
		newQuota := newQuota(3, 5, 1, makeSpec(8, 16))
		oldQuota := comp.GetNodeQuota()

		// Simulate quota comparison
		finalQuota := compareNodeQuota(comp.GetNodeQuota(), newQuota)
		comp.NodeQuota = finalQuota

		// Phase should remain CREATING for new components
		if comp.Phase != apiv1.Component_CREATING {
			t.Fatalf("expected CREATING phase for new component, got %s", comp.Phase)
		}

		// Should use new spec since it's larger
		if comp.NodeQuota.NodeSpec.Cpu != 8 {
			t.Fatalf("expected new spec to be used, got %d CPU", comp.NodeQuota.NodeSpec.Cpu)
		}

		// Node count should be max of old and new
		if comp.NodeQuota.NodeCount != 3 {
			t.Fatalf("expected node count 3, got %d", comp.NodeQuota.NodeCount)
		}

		_ = oldQuota // avoid unused variable
	}

	// Test case: available component with quota change should transition to MODIFYING
	{
		comp := &entity.Component{
			Phase:     apiv1.Component_AVAILABLE,
			NodeQuota: newQuota(2, 5, 1, makeSpec(4, 8)),
		}
		newQuota := newQuota(4, 5, 1, makeSpec(8, 16))
		oldQuota := comp.GetNodeQuota()

		// Simulate quota comparison and phase transition logic
		finalQuota := compareNodeQuota(comp.GetNodeQuota(), newQuota)
		comp.NodeQuota = finalQuota

		// Simulate phase transition for non-CREATING components with quota changes
		if comp.Phase != apiv1.Component_CREATING {
			// In real implementation, this would check reflect.DeepEqual
			if comp.NodeQuota.NodeCount != oldQuota.NodeCount ||
				comp.NodeQuota.NodeSpec.Cpu != oldQuota.NodeSpec.Cpu {
				comp.Phase = apiv1.Component_MODIFYING
			}
		}

		if comp.Phase != apiv1.Component_MODIFYING {
			t.Fatalf("expected MODIFYING phase for quota change, got %s", comp.Phase)
		}
	}
}

// Additional comprehensive edge case tests
func TestEdgeCasesAndBoundaryConditions(t *testing.T) {
	// Test zero and negative values
	t.Run("genTiKVNodeCount boundary values", func(t *testing.T) {
		cases := []struct {
			input    uint32
			expected uint32
			name     string
		}{
			{0, 3, "zero input"},
			{1, 3, "minimum input"},
			{2, 3, "below minimum"},
			{3, 3, "exact minimum"},
			{4, 3, "round down case"},
			{100, 99, "large number case"},
			{999, 999, "multiple of 3"},
			{1000, 999, "large round down"},
		}

		for _, tc := range cases {
			got := genTiKVNodeCount(tc.input)
			if got != tc.expected {
				t.Errorf("%s: genTiKVNodeCount(%d) = %d, want %d", tc.name, tc.input, got, tc.expected)
			}
		}
	})

	// Test compareNodeQuota edge cases
	t.Run("compareNodeQuota edge cases", func(t *testing.T) {
		makeSpec := func(cpu, mem uint32) *entity.NodeSpec {
			return &entity.NodeSpec{Cpu: cpu, MemoryGi: mem}
		}
		newQuota := func(count, max, min uint32, spec *entity.NodeSpec) *entity.NodeQuota {
			return &entity.NodeQuota{NodeCount: count, MaxNodeCount: max, MinNodeCount: min, NodeSpec: spec}
		}

		// Case: Equal specs but different counts
		{
			curr := newQuota(2, 5, 1, makeSpec(8, 16))
			newQ := newQuota(4, 6, 1, makeSpec(8, 16))
			got := compareNodeQuota(curr, newQ)
			if got.NodeCount != 4 {
				t.Errorf("equal specs: expected node count 4, got %d", got.NodeCount)
			}
			if got.NodeSpec.Cpu != 8 {
				t.Errorf("equal specs: expected to keep spec CPU 8, got %d", got.NodeSpec.Cpu)
			}
		}

		// Case: Zero values
		{
			curr := newQuota(0, 0, 0, makeSpec(0, 0))
			newQ := newQuota(1, 1, 1, makeSpec(1, 1))
			got := compareNodeQuota(curr, newQ)
			if got.NodeCount != 1 {
				t.Errorf("zero values: expected node count 1, got %d", got.NodeCount)
			}
		}

		// Case: Max boundary adjustments
		{
			curr := newQuota(10, 5, 1, makeSpec(8, 16))
			newQ := newQuota(15, 20, 1, makeSpec(16, 32))
			got := compareNodeQuota(curr, newQ)
			if got.MaxNodeCount < got.NodeCount {
				t.Errorf("max boundary: max node count %d should not be less than node count %d", got.MaxNodeCount, got.NodeCount)
			}
		}
	})
}

func TestStorageGenerationErrorScenarios(t *testing.T) {
	s := &serverImpl{}

	// Test storage generation with nil configs
	t.Run("nil configuration handling", func(t *testing.T) {
		nodeSpec := &entity.NodeSpec{
			Cpu:               8,
			MemoryGi:          32,
			LocalDiskSizeGi:   1000,
			AvailableCpu:      8,
			AvailableMemoryMi: 32000,
		}

		// Test with partially nil init config
		initConfig := &entity.InitSharedPoolConfig{
			TikvDataStorage: nil, // This should return an error
		}

		// This should return an error for nil storage config
		_, err := s.genDefaultTiKVStorages(apiv1.ProviderType_AWS, nodeSpec, initConfig)
		if err == nil {
			t.Error("expected error for nil TikvDataStorage, but got nil")
		}
	})
}

func TestProviderTypeValidation(t *testing.T) {
	s := &serverImpl{}

	// Test all unsupported provider types
	unsupportedProviders := []apiv1.ProviderType{
		apiv1.ProviderType_GCP,
		999, // Invalid enum value
	}

	for _, provider := range unsupportedProviders {
		t.Run(fmt.Sprintf("unsupported provider %v", provider), func(t *testing.T) {
			// Test TiDB storage generation
			_, err := s.genDefaultTiDBStorages(provider, &entity.RequestUnit{})
			if err == nil {
				t.Errorf("expected error for unsupported provider %v in TiDB storages", provider)
			}

			// Test PD storage generation
			_, err = s.genDefaultPDStorages(provider, &entity.RequestUnit{})
			if err == nil {
				t.Errorf("expected error for unsupported provider %v in PD storages", provider)
			}

			// Test TiFlash storage generation
			initConfig := &entity.InitSharedPoolConfig{
				TiflashCnStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
					Type: entity.GP3,
				},
			}
			nodeSpec := &apiv1.NodeSpec{Cpu: 8, MemoryGi: 32}

			_, err = s.genDefaultTiFlashCNStorages(provider, nodeSpec, initConfig)
			if err == nil {
				t.Errorf("expected error for unsupported provider %v in TiFlash CN storages", provider)
			}

			_, err = s.genDefaultTiFlashWNStorages(provider, initConfig)
			if err == nil {
				t.Errorf("expected error for unsupported provider %v in TiFlash WN storages", provider)
			}
		})
	}
}

func TestQuotaCalculationEdgeCases(t *testing.T) {
	makeSpec := func(cpu, mem uint32) *entity.NodeSpec {
		return &entity.NodeSpec{Cpu: cpu, MemoryGi: mem}
	}
	newQuota := func(count, max, min uint32, spec *entity.NodeSpec) *entity.NodeQuota {
		return &entity.NodeQuota{NodeCount: count, MaxNodeCount: max, MinNodeCount: min, NodeSpec: spec}
	}

	// Test overflow scenarios
	t.Run("quota overflow handling", func(t *testing.T) {
		// Large numbers that might cause overflow
		curr := newQuota(4294967290, 4294967295, 1, makeSpec(8, 16)) // Near uint32 max
		newQ := newQuota(10, 20, 1, makeSpec(16, 32))

		// Should not panic or overflow
		got := compareNodeQuota(curr, newQ)
		if got == nil {
			t.Error("compareNodeQuota should not return nil for large numbers")
		}
	})

	// Test memory vs CPU precedence
	t.Run("resource precedence in comparison", func(t *testing.T) {
		// Higher CPU, lower memory vs lower CPU, higher memory
		curr := newQuota(2, 5, 1, makeSpec(16, 8)) // High CPU, low memory
		newQ := newQuota(3, 6, 1, makeSpec(8, 32)) // Low CPU, high memory

		got := compareNodeQuota(curr, newQ)
		// Should choose based on memory being higher
		if got.NodeSpec.MemoryGi != 32 {
			t.Errorf("expected to choose higher memory spec, got memory %d", got.NodeSpec.MemoryGi)
		}
	})
}

func TestNodeSpecValidationEdgeCases(t *testing.T) {
	ctx := context.Background()

	// Test empty nodeSpecMap
	t.Run("empty nodeSpecMap", func(t *testing.T) {
		desired := &entity.NodeSpec{Cpu: 8, MemoryGi: 16}
		_, err := getNodeSpec(ctx, desired, map[string][]*entity.NodeSpec{}, false)
		if err == nil {
			t.Error("expected error for empty nodeSpecMap")
		}
	})

	// Test invalid node spec combinations
	t.Run("invalid node spec combinations", func(t *testing.T) {
		desired := &entity.NodeSpec{Cpu: 8, MemoryGi: 16}
		key := entity.GetComponentNodeSpec(desired)

		// All specs are invalid (zero resources)
		invalidSpecs := []*entity.NodeSpec{
			{Cpu: 8, MemoryGi: 16, WithLocalDisk: false, AvailableCpu: 0, AvailableMemoryMi: 16000},
			{Cpu: 8, MemoryGi: 16, WithLocalDisk: false, AvailableCpu: 8, AvailableMemoryMi: 0},
			{Cpu: 8, MemoryGi: 16, WithLocalDisk: true, LocalDiskSizeGi: 0, AvailableCpu: 8, AvailableMemoryMi: 16000},
		}

		nodeSpecMap := map[string][]*entity.NodeSpec{
			key: invalidSpecs,
		}

		_, err := getNodeSpec(ctx, desired, nodeSpecMap, false)
		if err == nil {
			t.Error("expected error when all node specs are invalid")
		}

		_, err = getNodeSpec(ctx, desired, nodeSpecMap, true)
		if err == nil {
			t.Error("expected error when all local disk specs are invalid")
		}
	})

	// Test mixed valid/invalid specs
	t.Run("mixed valid and invalid specs", func(t *testing.T) {
		desired := &entity.NodeSpec{Cpu: 8, MemoryGi: 16}
		key := entity.GetComponentNodeSpec(desired)

		mixedSpecs := []*entity.NodeSpec{
			{Cpu: 8, MemoryGi: 16, WithLocalDisk: false, AvailableCpu: 0, AvailableMemoryMi: 16000},                    // Invalid
			{Cpu: 8, MemoryGi: 16, WithLocalDisk: false, AvailableCpu: 8, AvailableMemoryMi: 16000},                    // Valid
			{Cpu: 8, MemoryGi: 16, WithLocalDisk: true, LocalDiskSizeGi: 0, AvailableCpu: 8, AvailableMemoryMi: 16000}, // Invalid local
		}

		nodeSpecMap := map[string][]*entity.NodeSpec{
			key: mixedSpecs,
		}

		// Should find the valid non-local spec
		spec, err := getNodeSpec(ctx, desired, nodeSpecMap, false)
		if err != nil {
			t.Errorf("should find valid non-local spec, got error: %v", err)
		}
		if spec.WithLocalDisk {
			t.Error("should return non-local disk spec when withLocalDisk=false")
		}

		// Should fail to find valid local spec
		_, err = getNodeSpec(ctx, desired, nodeSpecMap, true)
		if err == nil {
			t.Error("should fail to find valid local disk spec")
		}
	})
}

func TestDataConsistencyAndIntegrity(t *testing.T) {
	// Test that generated configurations are internally consistent
	t.Run("storage configuration consistency", func(t *testing.T) {
		s := &serverImpl{}

		nodeSpec := &entity.NodeSpec{
			Cpu:               8,
			MemoryGi:          32,
			LocalDiskSizeGi:   1000,
			AvailableCpu:      8,
			AvailableMemoryMi: 32000,
		}

		initConfig := &entity.InitSharedPoolConfig{
			TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           entity.GP3,
				SizeGi:         1500,
				Iops:           8000,
				ThroughputMbps: 400,
			},
			TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           entity.GP3,
				SizeGi:         150,
				Iops:           3000,
				ThroughputMbps: 125,
			},
		}

		storages, err := s.genDefaultTiKVStorages(apiv1.ProviderType_AWS, nodeSpec, initConfig)
		if err != nil {
			t.Fatalf("failed to generate storages: %v", err)
		}

		// Verify storage consistency
		var dataStorage, raftlogStorage *entity.Storage
		for _, storage := range storages {
			switch storage.Name {
			case apiv1.Component_Storage_DATA:
				dataStorage = storage
			case apiv1.Component_Storage_RAFTLOG:
				raftlogStorage = storage
			}
		}

		if dataStorage == nil || raftlogStorage == nil {
			t.Fatal("missing required storage components")
		}

		// Data storage should be larger than raftlog storage
		if dataStorage.SizeGi <= raftlogStorage.SizeGi {
			t.Errorf("data storage size %d should be larger than raftlog storage size %d",
				dataStorage.SizeGi, raftlogStorage.SizeGi)
		}

		// Both should have positive IOPS
		if dataStorage.Iops <= 0 || raftlogStorage.Iops <= 0 {
			t.Errorf("storage IOPS should be positive: data=%d, raftlog=%d",
				dataStorage.Iops, raftlogStorage.Iops)
		}
	})

	// Test quota constraints
	t.Run("quota constraint validation", func(t *testing.T) {
		makeSpec := func(cpu, mem uint32) *entity.NodeSpec {
			return &entity.NodeSpec{Cpu: cpu, MemoryGi: mem}
		}
		newQuotaFunc := func(count, max, min uint32, spec *entity.NodeSpec) *entity.NodeQuota {
			return &entity.NodeQuota{NodeCount: count, MaxNodeCount: max, MinNodeCount: min, NodeSpec: spec}
		}

		// Test quota constraints are maintained
		curr := newQuotaFunc(5, 10, 1, makeSpec(8, 16))
		newQ := newQuotaFunc(15, 12, 2, makeSpec(16, 32))

		result := compareNodeQuota(curr, newQ)

		// Node count should not exceed max
		if result.NodeCount > result.MaxNodeCount {
			t.Errorf("node count %d exceeds max %d", result.NodeCount, result.MaxNodeCount)
		}

		// Node count should be at least min
		if result.NodeCount < result.MinNodeCount {
			t.Errorf("node count %d below min %d", result.NodeCount, result.MinNodeCount)
		}

		// Max should be at least node count
		if result.MaxNodeCount < result.NodeCount {
			t.Errorf("max node count %d below node count %d", result.MaxNodeCount, result.NodeCount)
		}
	})
}

func TestMemoryLeaksAndResourceManagement(t *testing.T) {
	t.Run("large data structure handling", func(t *testing.T) {
		// Test with very large node spec maps to check for memory issues
		desired := &entity.NodeSpec{Cpu: 8, MemoryGi: 16}
		key := entity.GetComponentNodeSpec(desired)

		// Create a large number of node specs
		specs := make([]*entity.NodeSpec, 1000)
		for i := 0; i < 1000; i++ {
			specs[i] = &entity.NodeSpec{
				Cpu: 8, MemoryGi: 16, WithLocalDisk: false,
				AvailableCpu: 8, AvailableMemoryMi: 16000,
			}
		}

		nodeSpecMap := map[string][]*entity.NodeSpec{
			key: specs,
		}
		ctx := context.Background()

		// This should handle large data structures without issues
		_, err := getNodeSpec(ctx, desired, nodeSpecMap, false)
		if err != nil {
			t.Errorf("should handle large node spec map, got error: %v", err)
		}
	})

	t.Run("deep copy and mutation safety", func(t *testing.T) {
		makeSpec := func(cpu, mem uint32) *entity.NodeSpec {
			return &entity.NodeSpec{Cpu: cpu, MemoryGi: mem}
		}
		newQuotaFunc := func(count, max, min uint32, spec *entity.NodeSpec) *entity.NodeQuota {
			return &entity.NodeQuota{NodeCount: count, MaxNodeCount: max, MinNodeCount: min, NodeSpec: spec}
		}

		original := newQuotaFunc(4, 10, 1, makeSpec(8, 16))
		newQ := newQuotaFunc(6, 12, 1, makeSpec(16, 32))

		// Store original values
		originalNodeCount := original.NodeCount
		originalCpu := original.NodeSpec.Cpu

		// Call compareNodeQuota
		result := compareNodeQuota(original, newQ)

		// Verify the function doesn't mutate inputs when it shouldn't
		if original == newQ {
			// If same spec is chosen, original should be modified
			return
		}

		// If different spec is chosen, result should be newQ quota with adjusted counts
		if result.NodeSpec.Cpu == originalCpu {
			// Original spec chosen, verify original values are preserved or intentionally modified
			if original.NodeCount == originalNodeCount && result.NodeCount != originalNodeCount {
				t.Error("original quota was unexpectedly modified")
			}
		}

		// Result should never be nil
		if result == nil {
			t.Error("compareNodeQuota should never return nil")
		}
	})
}

func TestErrorPropagationAndWrapping(t *testing.T) {
	t.Run("error context preservation", func(t *testing.T) {
		ctx := context.Background()

		// Test getNodeSpec with nil desired should preserve error context
		_, err := getNodeSpec(ctx, nil, map[string][]*entity.NodeSpec{}, false)
		if err == nil {
			t.Error("expected error for nil desired node spec")
		}

		// Check that error contains meaningful information
		errStr := err.Error()
		if !strings.Contains(errStr, "node spec") {
			t.Errorf("error should contain context about node spec, got: %s", errStr)
		}
	})

	t.Run("storage generation error chains", func(t *testing.T) {
		s := &serverImpl{}

		// Test error propagation through storage generation chain
		_, err := s.genDefaultTiDBStorages(999, &entity.RequestUnit{}) // Invalid provider
		if err == nil {
			t.Error("expected error for invalid provider type")
		}

		// Check error contains provider information
		errStr := err.Error()
		if !strings.Contains(errStr, "provider") || !strings.Contains(errStr, "not supported") {
			t.Errorf("error should contain provider context, got: %s", errStr)
		}
	})
}

func TestIntegrationScenarios(t *testing.T) {
	t.Run("complete workflow integration", func(t *testing.T) {
		s := &serverImpl{}

		// Simulate a complete component generation workflow
		nodeSpec := &entity.NodeSpec{
			Cpu:               8,
			MemoryGi:          32,
			LocalDiskSizeGi:   1000,
			AvailableCpu:      8,
			AvailableMemoryMi: 32000,
		}

		initConfig := &entity.InitSharedPoolConfig{
			TikvDataStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           entity.GP3,
				SizeGi:         1500,
				Iops:           8000,
				ThroughputMbps: 400,
			},
			TikvRaftlogStorage: &apiv1.CreateSharedPoolRequest_InitSharedPoolConfig_Storage{
				Type:           entity.GP3,
				SizeGi:         150,
				Iops:           3000,
				ThroughputMbps: 125,
			},
		}

		// Generate TiKV storages
		tikvStorages, err := s.genDefaultTiKVStorages(apiv1.ProviderType_AWS, nodeSpec, initConfig)
		if err != nil {
			t.Fatalf("failed to generate TiKV storages: %v", err)
		}

		// Generate TiDB storages
		tidbStorages, err := s.genDefaultTiDBStorages(apiv1.ProviderType_AWS, &entity.RequestUnit{})
		if err != nil {
			t.Fatalf("failed to generate TiDB storages: %v", err)
		}

		// Generate PD storages
		pdStorages, err := s.genDefaultPDStorages(apiv1.ProviderType_AWS, &entity.RequestUnit{})
		if err != nil {
			t.Fatalf("failed to generate PD storages: %v", err)
		}

		// Verify all storages are generated
		if len(tikvStorages) == 0 {
			t.Error("TiKV storages should not be empty")
		}
		if len(tidbStorages) == 0 {
			t.Error("TiDB storages should not be empty")
		}
		if len(pdStorages) == 0 {
			t.Error("PD storages should not be empty")
		}

		// Verify storage types are consistent
		for _, storage := range tikvStorages {
			if storage.Type != entity.GP3 && storage.Type != entity.LocalSSD && storage.Type != entity.IO2 {
				t.Errorf("unexpected storage type in TiKV storages: %s", storage.Type)
			}
		}
	})

	t.Run("cross-provider consistency", func(t *testing.T) {
		s := &serverImpl{}

		// Test that AWS and AliCloud generate consistent structures
		awsTidbStorages, err := s.genDefaultTiDBStorages(apiv1.ProviderType_AWS, &entity.RequestUnit{})
		if err != nil {
			t.Fatalf("failed to generate AWS TiDB storages: %v", err)
		}

		aliTidbStorages, err := s.genDefaultTiDBStorages(apiv1.ProviderType_ALICLOUD, &entity.RequestUnit{})
		if err != nil {
			t.Fatalf("failed to generate AliCloud TiDB storages: %v", err)
		}

		// Both should have same number of storages
		if len(awsTidbStorages) != len(aliTidbStorages) {
			t.Errorf("AWS and AliCloud should generate same number of TiDB storages: AWS=%d, AliCloud=%d",
				len(awsTidbStorages), len(aliTidbStorages))
		}

		// Both should have same storage names (just different types)
		awsNames := make(map[string]bool)
		for _, storage := range awsTidbStorages {
			awsNames[storage.Name.String()] = true
		}

		for _, storage := range aliTidbStorages {
			if !awsNames[storage.Name.String()] {
				t.Errorf("AliCloud storage name %s not found in AWS storages", storage.Name.String())
			}
		}
	})
}

// TestInitialNodeCountLimits verifies that phase-based initial node count limits and their constants are defined and applied correctly
func TestInitialNodeCountLimits(t *testing.T) {
	// This test verifies that maxInitXXXCount constants are defined with expected values
	// These constants are used to limit initial node counts during CREATING phase to reduce costs

	t.Run("Verify initial max count constants", func(t *testing.T) {
		if maxInitTiDBCount != 6 {
			t.Errorf("expected maxInitTiDBCount = 6, got %d", maxInitTiDBCount)
		}
		if maxInitTiProxyCount != 2 {
			t.Errorf("expected maxInitTiProxyCount = 2, got %d", maxInitTiProxyCount)
		}
		if maxInitTiKVCount != 9 {
			t.Errorf("expected maxInitTiKVCount = 9, got %d", maxInitTiKVCount)
		}
		if maxInitTiFlashCount != 1 {
			t.Errorf("expected maxInitTiFlashCount = 1, got %d", maxInitTiFlashCount)
		}
	})

	t.Run("Verify per-CPU RU constants", func(t *testing.T) {
		if tidbPerCpuRu != 5000 {
			t.Errorf("expected tidbPerCpuRu = 5000, got %d", tidbPerCpuRu)
		}
		if tiproxyPerCpuRu != 15000 {
			t.Errorf("expected tiproxyPerCpuRu = 15000, got %d", tiproxyPerCpuRu)
		}
		if tikvPerCpuRu != 2500 {
			t.Errorf("expected tikvPerCpuRu = 2500, got %d", tikvPerCpuRu)
		}
		if tiflashPerCpuRu != 9000 {
			t.Errorf("expected tiflashPerCpuRu = 9000, got %d", tiflashPerCpuRu)
		}
	})

	// Note: Phase-based logic is tested through integration tests or manual verification
	// The quota generation functions accept a phase parameter and apply limits conditionally:
	// - CREATING phase: applies maxInitXXXCount limits
	// - Other phases (AVAILABLE, MODIFYING): no initial limits applied
}

func TestGenDefaultTiFlashQuota(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	// Create a valid ruConfig with tiflash_write and tiflash_compute
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tiflash_write": {
					MaxNodeCount: 10,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"8c16g": {Min: 0, Max: 0}, // 0 means unlimited
					},
				},
				"tiflash_compute": {
					MaxNodeCount: 10,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"8c16g": {Min: 0, Max: 0}, // 0 means unlimited
					},
				},
			},
		},
	}

	// Create nodeSpecMap with 8C16G spec
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"8C16G": {
			{
				Cpu:               8,
				MemoryGi:          16,
				AvailableCpu:      8,
				AvailableMemoryMi: 16384,
			},
		},
	}

	t.Run("tiflash write quota has MinNodeCount = 0", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Min: 100000, Max: 200000}

		quota, err := s.genDefaultTiFlashWriteQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		// Verify MinNodeCount is 0 for tiflash
		if quota.MinNodeCount != 0 {
			t.Errorf("expected MinNodeCount = 0, got %d", quota.MinNodeCount)
		}
	})

	t.Run("tiflash compute quota has MinNodeCount = 0", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Min: 100000, Max: 200000}

		quota, err := s.genDefaultTiFlashComputeQuota(ctx, totalRu, nodeSpecMap, ruConfig, false, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		// Verify MinNodeCount is 0 for tiflash
		if quota.MinNodeCount != 0 {
			t.Errorf("expected MinNodeCount = 0, got %d", quota.MinNodeCount)
		}
	})

	t.Run("tiflash write quota has MinNodeCount = 0 for non-CREATING phase", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Min: 100000, Max: 200000}

		quota, err := s.genDefaultTiFlashWriteQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_AVAILABLE)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		// Verify MinNodeCount is 0 for tiflash even in non-CREATING phase
		if quota.MinNodeCount != 0 {
			t.Errorf("expected MinNodeCount = 0 for AVAILABLE phase, got %d", quota.MinNodeCount)
		}
	})

	t.Run("tiflash compute quota has MinNodeCount = 0 for non-CREATING phase", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Min: 100000, Max: 200000}

		quota, err := s.genDefaultTiFlashComputeQuota(ctx, totalRu, nodeSpecMap, ruConfig, false, apiv1.Component_AVAILABLE)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		// Verify MinNodeCount is 0 for tiflash even in non-CREATING phase
		if quota.MinNodeCount != 0 {
			t.Errorf("expected MinNodeCount = 0 for AVAILABLE phase, got %d", quota.MinNodeCount)
		}
	})

	t.Run("tiflash write defaultNodeCount zero creates zero replicas", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Min: 100000, Max: 200000}
		ruConfig.Components["tiflash_write"].NodeSpecRuRange["8c16g"].DefaultNodeCount = uint32Ptr(0)
		defer func() {
			ruConfig.Components["tiflash_write"].NodeSpecRuRange["8c16g"].DefaultNodeCount = nil
		}()

		quota, err := s.genDefaultTiFlashWriteQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		if quota.NodeCount != 0 {
			t.Errorf("expected NodeCount = 0, got %d", quota.NodeCount)
		}
		if quota.MinNodeCount != 0 {
			t.Errorf("expected MinNodeCount = 0, got %d", quota.MinNodeCount)
		}
	})

	t.Run("tiflash write ignores defaultNodeCount after creating", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Min: 100000, Max: 200000}
		ruConfig.Components["tiflash_write"].NodeSpecRuRange["8c16g"].DefaultNodeCount = uint32Ptr(0)
		defer func() {
			ruConfig.Components["tiflash_write"].NodeSpecRuRange["8c16g"].DefaultNodeCount = nil
		}()

		quota, err := s.genDefaultTiFlashWriteQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_AVAILABLE)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		if quota.NodeCount != 1 {
			t.Errorf("expected NodeCount = 1, got %d", quota.NodeCount)
		}
		if quota.MinNodeCount != 0 {
			t.Errorf("expected MinNodeCount = 0, got %d", quota.MinNodeCount)
		}
	})
}

func TestGenDefaultTiProxyQuotaDefaultNodeCount(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	totalRu := &entity.RequestUnit{Min: 0, Max: 100000}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"2C4G": {
			{
				Cpu:               2,
				MemoryGi:          4,
				AvailableCpu:      2,
				AvailableMemoryMi: 4096,
			},
		},
	}
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tiproxy": {
					MaxNodeCount: 10,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"2c4g": {Min: 0, Max: 0},
					},
				},
			},
		},
	}

	t.Run("keeps legacy minimum two when defaultNodeCount is unset", func(t *testing.T) {
		quota, err := s.genDefaultTiProxyQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		if quota.NodeCount != 2 {
			t.Errorf("expected NodeCount = 2, got %d", quota.NodeCount)
		}
		if quota.MinNodeCount != 2 {
			t.Errorf("expected MinNodeCount = 2, got %d", quota.MinNodeCount)
		}
	})

	t.Run("allows explicit baseline one", func(t *testing.T) {
		ruConfig.Components["tiproxy"].NodeSpecRuRange["2c4g"].DefaultNodeCount = uint32Ptr(1)
		defer func() {
			ruConfig.Components["tiproxy"].NodeSpecRuRange["2c4g"].DefaultNodeCount = nil
		}()

		quota, err := s.genDefaultTiProxyQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		if quota.NodeCount != 1 {
			t.Errorf("expected NodeCount = 1, got %d", quota.NodeCount)
		}
		if quota.MinNodeCount != 1 {
			t.Errorf("expected MinNodeCount = 1, got %d", quota.MinNodeCount)
		}
	})

	t.Run("rejects explicit baseline zero", func(t *testing.T) {
		ruConfig.Components["tiproxy"].NodeSpecRuRange["2c4g"].DefaultNodeCount = uint32Ptr(0)
		defer func() {
			ruConfig.Components["tiproxy"].NodeSpecRuRange["2c4g"].DefaultNodeCount = nil
		}()

		_, err := s.genDefaultTiProxyQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error")
		}
	})
}

func TestGenDefaultTiProxyComponentDefaultNodeCountMarksModifying(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	totalRu := &entity.RequestUnit{Min: 0, Max: 100000}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"2C4G": {
			{
				Cpu:               2,
				MemoryGi:          4,
				AvailableCpu:      2,
				AvailableMemoryMi: 4096,
			},
		},
	}
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tiproxy": {
					MaxNodeCount: 10,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"2c4g": {Min: 0, Max: 0, DefaultNodeCount: uint32Ptr(3)},
					},
				},
			},
		},
	}
	tiproxyComp := &entity.Component{
		Id:            "test-cluster-id",
		ComponentType: apiv1.ComponentType_TIPROXY,
		Phase:         apiv1.Component_AVAILABLE,
		NodeQuota: &entity.NodeQuota{
			NodeCount:    1,
			MaxNodeCount: 10,
			MinNodeCount: 1,
			NodeSpec: &entity.NodeSpec{
				Cpu:      2,
				MemoryGi: 4,
			},
		},
	}

	err := s.genDefaultTiProxyComponent(ctx, tiproxyComp, totalRu, nodeSpecMap, ruConfig, false)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if tiproxyComp.GetNodeQuota().GetNodeCount() != 3 {
		t.Fatalf("expected NodeCount = 3, got %d", tiproxyComp.GetNodeQuota().GetNodeCount())
	}
	if tiproxyComp.GetPhase() != apiv1.Component_MODIFYING {
		t.Fatalf("expected phase = MODIFYING, got %s", tiproxyComp.GetPhase())
	}
}

func TestGenDefaultTiDBQuotaDefaultNodeCountRegenerate(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	totalRu := &entity.RequestUnit{Min: 0, Max: 100000}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"4C16G": {
			{
				Cpu:               4,
				MemoryGi:          16,
				AvailableCpu:      4,
				AvailableMemoryMi: 16384,
			},
		},
	}
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tidb": {
					MaxNodeCount: 10,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"4c16g": {Min: 0, Max: 0, DefaultNodeCount: uint32Ptr(2)},
					},
				},
			},
		},
	}

	quota, err := s.genDefaultTiDBQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_AVAILABLE)
	if err != nil {
		t.Fatalf("unexpected error: %v", err)
	}
	if quota.NodeCount != 2 {
		t.Errorf("expected NodeCount = 2, got %d", quota.NodeCount)
	}
	if quota.MinNodeCount != 1 {
		t.Errorf("expected MinNodeCount = 1, got %d", quota.MinNodeCount)
	}
}

func TestGenDefaultTiKVQuotaDefaultNodeCountValidation(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	totalRu := &entity.RequestUnit{Min: 0, Max: 100000}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"8C16G": {
			{
				Cpu:               8,
				MemoryGi:          16,
				AvailableCpu:      8,
				AvailableMemoryMi: 16384,
			},
		},
	}
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"tikv": {
					MaxNodeCount: 12,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"8c16g": {Min: 0, Max: 0},
					},
				},
			},
		},
	}

	t.Run("rejects non multiple of three", func(t *testing.T) {
		ruConfig.Components["tikv"].NodeSpecRuRange["8c16g"].DefaultNodeCount = uint32Ptr(4)
		defer func() {
			ruConfig.Components["tikv"].NodeSpecRuRange["8c16g"].DefaultNodeCount = nil
		}()

		_, err := s.genDefaultTiKVQuota(ctx, totalRu, nodeSpecMap, ruConfig, false, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error")
		}
	})

	t.Run("allows multiple of three", func(t *testing.T) {
		ruConfig.Components["tikv"].NodeSpecRuRange["8c16g"].DefaultNodeCount = uint32Ptr(6)
		defer func() {
			ruConfig.Components["tikv"].NodeSpecRuRange["8c16g"].DefaultNodeCount = nil
		}()

		quota, err := s.genDefaultTiKVQuota(ctx, totalRu, nodeSpecMap, ruConfig, false, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}
		if quota.NodeCount != 6 {
			t.Errorf("expected NodeCount = 6, got %d", quota.NodeCount)
		}
	})
}

func TestGenDefaultPDMicroserviceQuota(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"pd_tso": {
					MaxNodeCount: 2,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"4c8g": {Min: 0, Max: 0},
					},
				},
				"pd_scheduling": {
					MaxNodeCount: 4,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"4c8g": {Min: 0, Max: 0},
					},
				},
			},
		},
	}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"4C8G": {
			{
				Cpu:               4,
				MemoryGi:          8,
				AvailableCpu:      4,
				AvailableMemoryMi: 8192,
			},
		},
	}
	totalRu := &entity.RequestUnit{Max: 100000}

	t.Run("pd tso default quota keeps replicas at 2 and lowers min to 1", func(t *testing.T) {
		quota, err := s.genDefaultPDTsoQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		if quota.NodeCount != 2 {
			t.Errorf("expected NodeCount = 2, got %d", quota.NodeCount)
		}
		if quota.MaxNodeCount != 2 {
			t.Errorf("expected MaxNodeCount = 2, got %d", quota.MaxNodeCount)
		}
		if quota.MinNodeCount != 1 {
			t.Errorf("expected MinNodeCount = 1, got %d", quota.MinNodeCount)
		}
	})

	t.Run("pd scheduling default quota keeps replicas at 2, uses ru config max, and lowers min to 1", func(t *testing.T) {
		quota, err := s.genDefaultPDSchedulingQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		if quota.NodeCount != 2 {
			t.Errorf("expected NodeCount = 2, got %d", quota.NodeCount)
		}
		if quota.MaxNodeCount != 4 {
			t.Errorf("expected MaxNodeCount = 4, got %d", quota.MaxNodeCount)
		}
		if quota.MinNodeCount != 1 {
			t.Errorf("expected MinNodeCount = 1, got %d", quota.MinNodeCount)
		}
	})

	t.Run("pd tso rejects defaultNodeCount in phase 1", func(t *testing.T) {
		ruConfig.Components["pd_tso"].NodeSpecRuRange["4c8g"].DefaultNodeCount = uint32Ptr(1)
		defer func() {
			ruConfig.Components["pd_tso"].NodeSpecRuRange["4c8g"].DefaultNodeCount = nil
		}()

		_, err := s.genDefaultPDTsoQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error")
		}
	})

	t.Run("pd scheduling rejects defaultNodeCount in phase 1", func(t *testing.T) {
		ruConfig.Components["pd_scheduling"].NodeSpecRuRange["4c8g"].DefaultNodeCount = uint32Ptr(1)
		defer func() {
			ruConfig.Components["pd_scheduling"].NodeSpecRuRange["4c8g"].DefaultNodeCount = nil
		}()

		_, err := s.genDefaultPDSchedulingQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error")
		}
	})
}

func TestGenDefaultPDQuotaRejectsDefaultNodeCount(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}
	totalRu := &entity.RequestUnit{Max: 100000}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"4C8G": {
			{
				Cpu:               4,
				MemoryGi:          8,
				AvailableCpu:      4,
				AvailableMemoryMi: 8192,
			},
		},
	}
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"pd": {
					MaxNodeCount: 3,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"4c8g": {Min: 0, Max: 0, DefaultNodeCount: uint32Ptr(1)},
					},
				},
			},
		},
	}

	_, err := s.genDefaultPDQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
	if err == nil {
		t.Fatal("expected error")
	}
}

func TestGenDefaultPDMicroserviceComponentPreservesExistingMinNodeCount(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"pd_tso": {
					MaxNodeCount: 2,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"4c8g": {Min: 0, Max: 0},
					},
				},
				"pd_scheduling": {
					MaxNodeCount: 2,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"4c8g": {Min: 0, Max: 0},
					},
				},
			},
		},
	}
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"4C8G": {
			{
				Cpu:               4,
				MemoryGi:          8,
				AvailableCpu:      4,
				AvailableMemoryMi: 8192,
			},
		},
	}
	totalRu := &entity.RequestUnit{Max: 100000}

	t.Run("pd tso preserves existing min during regeneration", func(t *testing.T) {
		comp := &entity.Component{
			Id:            "default",
			ComponentType: apiv1.ComponentType_PD_TSO,
			Phase:         apiv1.Component_AVAILABLE,
			NodeQuota: &entity.NodeQuota{
				NodeCount:    2,
				MinNodeCount: 2,
				MaxNodeCount: 2,
				NodeSpec:     &entity.NodeSpec{Cpu: 2, MemoryGi: 4},
			},
		}

		if err := s.genDefaultPDTsoComponent(ctx, nil, comp, totalRu, nodeSpecMap, ruConfig, false); err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		if comp.GetNodeQuota().GetMinNodeCount() != 2 {
			t.Errorf("expected existing MinNodeCount = 2 to be preserved, got %d", comp.GetNodeQuota().GetMinNodeCount())
		}
		if comp.GetNodeQuota().GetNodeSpec().GetCpu() != 4 {
			t.Errorf("expected regenerated node spec CPU = 4, got %d", comp.GetNodeQuota().GetNodeSpec().GetCpu())
		}
	})

	t.Run("pd scheduling preserves existing min during regeneration", func(t *testing.T) {
		comp := &entity.Component{
			Id:            "default",
			ComponentType: apiv1.ComponentType_PD_SCHEDULING,
			Phase:         apiv1.Component_AVAILABLE,
			NodeQuota: &entity.NodeQuota{
				NodeCount:    2,
				MinNodeCount: 2,
				MaxNodeCount: 2,
				NodeSpec:     &entity.NodeSpec{Cpu: 2, MemoryGi: 4},
			},
		}

		if err := s.genDefaultPDSchedulingComponent(ctx, nil, comp, totalRu, nodeSpecMap, ruConfig, false); err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		if comp.GetNodeQuota().GetMinNodeCount() != 2 {
			t.Errorf("expected existing MinNodeCount = 2 to be preserved, got %d", comp.GetNodeQuota().GetMinNodeCount())
		}
		if comp.GetNodeQuota().GetNodeSpec().GetCpu() != 4 {
			t.Errorf("expected regenerated node spec CPU = 4, got %d", comp.GetNodeQuota().GetNodeSpec().GetCpu())
		}
	})
}

func TestGenDefaultResourceManagerQuota(t *testing.T) {
	ctx := context.Background()
	s := &serverImpl{}

	// Create a valid ruConfig with resource_manager
	ruConfig := &entity.RuConfig{
		RuConfig: apiv1.RuConfig{
			Components: map[string]*apiv1.RuConfig_ComponentsMap{
				"resource_manager": {
					MaxNodeCount: 2,
					NodeSpecRuRange: map[string]*apiv1.RuConfig_RuRange{
						"2c4g": {Min: 0, Max: 0}, // 0 means unlimited
					},
				},
			},
		},
	}

	// Create nodeSpecMap with 2C4G spec (uppercase key format)
	nodeSpecMap := map[string][]*entity.NodeSpec{
		"2C4G": {
			{
				Cpu:               2,
				MemoryGi:          4,
				AvailableCpu:      2,
				AvailableMemoryMi: 4096,
			},
		},
	}

	t.Run("generate resource manager quota with valid config", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Max: 100000}

		quota, err := s.genDefaultResourceManagerQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err != nil {
			t.Fatalf("unexpected error: %v", err)
		}

		// Verify fixed count is 2
		if quota.NodeCount != 2 {
			t.Errorf("expected NodeCount = 2, got %d", quota.NodeCount)
		}
		if quota.MaxNodeCount != 2 {
			t.Errorf("expected MaxNodeCount = 2, got %d", quota.MaxNodeCount)
		}
		if quota.MinNodeCount != 2 {
			t.Errorf("expected MinNodeCount = 2, got %d", quota.MinNodeCount)
		}

		// Verify node spec is 2c4g
		if quota.NodeSpec.Cpu != 2 {
			t.Errorf("expected CPU = 2, got %d", quota.NodeSpec.Cpu)
		}
		if quota.NodeSpec.MemoryGi != 4 {
			t.Errorf("expected MemoryGi = 4, got %d", quota.NodeSpec.MemoryGi)
		}
	})

	t.Run("rejects defaultNodeCount in phase 1", func(t *testing.T) {
		totalRu := &entity.RequestUnit{Max: 100000}
		ruConfig.Components["resource_manager"].NodeSpecRuRange["2c4g"].DefaultNodeCount = uint32Ptr(1)
		defer func() {
			ruConfig.Components["resource_manager"].NodeSpecRuRange["2c4g"].DefaultNodeCount = nil
		}()

		_, err := s.genDefaultResourceManagerQuota(ctx, totalRu, nodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error")
		}
	})

	t.Run("error when resource_manager not in ruConfig", func(t *testing.T) {
		emptyRuConfig := &entity.RuConfig{
			RuConfig: apiv1.RuConfig{
				Components: map[string]*apiv1.RuConfig_ComponentsMap{},
			},
		}
		totalRu := &entity.RequestUnit{Max: 100000}

		_, err := s.genDefaultResourceManagerQuota(ctx, totalRu, nodeSpecMap, emptyRuConfig, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error when resource_manager not in ruConfig")
		}

		// Verify it's an Internal error
		st, ok := status.FromError(err)
		if !ok {
			t.Fatalf("expected gRPC status error, got %v", err)
		}
		if st.Code() != codes.Internal {
			t.Errorf("expected Internal error code, got %v", st.Code())
		}
	})

	t.Run("error when nodeSpec not found in nodeSpecMap", func(t *testing.T) {
		emptyNodeSpecMap := map[string][]*entity.NodeSpec{}
		totalRu := &entity.RequestUnit{Max: 100000}

		_, err := s.genDefaultResourceManagerQuota(ctx, totalRu, emptyNodeSpecMap, ruConfig, apiv1.Component_CREATING)
		if err == nil {
			t.Fatal("expected error when nodeSpec not in nodeSpecMap")
		}
	})
}

func TestResourceManagerComponentCreationLogic(t *testing.T) {
	// Test the logic: only create new component if feature flag is enabled,
	// but always process existing component

	t.Run("component not created when feature flag disabled and not exists", func(t *testing.T) {
		// When feature flag is disabled and component doesn't exist,
		// component should NOT be created
		ruConfig := &entity.RuConfig{
			RuConfig: apiv1.RuConfig{
				FeatureFlags: &apiv1.RuConfig_FeatureFlags{
					EnableResourceManager: false,
				},
				Components: map[string]*apiv1.RuConfig_ComponentsMap{},
			},
		}

		// Simulate the logic in generatedSharedPoolComponent
		defaultComponentMap := make(map[apiv1.ComponentType]*entity.Component)
		var resourceManagerComp *entity.Component

		_, ok := defaultComponentMap[apiv1.ComponentType_RESOURCE_MANAGER]
		if !ok && ruConfig.GetFeatureFlags().GetEnableResourceManager() {
			resourceManagerComp = &entity.Component{
				ComponentType: apiv1.ComponentType_RESOURCE_MANAGER,
			}
		}

		if resourceManagerComp != nil {
			t.Error("resource manager component should NOT be created when feature flag is disabled")
		}
	})

	t.Run("component created when feature flag enabled and not exists", func(t *testing.T) {
		// When feature flag is enabled and component doesn't exist,
		// component should be created
		ruConfig := &entity.RuConfig{
			RuConfig: apiv1.RuConfig{
				FeatureFlags: &apiv1.RuConfig_FeatureFlags{
					EnableResourceManager: true,
				},
				Components: map[string]*apiv1.RuConfig_ComponentsMap{},
			},
		}

		// Simulate the logic in generatedSharedPoolComponent
		defaultComponentMap := make(map[apiv1.ComponentType]*entity.Component)
		var resourceManagerComp *entity.Component

		_, ok := defaultComponentMap[apiv1.ComponentType_RESOURCE_MANAGER]
		if !ok && ruConfig.GetFeatureFlags().GetEnableResourceManager() {
			resourceManagerComp = &entity.Component{
				ComponentType: apiv1.ComponentType_RESOURCE_MANAGER,
			}
		}

		if resourceManagerComp == nil {
			t.Error("resource manager component should be created when feature flag is enabled")
		}
	})

	t.Run("existing component processed regardless of feature flag", func(t *testing.T) {
		// When component already exists, it should be processed
		// regardless of feature flag status
		ruConfig := &entity.RuConfig{
			RuConfig: apiv1.RuConfig{
				FeatureFlags: &apiv1.RuConfig_FeatureFlags{
					EnableResourceManager: false, // flag is disabled
				},
				Components: map[string]*apiv1.RuConfig_ComponentsMap{},
			},
		}

		// Simulate existing component in defaultComponentMap
		existingComp := &entity.Component{
			ComponentType: apiv1.ComponentType_RESOURCE_MANAGER,
			Phase:         apiv1.Component_AVAILABLE,
		}
		defaultComponentMap := map[apiv1.ComponentType]*entity.Component{
			apiv1.ComponentType_RESOURCE_MANAGER: existingComp,
		}

		// Simulate the logic in generatedSharedPoolComponent
		resourceManagerComp, ok := defaultComponentMap[apiv1.ComponentType_RESOURCE_MANAGER]
		if !ok && ruConfig.GetFeatureFlags().GetEnableResourceManager() {
			resourceManagerComp = &entity.Component{
				ComponentType: apiv1.ComponentType_RESOURCE_MANAGER,
			}
		}

		// Existing component should be found and processed
		if resourceManagerComp == nil {
			t.Error("existing resource manager component should be processed even when feature flag is disabled")
		}
		if resourceManagerComp != existingComp {
			t.Error("should use existing component, not create new one")
		}
	})
}