package dag

import (
	"testing"
	"time"

	"github.com/stretchr/testify/assert"
	"github.com/stretchr/testify/require"
	"github.com/tidbcloud/cluster-service-ng/domain/workflow/action"
	braction "github.com/tidbcloud/cluster-service-ng/domain/workflow/action/br"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/actions"
	"github.com/tidbcloud/cluster-service-ng/pkg/runtime/dag"
)

type alwaysPendingAction struct{}

func (alwaysPendingAction) Name() string                                 { return "always-pending" }
func (alwaysPendingAction) Start(actions.ActionArgs) error               { return nil }
func (alwaysPendingAction) CheckFinish(actions.ActionArgs) (bool, error) { return false, nil }

func TestCoprocessorWorkerWaitActionSupportsShortTimeout(t *testing.T) {
	wrapped := newCoprocessorWorkerWaitAction(alwaysPendingAction{}, time.Millisecond, time.Nanosecond)
	start := time.Now()
	err := wrapped.Invoke(actions.ActionArgs{})
	require.ErrorContains(t, err, "timeout")
	assert.Less(t, time.Since(start), time.Second)
}

type pendingThenDoneAction struct {
	checks int
}

func (*pendingThenDoneAction) Name() string                   { return "pending-then-done" }
func (*pendingThenDoneAction) Start(actions.ActionArgs) error { return nil }
func (a *pendingThenDoneAction) CheckFinish(actions.ActionArgs) (bool, error) {
	a.checks++
	return a.checks == 3, nil
}

func TestCoprocessorWorkerWaitActionRetriesUntilDone(t *testing.T) {
	asyncAction := &pendingThenDoneAction{}
	wrapped := newCoprocessorWorkerWaitAction(asyncAction, time.Millisecond, time.Second)
	require.NoError(t, wrapped.Invoke(actions.ActionArgs{}))
	assert.Equal(t, 3, asyncAction.checks)
}

func TestInitDagPlans(t *testing.T) {
	dagPlans := dag.NewPlans()

	// init dag
	initDagPlans(dagPlans)
}

func TestCassandraProxyDagOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanCreateCassandraProxy)

	assertHasOnlyChild(t, plan, PlanCreateCassandraProxy, action.SendMeteringMetadataActionName)
	assertHasOnlyChild(t, plan, action.SendMeteringMetadataActionName, action.WaitCassandraProxyReadyActionName)
	assertHasOnlyChild(t, plan, action.WaitCassandraProxyReadyActionName, action.ApplyCassandraProxyPLSActionName)
	assertHasOnlyChild(t, plan, action.ApplyCassandraProxyPLSActionName, action.ApplyCassandraProxyLBActionName)
	assertHasOnlyChild(t, plan, action.ApplyCassandraProxyLBActionName, action.ApplyCassandraProxyActionName)
}

func TestUpdateBackupGCSettingUsesTransitionSafeOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanUpdateBackupGcSetting)

	// Runtime is leaf-first: raise TTL for both delays, update TiBRGC, then
	// apply the final lifecycle rules for the desired delay.
	assertHasOnlyChild(t, plan, PlanUpdateBackupGcSetting, braction.ApplyBucketLifecycleRulesActionName)
	assertHasOnlyChild(t, plan, braction.ApplyBucketLifecycleRulesActionName, braction.ApplyTiBRGCActionName)
	assertHasOnlyChild(t, plan, braction.ApplyTiBRGCActionName, braction.ApplyTransitionSafeLifecycleRulesActionName)
}

func TestUpgradeTiBRAutoBackupValidatesBeforeDelete(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanUpgradeTiBRAutoBackup)

	// Runtime is leaf-first: validate placement, delete the current CR, then
	// create and wait for the desired CR.
	assertHasOnlyChild(t, plan, PlanUpgradeTiBRAutoBackup, braction.CreateTiBRAutoBackupActionName)
	assertHasOnlyChild(t, plan, braction.CreateTiBRAutoBackupActionName, braction.DeleteTiBRAutoBackupActionName)
	assertHasOnlyChild(t, plan, braction.DeleteTiBRAutoBackupActionName, braction.ValidateTiBRAutoBackupPlacementActionName)
}

func TestCassandraProxyDagReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanCreateCassandraProxy)
	assignMockActions(t, plan,
		action.SendMeteringMetadataActionName,
		action.WaitCassandraProxyReadyActionName,
		action.ApplyCassandraProxyPLSActionName,
		action.ApplyCassandraProxyLBActionName,
		action.ApplyCassandraProxyActionName,
	)

	done := map[string]bool{}
	ready := readyActionNames(t, plan, done)
	assert.ElementsMatch(t, []string{action.ApplyCassandraProxyActionName}, ready)

	done[action.ApplyCassandraProxyActionName] = true
	ready = readyActionNames(t, plan, done)
	assert.ElementsMatch(t, []string{action.ApplyCassandraProxyLBActionName}, ready)

	done[action.ApplyCassandraProxyLBActionName] = true
	ready = readyActionNames(t, plan, done)
	assert.ElementsMatch(t, []string{action.ApplyCassandraProxyPLSActionName}, ready)

	done[action.ApplyCassandraProxyPLSActionName] = true
	ready = readyActionNames(t, plan, done)
	assert.ElementsMatch(t, []string{action.WaitCassandraProxyReadyActionName}, ready)

	done[action.WaitCassandraProxyReadyActionName] = true
	ready = readyActionNames(t, plan, done)
	assert.ElementsMatch(t, []string{action.SendMeteringMetadataActionName}, ready)
}

func TestDeleteTiDBClusterDagIncludesCassandraProxyApply(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDeleteTiDBCluster)

	assertHasOnlyChild(t, plan, action.SendMeteringMetadataActionName, action.DeleteTiDBClusterMetaName)
	assertHasOnlyChild(t, plan, action.DeleteTiDBClusterMetaName, action.CleanUpCertActionName)
	assertHasOnlyChild(t, plan, action.CleanUpCertActionName, action.DeleteKeyspaceRoleActionName)
	assertHasOnlyChild(t, plan, action.DeleteKeyspaceRoleActionName, action.DeleteKeySpaceActionName)
	assertHasOnlyChild(t, plan, action.DeleteKeySpaceActionName, action.DeletePlacementPolicyActionName)
	assertHasOnlyChild(t, plan, action.DeletePlacementPolicyActionName, action.ApplyTiProxyGroupActionName)
	assertHasOnlyChild(t, plan, action.ApplyTiProxyGroupActionName, action.ApplyCassandraProxyActionName)
	assertHasOnlyChild(t, plan, action.ApplyCassandraProxyActionName, action.DeleteCassandraProxyLBActionName)
	assertHasOnlyChild(t, plan, action.DeleteCassandraProxyLBActionName, action.DeleteCassandraProxyPLSActionName)
	assertHasOnlyChild(t, plan, action.DeleteCassandraProxyPLSActionName, action.DeleteO11yTiDBGroupAgentsByClusterActionName)
}

func TestDeleteSharedPoolDagDeletesConfigCenterMeta(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDeleteSharedPool)

	assertHasOnlyChild(t, plan, action.SendMeteringMetadataActionName, action.DeleteSharedPoolMetaName)
	assertHasOnlyChild(t, plan, action.DeleteSharedPoolMetaName, action.PruneDataplaneActionName)
}

func TestPauseSharedPoolDagMeteringOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanPauseSharedPool)

	assertHasOnlyChild(t, plan, PlanPauseSharedPool, action.CompletePauseActionName)
	assertHasOnlyChild(t, plan, action.CompletePauseActionName, action.SendPausedMeteringMetadataActionName)
	assertHasOnlyChild(t, plan, action.SendPausingMeteringMetadataActionName, action.PausePrecheckActionName)
}

func TestResumeSharedPoolDagMeteringOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanResumeSharedPool)

	assertHasOnlyChild(t, plan, PlanResumeSharedPool, action.CompleteResumeActionName)
	assertHasOnlyChild(t, plan, action.CompleteResumeActionName, action.SendActiveMeteringMetadataActionName)
	assertHasOnlyChild(t, plan, action.ApplyResumeComputeActionName, action.SendResumingMeteringMetadataActionName)
}

func TestDeleteTiDBGroupDagDoesNotDeleteO11yTiDBGroupAgent(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDeleteTiDBGroup)

	assert.Nil(t, plan.Get(action.DeleteO11yTiDBGroupAgentActionName))
}

func TestCreateTiDBClusterDagKeyspaceRoleOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanCreateTiDBCluster)

	assertHasOnlyChild(t, plan, action.UpdateComponentGroupActionName, action.CreateKeyspaceRoleActionName)
	assertHasOnlyChild(t, plan, action.CreateKeyspaceRoleActionName, action.SetKeySpaceActionName)
}

func TestUpdateComponentDagEnsuresUpgradePolicyBeforeComponentUpdate(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanUpdateComponent)

	assertHasOnlyChild(
		t,
		plan,
		action.UpdateComponentGroupActionName,
		action.EnsureClusterUpgradePolicyActionName,
	)
	assertHasOnlyChild(
		t,
		plan,
		action.EnsureClusterUpgradePolicyActionName,
		action.CreateVolumeAttributesClassActionName,
	)
}

func TestDeleteTiKVGroupDagOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDeleteTiKVGroup)

	assertHasOnlyChild(t, plan, PlanDeleteTiKVGroup, action.SendMeteringMetadataActionName)
	assertHasOnlyChild(t, plan, action.SendMeteringMetadataActionName, action.CleanUpCertActionName)
	assertHasOnlyChild(t, plan, action.CleanUpCertActionName, action.CommitDeleteComponentActionName)
	assertHasOnlyChild(t, plan, action.CommitDeleteComponentActionName, action.DeleteTiKVComponentConfigActionName)
	assertHasOnlyChild(t, plan, action.DeleteTiKVComponentConfigActionName, action.WaitTiKVGroupGoneActionName)
	assertHasOnlyChild(t, plan, action.WaitTiKVGroupGoneActionName, action.ApplyComponentGroupActionName)
}

func TestDeleteTiKVGroupDagReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDeleteTiKVGroup)
	assignMockActions(t, plan,
		action.SendMeteringMetadataActionName,
		action.CleanUpCertActionName,
		action.CommitDeleteComponentActionName,
		action.DeleteTiKVComponentConfigActionName,
		action.WaitTiKVGroupGoneActionName,
		action.ApplyComponentGroupActionName,
	)

	done := map[string]bool{}
	ready := readyActionNamesForRoot(t, plan, PlanDeleteTiKVGroup, done)
	assert.ElementsMatch(t, []string{action.ApplyComponentGroupActionName}, ready)

	done[action.ApplyComponentGroupActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiKVGroup, done)
	assert.ElementsMatch(t, []string{action.WaitTiKVGroupGoneActionName}, ready)

	done[action.WaitTiKVGroupGoneActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiKVGroup, done)
	assert.ElementsMatch(t, []string{action.DeleteTiKVComponentConfigActionName}, ready)

	done[action.DeleteTiKVComponentConfigActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiKVGroup, done)
	assert.ElementsMatch(t, []string{action.CommitDeleteComponentActionName}, ready)

	done[action.CommitDeleteComponentActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiKVGroup, done)
	assert.ElementsMatch(t, []string{action.CleanUpCertActionName}, ready)

	done[action.CleanUpCertActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiKVGroup, done)
	assert.ElementsMatch(t, []string{action.SendMeteringMetadataActionName}, ready)
}

