Throw error on custom deploy when dag deploy is enabled (#824)

cloud/deploy/deploy.go

@@ -63,7 +63,10 @@ var (
 	azureUploader        = azure.Upload
 )
 
-var errDagsParseFailed = errors.New("your local DAGs did not parse. Fix the listed errors or use `astro deploy [deployment-id] -f` to force deploy") //nolint:revive
+var (
+	errDagsParseFailed = errors.New("your local DAGs did not parse. Fix the listed errors or use `astro deploy [deployment-id] -f` to force deploy")            //nolint:revive
+	customDeployError  = errors.New("Dag Deploy is enabled. To perform custom image deploy, you will need to contact Astronomer Support to disable dag deploy") //nolint:revive
+)
 
 type deploymentInfo struct {
 	deploymentID     string
@@ -525,6 +528,9 @@ func buildImage(c *config.Context, path, currentVersion, deployImage, imageName
 			}
 		}
 	} else {
+		if dagDeployEnabled {
+			return "", customDeployError
+		}
 		// skip build if an imageName is passed
 		fmt.Println(composeSkipImageBuildingPromptMsg)
 

cloud/deploy/deploy_test.go

@@ -158,10 +158,10 @@ func TestDeployWithDagsDeploySuccess(t *testing.T) {
 
 	mockClient.On("GetDeployment", mock.Anything).Return(mockDeplyResp, nil).Times(5)
 	mockClient.On("ListDeployments", org, ws).Return([]astro.Deployment{{ID: "test-id", DagDeployEnabled: true}}, nil).Times(1)
-	mockClient.On("GetDeploymentConfig").Return(astro.DeploymentConfig{RuntimeReleases: []astro.RuntimeRelease{{Version: "4.2.5"}}}, nil).Times(6)
-	mockClient.On("CreateImage", mock.Anything).Return(&astro.Image{}, nil).Times(6)
-	mockClient.On("DeployImage", mock.Anything).Return(&astro.Image{}, nil).Times(6)
-	mockClient.On("InitiateDagDeployment", astro.InitiateDagDeploymentInput{RuntimeID: runtimeID}).Return(astro.InitiateDagDeployment{ID: initiatedDagDeploymentID, DagURL: dagURL}, nil).Times(2)
+	mockClient.On("GetDeploymentConfig").Return(astro.DeploymentConfig{RuntimeReleases: []astro.RuntimeRelease{{Version: "4.2.5"}}}, nil).Times(5)
+	mockClient.On("CreateImage", mock.Anything).Return(&astro.Image{}, nil).Times(5)
+	mockClient.On("DeployImage", mock.Anything).Return(&astro.Image{}, nil).Times(5)
+	mockClient.On("InitiateDagDeployment", astro.InitiateDagDeploymentInput{RuntimeID: runtimeID}).Return(astro.InitiateDagDeployment{ID: initiatedDagDeploymentID, DagURL: dagURL}, nil).Times(5)
 
 	azureUploader = func(sasLink string, file io.Reader) (string, error) {
 		return "version-id", nil
@@ -175,14 +175,13 @@ func TestDeployWithDagsDeploySuccess(t *testing.T) {
 		Status:                   "SUCCEEDED",
 		Message:                  "Dags uploaded successfully",
 	}
-	mockClient.On("ReportDagDeploymentStatus", reportDagDeploymentStatusInput).Return(astro.DagDeploymentStatus{}, nil).Times(2)
+	mockClient.On("ReportDagDeploymentStatus", reportDagDeploymentStatusInput).Return(astro.DagDeploymentStatus{}, nil).Times(5)
 
 	mockImageHandler := new(mocks.ImageHandler)
 	airflowImageHandler = func(image string) airflow.ImageHandler {
 		mockImageHandler.On("Build", mock.Anything).Return(nil)
 		mockImageHandler.On("Push", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(nil)
 		mockImageHandler.On("GetLabel", runtimeImageLabel).Return("", nil)
-		mockImageHandler.On("TagLocalImage", mock.Anything).Return(nil)
 		return mockImageHandler
 	}
 
@@ -224,14 +223,6 @@ func TestDeployWithDagsDeploySuccess(t *testing.T) {
 	deployInput.Pytest = "pytest"
 	deployInput.Prompt = false
 	err = Deploy(deployInput, mockClient)
-
-	assert.NoError(t, err)
-
-	// test custom image
-	defer testUtil.MockUserInput(t, "y")()
-	deployInput.ImageName = "custom-image"
-	err = Deploy(deployInput, mockClient)
-
 	assert.NoError(t, err)
 
 	config.CFG.ProjectDeployment.SetProjectString("test-id")
@@ -251,6 +242,12 @@ func TestDeployWithDagsDeploySuccess(t *testing.T) {
 	err = Deploy(deployInput, mockClient)
 	assert.NoError(t, err)
 
+	// test custom image with dag deploy enabled
+	defer testUtil.MockUserInput(t, "y")()
+	deployInput.ImageName = "custom-image"
+	err = Deploy(deployInput, mockClient)
+	assert.ErrorIs(t, err, customDeployError)
+
 	defer os.RemoveAll("./testfiles/dags/")
 
 	mockClient.AssertExpectations(t)