package action

import (
	"testing"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/mock"
	"github.com/stretchr/testify/require"
	apiv1 "github.com/tidbcloud/cluster-service-ng/api/cluster-service-ng/v1beta"
	dplanemgrv1 "github.com/tidbcloud/cluster-service-ng/api/dataplane-manager/v1"
	"github.com/tidbcloud/cluster-service-ng/domain/entity"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow"
	dataplanemgrmocks "github.com/tidbcloud/cluster-service-ng/pkg/dataplanemgr/mocks"
	"github.com/tidbcloud/cluster-service-ng/pkg/nodegroup"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	repomnocks "github.com/tidbcloud/cluster-service-ng/pkg/storage/repo/mocks"
	"google.golang.org/grpc/codes"
	"google.golang.org/grpc/status"
	"google.golang.org/protobuf/encoding/protojson"
)

func mustMarshalNodeInstanceGroup(group *entity.NodeInstanceGroup) string {
	data, err := protojson.Marshal(group)
	if err != nil {
		panic(err)
	}
	return string(data)
}

func TestCreateNodeInstanceGroupAction_Name(t *testing.T) {
	stateRepo := repomnocks.NewDataplaneNodeInstanceGroupStateRepo(t)
	lockRepo := repomnocks.NewDataplaneNodeInstanceGroupLockRepo(t)
	dataplaneClient := dataplanemgrmocks.NewClient(t)
	dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)

	action := NewCreateNodeInstanceGroupAction(stateRepo, lockRepo, dataplaneClient)

	expected := CreateNodeInstanceGroupActionName
	assert.Equal(t, expected, action.Name())
}