func TestSetSharedPoolPDDeploymentModeToMicroserviceDagReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanSetSharedPoolPDDeploymentModeToMicroservice)
	assignMockActions(t, plan,
		action.CreateClusterCertActionName,
		action.ApplyPDMicroserviceComponentsActionName,
		action.PatchPDDeploymentModeActionName,
		action.WaitPDMicroserviceComponentsReadyName,
		action.CommitPDDeploymentModeActionName,
	)

	done := map[string]bool{}
	ready := readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMicroservice, done)
	assert.ElementsMatch(t, []string{action.CreateClusterCertActionName}, ready)

	done[action.CreateClusterCertActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMicroservice, done)
	assert.ElementsMatch(t, []string{action.ApplyPDMicroserviceComponentsActionName}, ready)

	done[action.ApplyPDMicroserviceComponentsActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMicroservice, done)
	assert.ElementsMatch(t, []string{action.PatchPDDeploymentModeActionName}, ready)

	done[action.PatchPDDeploymentModeActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMicroservice, done)
	assert.ElementsMatch(t, []string{action.WaitPDMicroserviceComponentsReadyName}, ready)

	done[action.WaitPDMicroserviceComponentsReadyName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMicroservice, done)
	assert.ElementsMatch(t, []string{action.CommitPDDeploymentModeActionName}, ready)
}

func TestSetSharedPoolPDDeploymentModeToMonolithicDagReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanSetSharedPoolPDDeploymentModeToMonolithic)
	assignMockActions(t, plan,
		action.PatchPDDeploymentModeActionName,
		action.DeletePDMicroserviceComponentsActionName,
		action.WaitPDMicroserviceComponentsGoneName,
		action.CommitPDDeploymentModeActionName,
	)

	done := map[string]bool{}
	ready := readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMonolithic, done)
	assert.ElementsMatch(t, []string{action.PatchPDDeploymentModeActionName}, ready)

	done[action.PatchPDDeploymentModeActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMonolithic, done)
	assert.ElementsMatch(t, []string{action.DeletePDMicroserviceComponentsActionName}, ready)

	done[action.DeletePDMicroserviceComponentsActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMonolithic, done)
	assert.ElementsMatch(t, []string{action.WaitPDMicroserviceComponentsGoneName}, ready)

	done[action.WaitPDMicroserviceComponentsGoneName] = true
	ready = readyActionNamesForRoot(t, plan, PlanSetSharedPoolPDDeploymentModeToMonolithic, done)
	assert.ElementsMatch(t, []string{action.CommitPDDeploymentModeActionName}, ready)
}

func TestDeleteTiDBClusterDagCassandraProxyReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDeleteTiDBCluster)
	assignMockActions(t, plan,
		action.SendMeteringMetadataActionName,
		action.DeleteTiDBClusterMetaName,
		action.DeleteO11yTiDBGroupAgentsByClusterActionName,
		action.CleanUpCertActionName,
		action.DeleteKeyspaceRoleActionName,
		action.DeleteKeySpaceActionName,
		action.DeletePlacementPolicyActionName,
		action.ApplyTiProxyGroupActionName,
		action.ApplyCassandraProxyActionName,
		action.DeleteCassandraProxyLBActionName,
		action.DeleteCassandraProxyPLSActionName,
	)

	done := map[string]bool{}
	ready := readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeleteO11yTiDBGroupAgentsByClusterActionName)

	done[action.DeleteO11yTiDBGroupAgentsByClusterActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeleteCassandraProxyPLSActionName)

	done[action.DeleteCassandraProxyPLSActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeleteCassandraProxyLBActionName)

	done[action.DeleteCassandraProxyLBActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.ApplyCassandraProxyActionName)

	done[action.ApplyCassandraProxyActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.ApplyTiProxyGroupActionName)

	done[action.ApplyTiProxyGroupActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeletePlacementPolicyActionName)

	done[action.DeletePlacementPolicyActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeleteKeySpaceActionName)

	done[action.DeleteKeySpaceActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeleteKeyspaceRoleActionName)

	done[action.DeleteKeyspaceRoleActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.CleanUpCertActionName)

	done[action.CleanUpCertActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.DeleteTiDBClusterMetaName)

	done[action.DeleteTiDBClusterMetaName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDeleteTiDBCluster, done)
	assert.Contains(t, ready, action.SendMeteringMetadataActionName)
}

