do not enable/disable dag deploy if already enabled/disabled (#880)

cloud/deployment/deployment.go

@@ -462,10 +462,18 @@ func Update(deploymentID, label, ws, description, deploymentName, dagDeploy stri
 	}
 
 	if dagDeploy == "enable" {
+		if currentDeployment.DagDeployEnabled {
+			fmt.Println("\nDAG-only deploys is already enabled for this deployment.")
+			return nil
+		}
 		fmt.Printf("\nYou enabled DAG-only deploys for this Deployment. Running tasks will not be interrupted and new tasks will continue to be scheduled." +
 			"\nRun `astro deploy --dags` after this command to push new changes. It may take a few minutes for the Airflow UI to update..\n\n")
 		deploymentUpdate.DagDeployEnabled = true
 	} else if dagDeploy == "disable" {
+		if !currentDeployment.DagDeployEnabled {
+			fmt.Println("\nDAG-only deploys is already disabled for this deployment.")
+			return nil
+		}
 		if config.CFG.ShowWarnings.GetBool() {
 			i, _ := input.Confirm("\nWarning: This command will disable DAG-only deploys for this Deployment. Running tasks will not be interrupted, but new tasks will not be scheduled" +
 				"\nRun `astro deploy` after this command to restart your DAGs. It may take a few minutes for the Airflow UI to update." +

cloud/deployment/deployment_test.go

@@ -743,8 +743,52 @@ func TestUpdate(t *testing.T) {
 		mockClient.AssertExpectations(t)
 	})
 
+	t.Run("do not update deployment to enable dag deploy if already enabled", func(t *testing.T) {
+		mockClient := new(astro_mocks.Client)
+		deploymentResp = astro.Deployment{
+			ID:               "test-id",
+			DagDeployEnabled: true,
+		}
+
+		mockClient.On("ListDeployments", org, ws).Return([]astro.Deployment{deploymentResp}, nil).Once()
+		err := Update("test-id", "", ws, "", "", "enable", 5, 3, []astro.WorkerQueue{}, true, mockClient)
+
+		assert.NoError(t, err)
+		mockClient.AssertExpectations(t)
+	})
+
 	t.Run("update deployment to disable dag deploy", func(t *testing.T) {
 		mockClient := new(astro_mocks.Client)
+		deploymentResp = astro.Deployment{
+			ID:               "test-id",
+			RuntimeRelease:   astro.RuntimeRelease{Version: "4.2.5"},
+			DeploymentSpec:   astro.DeploymentSpec{Scheduler: astro.Scheduler{AU: 5, Replicas: 3}},
+			DagDeployEnabled: true,
+			Cluster: astro.Cluster{
+				NodePools: []astro.NodePool{
+					{
+						ID:               "test-node-pool-id",
+						IsDefault:        true,
+						NodeInstanceType: "test-default-node-pool",
+						CreatedAt:        time.Time{},
+					},
+				},
+			},
+			WorkerQueues: []astro.WorkerQueue{
+				{
+					ID:         "test-queue-id",
+					Name:       "default",
+					IsDefault:  true,
+					NodePoolID: "test-default-node-pool",
+				},
+				{
+					Name:       "test-queue",
+					IsDefault:  false,
+					NodePoolID: "test-node-pool-id",
+				},
+			},
+		}
+
 		deploymentUpdateInput := astro.UpdateDeploymentInput{
 			ID:               "test-id",
 			ClusterID:        "",
@@ -770,6 +814,20 @@ func TestUpdate(t *testing.T) {
 		assert.NoError(t, err)
 		mockClient.AssertExpectations(t)
 	})
+
+	t.Run("do not update deployment to disable dag deploy if already disabled", func(t *testing.T) {
+		mockClient := new(astro_mocks.Client)
+		deploymentResp = astro.Deployment{
+			ID:               "test-id",
+			DagDeployEnabled: false,
+		}
+
+		mockClient.On("ListDeployments", org, ws).Return([]astro.Deployment{deploymentResp}, nil).Once()
+		err := Update("test-id", "", ws, "", "", "disable", 5, 3, []astro.WorkerQueue{}, true, mockClient)
+
+		assert.NoError(t, err)
+		mockClient.AssertExpectations(t)
+	})
 }
 
 func TestDelete(t *testing.T) {