func TestCreateNodeInstanceGroupAction_Invoke(t *testing.T) {
	tests := []struct {
		name          string
		args          actions.ActionArgs
		setupMocks    func(*repomnocks.DataplaneNodeInstanceGroupStateRepo, *repomnocks.DataplaneNodeInstanceGroupLockRepo, *dataplanemgrmocks.Client)
		wantCode      codes.Code
		errorContains string
	}{
		{
			name: "success defaults missing placement group enabled to false",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-abc"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-abc", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-abc", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				// No sibling NodePools → fall back to DAG arg (= false here, no PG arg set).
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: nil}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.CreateNodePoolRequest) bool {
					return req.DataPlaneId == "dataplane-abc" &&
						req.NodePool.InstanceType == "m5.xlarge" &&
						req.NodePool.NodeType == dplanemgrv1.NodeType_Dedicated &&
						!req.NodePool.PlacementGroupEnabled
				})).Return(&dplanemgrv1.NodePool{}, nil)

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-abc", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
		{
			name: "success passes placement group enabled true",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-pg"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
				workflow.PlacementGroupEnabledKey: actions.ActionArgBool(true),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-pg", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-pg", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				// No sibling → use DAG arg (= true).
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: nil}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.CreateNodePoolRequest) bool {
					return req.DataPlaneId == "dataplane-pg" && req.NodePool.PlacementGroupEnabled
				})).Return(&dplanemgrv1.NodePool{}, nil)

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-pg", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
		{
			name: "malformed placement group enabled returns invalid argument",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-bad-pg"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
				workflow.PlacementGroupEnabledKey: actions.ActionArgString("not-bool"),
			},
			wantCode:      codes.InvalidArgument,
			errorContains: "invalid placement_group_enabled",
		},
		{
			name: "missing dataplane id",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString(""),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
			},
			setupMocks:    nil,
			wantCode:      codes.InvalidArgument,
			errorContains: "dataplane_id",
		},
		{
			name: "create nodepool fails",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-err"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-err", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-err", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: nil}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.AnythingOfType("*dedicatedv1.CreateNodePoolRequest")).
					Return(nil, status.Errorf(codes.Internal, "internal error"))
			},
			wantCode:      codes.Internal,
			errorContains: "failed to create nodepool",
		},
		{
			name: "already exists - ignored",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-exists"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-exists", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-exists", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: nil}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.AnythingOfType("*dedicatedv1.CreateNodePoolRequest")).
					Return(nil, status.Errorf(codes.AlreadyExists, "nodepool already exists"))

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-exists", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
		{
			name: "skips creation for deleting dataplane",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-deleting"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-deleting", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-deleting", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Deleting}, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: nil}, nil)
				// CreateNodePool should NOT be called.
			},
			wantCode: codes.OK,
		},
		{
			name: "skips creation when dataplane not found",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-not-found"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-not-found", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-not-found", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(nil, status.Errorf(codes.NotFound, "dataplane not found"))
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{NodePools: nil}, nil)
				// CreateNodePool should NOT be called when dataplane is not found.
			},
			wantCode: codes.OK,
		},
		{
			// fgksgf's regression: legacy single-AZ EKS dataplane where the
			// primary pool was created before this design with
			// Spec.PlacementGroupEnabled=false. The fallback NodePool must
			// inherit that false from the sibling on the wire, otherwise DPM
			// would honour the SharedPool gate's topology-derived true (if
			// the SharedPool is new) and silently migrate the fallback onto
			// the partition path while the primary stays on the legacy
			// per-zone path.
			name: "inherits false from legacy single-AZ sibling when DAG arg is true",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-legacy"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
				// SharedPool gate on a newly created SharedPool: topology-derived true.
				workflow.PlacementGroupEnabledKey: actions.ActionArgBool(true),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-legacy", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-legacy", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				// Sibling list: legacy primary (workload, PlacementGroupEnabled=false)
				// + builtin admin pool (no Spec.Labels). The builtin is filtered out
				// by the LabelNodeSpec discriminator; the legacy primary's value
				// wins.
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{Name: "primary-legacy", Labels: map[string]string{nodegroup.LabelNodeSpec: "4C16G"}, PlacementGroupEnabled: false},
						{Name: "builtin-admin", PlacementGroupEnabled: false},
					},
				}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.CreateNodePoolRequest) bool {
					return req.DataPlaneId == "dataplane-legacy" && !req.NodePool.PlacementGroupEnabled
				})).Return(&dplanemgrv1.NodePool{}, nil)

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-legacy", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
		{
			// Symmetric: legacy primary has PlacementGroupEnabled=true (the
			// SharedPool gate for a new AWS single-AZ SharedPool is true),
			// and the new sibling must inherit that true.
			name: "inherits true from sibling when DAG arg is also true",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-sibling-true"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
				workflow.PlacementGroupEnabledKey: actions.ActionArgBool(true),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-sibling-true", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-sibling-true", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						{Name: "primary-true", Labels: map[string]string{nodegroup.LabelNodeSpec: "4C16G"}, PlacementGroupEnabled: true},
					},
				}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.CreateNodePoolRequest) bool {
					return req.DataPlaneId == "dataplane-sibling-true" && req.NodePool.PlacementGroupEnabled
				})).Return(&dplanemgrv1.NodePool{}, nil)

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-sibling-true", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
		{
			// First NodePool in a fresh dataplane: only builtin admin pools
			// exist (no Spec.Labels), so the LabelNodeSpec discriminator
			// filters them all out and the helper falls back to the DAG arg.
			// A new SharedPool on a single-AZ EKS dataplane has gate=true,
			// so the first NodePool ends up with PlacementGroupEnabled=true.
			name: "first nodepool on fresh dataplane uses DAG arg after builtin filter",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-fresh"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
				workflow.PlacementGroupEnabledKey: actions.ActionArgBool(true),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-fresh", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-fresh", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).Return(&dplanemgrv1.ListNodePoolsResponse{
					NodePools: []*dplanemgrv1.NodePool{
						// Only built-in admin pools; no LabelNodeSpec discriminator.
						{Name: "builtin-admin-1"},
						{Name: "builtin-admin-2"},
					},
				}, nil)
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.CreateNodePoolRequest) bool {
					return req.DataPlaneId == "dataplane-fresh" && req.NodePool.PlacementGroupEnabled
				})).Return(&dplanemgrv1.NodePool{}, nil)

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-fresh", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
		{
			// ListNodePools fails (transient RPC error): helper logs a warning
			// and falls back to the DAG arg so a flaky DPM does not block
			// NodePool creation. This is the resilience path fgksgf called
			// out for the still-open finding.
			name: "listnodepools failure falls back to DAG arg",
			args: actions.ActionArgs{
				workflow.DataplaneIdKey: actions.ActionArgString("dataplane-flaky"),
				workflow.NodeInstanceGroupKey: actions.ActionArgString(mustMarshalNodeInstanceGroup(&entity.NodeInstanceGroup{
					NodeSpec:            &apiv1.NodeSpec{Cpu: 4, MemoryGi: 16},
					PrimaryInstanceType: "m5.xlarge",
				})),
				workflow.PlacementGroupEnabledKey: actions.ActionArgBool(true),
			},
			setupMocks: func(stateRepo *repomnocks.DataplaneNodeInstanceGroupStateRepo, lockRepo *repomnocks.DataplaneNodeInstanceGroupLockRepo, dataplaneClient *dataplanemgrmocks.Client) {
				lock := &entity.DataplaneNodeInstanceGroupLock{}
				lock.FencingToken = 1
				lockRepo.On("TryAcquire", mock.Anything, "dataplane-flaky", "4C16G", "m5.xlarge", mock.AnythingOfType("string"), DefaultLockLeaseDuration).Return(lock, nil)
				lockRepo.On("Release", mock.Anything, "dataplane-flaky", "4C16G", "m5.xlarge", mock.AnythingOfType("string")).Return(nil)

				dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)
				dataplaneClient.On("GetDataPlane", mock.Anything, mock.Anything).Return(&dplanemgrv1.DataPlane{Status: dplanemgrv1.DataPlane_Active}, nil)
				dataplaneClient.On("ListNodePools", mock.Anything, mock.Anything).
					Return(nil, status.Errorf(codes.Unavailable, "transient RPC error"))
				// No siblings → use DAG arg (= true).
				dataplaneClient.On("CreateNodePool", mock.Anything, mock.MatchedBy(func(req *dplanemgrv1.CreateNodePoolRequest) bool {
					return req.DataPlaneId == "dataplane-flaky" && req.NodePool.PlacementGroupEnabled
				})).Return(&dplanemgrv1.NodePool{}, nil)

				state := &entity.DataplaneNodeInstanceGroupState{}
				state.ID = "state-123"
				stateRepo.On("LoadOrCreate", mock.Anything, "dataplane-flaky", "4C16G", "m5.xlarge").Return(state, nil)
			},
			wantCode: codes.OK,
		},
	}

	for _, tt := range tests {
		t.Run(tt.name, func(t *testing.T) {
			stateRepo := repomnocks.NewDataplaneNodeInstanceGroupStateRepo(t)
			lockRepo := repomnocks.NewDataplaneNodeInstanceGroupLockRepo(t)
			dataplaneClient := dataplanemgrmocks.NewClient(t)
			dataplaneClient.On("GetDataPlaneServiceClient").Return(dataplaneClient)

			if tt.setupMocks != nil {
				tt.setupMocks(stateRepo, lockRepo, dataplaneClient)
			}

			action := NewCreateNodeInstanceGroupAction(stateRepo, lockRepo, dataplaneClient)
			err := action.Invoke(tt.args)
			if tt.wantCode == codes.OK {
				require.NoError(t, err)
				return
			}
			require.Error(t, err)
			assert.Equal(t, tt.wantCode, status.Code(err))
			if tt.errorContains != "" {
				assert.Contains(t, err.Error(), tt.errorContains)
			}
		})
	}
}