func assertHasOnlyChild(t *testing.T, plan dag.Dag, nodeName, childName string) {
	t.Helper()
	node := plan.Get(nodeName)
	require.NotNil(t, node)
	names := make([]string, 0, len(node.Children))
	for name := range node.Children {
		names = append(names, name)
	}
	assert.ElementsMatch(t, []string{childName}, names)
}

func assignMockActions(t *testing.T, plan dag.Dag, actionNames ...string) {
	t.Helper()
	for _, name := range actionNames {
		node := plan.Get(name)
		require.NotNil(t, node)
		require.NoError(t, node.AssignAction(&actions.MockActionBase{ActionName: name}, nil))
	}
}

func readyActionNames(t *testing.T, plan dag.Dag, done map[string]bool) []string {
	t.Helper()
	return readyActionNamesForRoot(t, plan, PlanCreateCassandraProxy, done)
}

func readyActionNamesForRoot(t *testing.T, plan dag.Dag, root string, done map[string]bool) []string {
	t.Helper()
	ready, _, err := plan.GetReadyActionTasks(root, func(action *dag.DagAction) (bool, error) {
		return done[action.Action.Name()], nil
	})
	require.NoError(t, err)
	names := make([]string, 0, len(ready))
	for _, action := range ready {
		names = append(names, action.Action.Name())
	}
	return names
}

func TestEnableCoprocessorWorkerDagOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanEnableCoprocessorWorker)

	// Verify dependency chain (SafeAddDependNode stores in reverse order)
	assertHasOnlyChild(t, plan, PlanEnableCoprocessorWorker, action.SendMeteringMetadataActionName)
	assertHasOnlyChild(t, plan, action.SendMeteringMetadataActionName, action.WaitTiKVGroupRemoteWorkerRolloutActionName)
	assertHasOnlyChild(t, plan, action.WaitTiKVGroupRemoteWorkerRolloutActionName, action.UpdateComponentGroupActionName)
	assertHasOnlyChild(t, plan, action.UpdateComponentGroupActionName, action.WaitCoprocessorWorkerReadyActionName)
	assertHasOnlyChild(t, plan, action.WaitCoprocessorWorkerReadyActionName, action.ApplyComponentHpaActionName)
	assertHasOnlyChild(t, plan, action.ApplyComponentHpaActionName, action.ApplyComponentGroupActionName)
	assertHasOnlyChild(t, plan, action.ApplyComponentGroupActionName, action.CreateVolumeAttributesClassActionName)
	assertHasOnlyChild(t, plan, action.CreateVolumeAttributesClassActionName, action.CreateClusterCertActionName)
}

func TestEnableCoprocessorWorkerDagReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanEnableCoprocessorWorker)
	assignMockActions(t, plan,
		action.SendMeteringMetadataActionName,
		action.WaitTiKVGroupRemoteWorkerRolloutActionName,
		action.UpdateComponentGroupActionName,
		action.ApplyComponentHpaActionName,
		action.WaitCoprocessorWorkerReadyActionName,
		action.ApplyComponentGroupActionName,
		action.CreateVolumeAttributesClassActionName,
		action.CreateClusterCertActionName,
	)

	// Execution order: cert -> VAC -> apply copr worker -> apply HPA -> wait ready -> patch TiKV -> wait rollout -> metering
	done := map[string]bool{}
	ready := readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.CreateClusterCertActionName}, ready)

	done[action.CreateClusterCertActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.CreateVolumeAttributesClassActionName}, ready)

	done[action.CreateVolumeAttributesClassActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.ApplyComponentGroupActionName}, ready)

	done[action.ApplyComponentGroupActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.ApplyComponentHpaActionName}, ready)

	done[action.ApplyComponentHpaActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.WaitCoprocessorWorkerReadyActionName}, ready)

	done[action.WaitCoprocessorWorkerReadyActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.UpdateComponentGroupActionName}, ready)

	done[action.UpdateComponentGroupActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.WaitTiKVGroupRemoteWorkerRolloutActionName}, ready)

	done[action.WaitTiKVGroupRemoteWorkerRolloutActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanEnableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.SendMeteringMetadataActionName}, ready)
}

func TestDisableCoprocessorWorkerDagOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDisableCoprocessorWorker)

	// Verify dependency chain (SafeAddDependNode stores in reverse order)
	assertHasOnlyChild(t, plan, PlanDisableCoprocessorWorker, action.SendMeteringMetadataActionName)
	assertHasOnlyChild(t, plan, action.SendMeteringMetadataActionName, action.CleanUpCertActionName)
	assertHasOnlyChild(t, plan, action.CleanUpCertActionName, action.WaitCoprocessorWorkerGoneActionName)
	assertHasOnlyChild(t, plan, action.WaitCoprocessorWorkerGoneActionName, action.ApplyComponentGroupActionName)
	assertHasOnlyChild(t, plan, action.ApplyComponentGroupActionName, action.ApplyComponentHpaActionName)
	assertHasOnlyChild(t, plan, action.ApplyComponentHpaActionName, action.WaitTiKVGroupRemoteWorkerRolloutActionName)
	assertHasOnlyChild(t, plan, action.WaitTiKVGroupRemoteWorkerRolloutActionName, action.UpdateComponentGroupActionName)
	assertHasOnlyChild(t, plan, action.UpdateComponentGroupActionName, action.WaitDefaultTiKVWorkerReadyActionName)
}

func TestDisableCoprocessorWorkerDagReadyOrder(t *testing.T) {
	dagPlans := dag.NewPlans()
	initDagPlans(dagPlans)
	plan := dagPlans.GetAndCloneDagByRoot(PlanDisableCoprocessorWorker)
	assignMockActions(t, plan,
		action.SendMeteringMetadataActionName,
		action.CleanUpCertActionName,
		action.WaitCoprocessorWorkerGoneActionName,
		action.ApplyComponentGroupActionName,
		action.ApplyComponentHpaActionName,
		action.WaitTiKVGroupRemoteWorkerRolloutActionName,
		action.UpdateComponentGroupActionName,
		action.WaitDefaultTiKVWorkerReadyActionName,
	)

	// Execution order: wait default worker ready -> patch TiKV -> wait rollout -> apply HPA delete -> apply copr worker deleting -> wait gone -> cleanup cert -> metering
	done := map[string]bool{}
	ready := readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.WaitDefaultTiKVWorkerReadyActionName}, ready)

	done[action.WaitDefaultTiKVWorkerReadyActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.UpdateComponentGroupActionName}, ready)

	done[action.UpdateComponentGroupActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.WaitTiKVGroupRemoteWorkerRolloutActionName}, ready)

	done[action.WaitTiKVGroupRemoteWorkerRolloutActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.ApplyComponentHpaActionName}, ready)

	done[action.ApplyComponentHpaActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.ApplyComponentGroupActionName}, ready)

	done[action.ApplyComponentGroupActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.WaitCoprocessorWorkerGoneActionName}, ready)

	done[action.WaitCoprocessorWorkerGoneActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.CleanUpCertActionName}, ready)

	done[action.CleanUpCertActionName] = true
	ready = readyActionNamesForRoot(t, plan, PlanDisableCoprocessorWorker, done)
	assert.ElementsMatch(t, []string{action.SendMeteringMetadataActionName}, ready)
}