Repository: SFDO-Tooling/CumulusCI
Pull request #3598 (2023-05-25 22:09:50)

Use Marketing Cloud's validate endpoint to update deployment packages


Patch:
diff --git a/cumulusci/tasks/marketing_cloud/deploy.py b/cumulusci/tasks/marketing_cloud/deploy.py
index df68d3bff..840fe942b 100644
--- a/cumulusci/tasks/marketing_cloud/deploy.py
+++ b/cumulusci/tasks/marketing_cloud/deploy.py
@@ -2,7 +2,9 @@ import json
 import uuid
 import zipfile
 from collections import defaultdict
+from enum import Enum
 from pathlib import Path
+from typing import Dict, Optional
 
 import requests
 
@@ -14,7 +16,9 @@ from cumulusci.utils.http.requests_utils import safe_json_from_response
 
 from .base import BaseMarketingCloudTask
 
-MCPM_ENDPOINT = "https://spf.{}.marketingcloudapps.com/api"
+MCPM_BASE_ENDPOINT = "https://spf.{}.marketingcloudapps.com/api"
+
+MCPM_JOB_ID_HEADER = "x-mcpm-job-id"
 
 PAYLOAD_CONFIG_VALUES = {"preserveCategories": True}
 
@@ -25,14 +29,25 @@ PAYLOAD_NAMESPACE_VALUES = {
     "timestamp": True,
 }
 
-IN_PROGRESS_STATUS = "IN_PROGRESS"
+IN_PROGRESS_STATUSES = ("NOT_STARTED", "IN_PROGRESS")
 FINISHED_STATUSES = ("DONE",)
 ERROR_STATUSES = ("FATAL_ERROR", "ERROR")
 
+
 UNKNOWN_STATUS_MESSAGE = "Received unknown deploy status: {}"
 
 
+class PollAction(Enum):
+    validating = "VALIDATING"
+    deploying = "DEPLOYING"
+
+
 class MarketingCloudDeployTask(BaseMarketingCloudTask):
+    # This task executes multiple polling loops.
+    # This enables the task to determine which endpoints should be polled.
+    current_action: Optional[PollAction] = None
+    validation_not_found_count = 0
+    job_name: Optional[str]
 
     task_options = {
         "package_zip_file": {
@@ -62,6 +77,9 @@ class MarketingCloudDeployTask(BaseMarketingCloudTask):
         )
 
     def _run_task(self):
+        # Marketing Cloud validation _requires_ a 2-second static polling interval
+        self.poll_interval_s = 2
+
         pkg_zip_file = Path(self.options["package_zip_file"])
         if not pkg_zip_file.is_file():
             self.logger.error(f"Package zip file not valid: {pkg_zip_file.name}")
@@ -72,60 +90,26 @@ class MarketingCloudDeployTask(BaseMarketingCloudTask):
                 zf.extractall(temp_dir)
                 payload = self._construct_payload(Path(temp_dir), self.custom_inputs)
 
+        # These initialization steps are not done in _init_options()
+        # because they use MC authorization. We don't want to freeze
+        # the responses.
+        endpoint_option = self.options.get("endpoint")
+        self.endpoint = endpoint_option or MCPM_BASE_ENDPOINT.format(
+            self.get_mc_stack_key()
+        )
+
         self.headers = {
             "Authorization": f"Bearer {self.mc_config.access_token}",
             "SFMC-TSSD": self.mc_config.tssd,
         }
-        custom_endpoint = self.options.get("endpoint")
-        self.endpoint = (
-            custom_endpoint
-            if custom_endpoint
-            else MCPM_ENDPOINT.format(self.get_mc_stack_key())
-        )
 
-        self.logger.info(f"Deploying package to: {self.endpoint}/deployments")
-        response = requests.post(
-            f"{self.endpoint}/deployments",
-            json=payload,
-            headers=self.headers,
-        )
-        response_data = safe_json_from_response(response)
+        self._validate_package(payload)
+        self._reset_poll()
 
-        self.job_id = response_data["id"]
-        self.logger.info(f"Started deploy job with Id: {self.job_id}")
-        self._poll()
-
-    def _poll_action(self):
-        """
-        Poll something and process the response.
-        Set `self.poll_complete = True` to break polling loop.
-        """
-        response = requests.get(
-            f"{self.endpoint}/deployments/{self.job_id}", headers=self.headers
+        payload = self._update_payload_entities_with_actions(
+            payload,
         )
-        response_data = safe_json_from_response(response)
-        deploy_status = response_data["status"]
-        self.logger.info(f"Deployment status is: {deploy_status}")
-
-        if deploy_status != IN_PROGRESS_STATUS:
-            self._process_completed_deploy(response_data)
-
-    def _process_completed_deploy(self, response_data: dict):
-        deploy_status = response_data["status"]
-        assert (
-            deploy_status != IN_PROGRESS_STATUS
-        ), "Deploy should be in a completed state before processing."
-
-        if deploy_status in FINISHED_STATUSES:
-            self.poll_complete = True
-            self._validate_response(response_data)
-        elif deploy_status in ERROR_STATUSES:
-            self.poll_complete = True
-            self._report_error(response_data)
-        else:
-            self.logger.error(UNKNOWN_STATUS_MESSAGE.format(deploy_status))
-            self.poll_complete = True
-            self._report_error(response_data)
+        self._deploy_package(payload)
 
     def _construct_payload(self, dir_path, custom_inputs=None):
         dir_path = Path(dir_path)
@@ -134,7 +118,7 @@ class MarketingCloudDeployTask(BaseMarketingCloudTask):
         payload = defaultdict(lambda: defaultdict(dict))
         payload["namespace"] = PAYLOAD_NAMESPACE_VALUES
         payload["config"] = PAYLOAD_CONFIG_VALUES
-        payload["name"] = self.options.get("name", str(uuid.uuid4()))
+        payload["name"] = self.job_name = self.options.get("name", str(uuid.uuid4()))  # type: ignore
 
         try:
             with open(dir_path / "info.json", "r") as f:
@@ -187,9 +171,153 @@ class MarketingCloudDeployTask(BaseMarketingCloudTask):
 
         return payload
 
-    def _validate_response(self, deploy_info: dict):
+    def _validate_package(self, payload: Dict) -> None:
+        """Sends the payload to MC for validation.
+        Returns a dict of allowable actions for the target MC instance."""
+        self.current_action = PollAction.validating
+        self.logger.info(f"Validating package at: {self.endpoint}/validate")
+        assert self.job_name
+        response = requests.post(
+            f"{self.endpoint}/validate",
+            json=payload,
+            headers={
+                MCPM_JOB_ID_HEADER: self.job_name,
+                **self.headers,
+            },
+        )
+        response.raise_for_status()
+        response_data = safe_json_from_response(response)
+        self.validate_id = response_data["id"]
+        self.logger.info(f"Started package validation with Id: {self.validate_id}")
+        self._poll()
+
+    def _update_payload_entities_with_actions(self, payload: Dict) -> Dict:
+        """Include available entity action returned from the validation
+        endpoint into the payload used for package deployment."""
+
+        for entity_type in payload["entities"]:
+            this_entity_type = payload["entities"][entity_type]
+            for entity_id in this_entity_type:
+                this_entity = payload["entities"][entity_type][entity_id]
+                action = self.action_for_entity(entity_type, entity_id)
+                this_entity["action"] = action
+
+        if self.debug_mode:
+            self.logger.debug(f"Payload updated with actions:\n{json.dumps(payload)}")
+
+        return payload
+
+    def action_for_entity(self, entity: str, entity_id: str) -> Optional[Dict]:
+        """Fetch the corresponding action for the given entity with the specified Id"""
+        try:
+            for action_info in self.validation_response["entities"][entity][entity_id][
+                "actions"
+            ].values():
+                if action_info["available"]:
+                    return action_info
+        except KeyError:
+            # if no actions are defined for this entity just move on
+            pass
+
+    def _deploy_package(self, payload: Dict) -> None:
+        self.current_action = PollAction.deploying
+        self.logger.info(f"Deploying package to: {self.endpoint}/deployments")
+        response = requests.post(
+            f"{self.endpoint}/deployments",
+            json=payload,
+            headers=self.headers,
+        )
+        response.raise_for_status()
+        response_data = safe_json_from_response(response)
+        self.deploy_id = response_data["id"]
+        self.logger.info(f"Started package deploy with Id: {self.validate_id}")
+        self._poll()
+
+    def _poll_action(self) -> None:
+        """
+        Poll based on the current action being taken.
+        """
+        if self.current_action == PollAction.validating:
+            self._poll_validating()
+        elif self.current_action == PollAction.deploying:
+            self._poll_deploying()
+        else:
+            raise Exception(
+                f"PollAction {self.current_action} does not have a polling handler defined."
+            )
+
+    def _poll_validating(self) -> None:
+        assert self.job_name
+        response = requests.get(
+            f"{self.endpoint}/validate/{self.validate_id}",
+            headers={MCPM_JOB_ID_HEADER: self.job_name, **self.headers},
+        )
+        response_data = safe_json_from_response(response)
+        validation_status = response_data["status"]
+        self.logger.info(f"Validation status is: {validation_status}")
+
+        # Handle eccentricities of Marketing Cloud validation polling.
+        # We may get back a NOT_FOUND result if our request is routed
+        # to the wrong pod by MC. This should be fixed, but hasn't been
+        # deployed to all MC stacks yet.
+
+        # Observed behavior is that response switches between NOT_FOUND
+        # and IN_PROGRESS. We'll track these responses and fail the job
+        # only if we consistently receive NOT_FOUND.
+
+        # TODO: add a timeout.
+        if validation_status == "NOT_FOUND":
+            self.validation_not_found_count += 1
+            if self.validation_not_found_count > 10:
+                raise DeploymentException(
+                    f"Unable to find status on validation: {self.validate_id}"
+                )
+        else:
+            # Reset if we get back a result other than NOT_FOUND
+            self.validation_not_found_count = 0
+            if validation_status not in IN_PROGRESS_STATUSES:
+                self.poll_complete = True
+                if self.debug_mode:  # pragma: nocover
+                    self.logger.debug(
+                        f"Validation Response:\n{json.dumps(response_data)}"
+                    )
+                self.validation_response = response_data
+
+    def _poll_deploying(self) -> None:
+        response = requests.get(
+            f"{self.endpoint}/deployments/{self.deploy_id}", headers=self.headers
+        )
+        response_data = safe_json_from_response(response)
+        deploy_status = response_data["status"]
+        self.logger.info(f"Deployment status is: {deploy_status}")
+
+        if deploy_status not in IN_PROGRESS_STATUSES:
+            self._process_completed_deploy(response_data)
+
+    def _poll_update_interval(self):
+        # Marketing Cloud validation _requires_ a 2-second static polling interval.
+        # Override the base class to remove backoff.
+        pass
+
+    def _process_completed_deploy(self, response_data: Dict):
+        deploy_status = response_data["status"]
+        assert (
+            deploy_status != IN_PROGRESS_STATUSES
+        ), "Deploy should be in a completed state before processing."
+
+        self.poll_complete = True
+        if deploy_status in FINISHED_STATUSES:
+            self._validate_response(response_data)
+        elif deploy_status in ERROR_STATUSES:
+            self._report_error(response_data)
+        else:
+            self.logger.error(UNKNOWN_STATUS_MESSAGE.format(deploy_status))
+            self._report_error(response_data)
+
+    def _validate_response(self, deploy_info: Dict) -> None:
         """Checks for any errors present in the response to the deploy request.
-        Displays errors if present, else informs use that the deployment was successful."""
+        Displays errors if present, else informs use that the deployment was successful.
+        """
         has_error = False
         for entity, info in deploy_info["entities"].items():
             if not info:
@@ -205,12 +333,12 @@ class MarketingCloudDeployTask(BaseMarketingCloudTask):
         if has_error:
             raise DeploymentException("Marketing Cloud reported deployment failures.")
 
-        self.logger.info("Deployment completed successfully.")
+        self.logger.info(f"Deployment ({self.deploy_id}) completed successfully.")
 
-    def _report_error(self, response_data: dict):
+    def _report_error(self, response_data: Dict) -> None:
         deploy_status = response_data["status"]
         self.logger.error(
-            f"Received status of: {deploy_status}. Received the following data from Marketing Cloud:\n{response_data}^"
+            f"Received status of: {deploy_status}. Received the following data from Marketing Cloud:\n{json.dumps(response_data)}"
         )
         raise DeploymentException(
             f"Marketing Cloud deploy finished with status of: {deploy_status}"


Test patch:
diff --git a/cumulusci/tasks/marketing_cloud/tests/test_deploy.py b/cumulusci/tasks/marketing_cloud/tests/test_deploy.py
index c018b99e1..1a915dbe8 100644
--- a/cumulusci/tasks/marketing_cloud/tests/test_deploy.py
+++ b/cumulusci/tasks/marketing_cloud/tests/test_deploy.py
@@ -13,7 +13,8 @@ from cumulusci.core.config.marketing_cloud_service_config import (
 )
 from cumulusci.core.exceptions import DeploymentException
 from cumulusci.tasks.marketing_cloud.deploy import (
-    MCPM_ENDPOINT,
+    MCPM_BASE_ENDPOINT,
+    MCPM_JOB_ID_HEADER,
     UNKNOWN_STATUS_MESSAGE,
     MarketingCloudDeployTask,
 )
@@ -24,6 +25,12 @@ TEST_TSSD = "asdf-qwerty"
 STACK_KEY = "S4"
 TEST_ZIPFILE = "test-mc-pkg.zip"
 PAYLOAD_FILE = "expected-payload.json"
+JOB_NAME = "foo_job"
+
+# This JSON does not necessarily include the complete data provided by Marketing Cloud
+# It is intended to exercise functionality in the class under test.
+# The response is designed to go with the content of `expected-payload.json`
+VALIDATION_RESPONSE_FILE = "validation-response.json"
 
 
 @pytest.fixture
@@ -36,6 +43,7 @@ def task(project_config):
                 "options": {
                     "package_zip_file": test_zip_file.resolve(),
                     "custom_inputs": "companyName:Acme",
+                    "name": JOB_NAME,
                 }
             }
         ),
@@ -48,9 +56,23 @@ def task(project_config):
         "mc",
         None,
     )
+    task.poll_interval_s = 0  # Do not poll during testing
     return task
 
 
+@pytest.fixture
+def validation_response():
+    with open(Path(__file__).parent.absolute() / VALIDATION_RESPONSE_FILE) as f:
+        return json.load(f)
+
+
+@pytest.fixture
+def expected_payload():
+    expected_payload_file = Path(__file__).parent.absolute() / PAYLOAD_FILE
+    with open(expected_payload_file, "r") as f:
+        return json.load(f)
+
+
 @pytest.fixture
 def task_without_custom_inputs(project_config):
     test_zip_file = Path(__file__).parent.absolute() / TEST_ZIPFILE
@@ -61,6 +83,7 @@ def task_without_custom_inputs(project_config):
                 "options": {
                     "package_zip_file": test_zip_file.resolve(),
                     "custom_inputs": None,
+                    "name": JOB_NAME,
                 }
             }
         ),
@@ -87,54 +110,133 @@ def mocked_responses():
         yield rsps
 
 
-class TestMarketingCloudDeployTask:
-    def test_run_task__deploy_succeeds_with_custom_inputs(self, task, mocked_responses):
+@pytest.fixture
+def mocked_validation_responses(mocked_responses):
+    mocked_responses.add(
+        "POST",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate",
+        json={"id": "0be865fe-efb2-479d-99c2-c1b608155369"},
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+    )
+    mocked_responses.add(
+        "GET",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate/0be865fe-efb2-479d-99c2-c1b608155369",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json={"status": "DONE"},
+    )
+
+    yield mocked_responses
+
+
+@pytest.fixture
+def mocked_polling_validation_responses(mocked_responses, validation_response):
+    mocked_responses.add(
+        "POST",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json={"id": "0be865fe-efb2-479d-99c2-c1b608155369"},
+    )
+    mocked_responses.add(
+        "GET",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate/0be865fe-efb2-479d-99c2-c1b608155369",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json={"status": "NOT_FOUND"},
+    )
+    mocked_responses.add(
+        "GET",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate/0be865fe-efb2-479d-99c2-c1b608155369",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json={"status": "IN_PROGRESS"},
+    )
+    mocked_responses.add(
+        "GET",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate/0be865fe-efb2-479d-99c2-c1b608155369",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json={"status": "NOT_FOUND"},
+    )
+    mocked_responses.add(
+        "GET",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate/0be865fe-efb2-479d-99c2-c1b608155369",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json=validation_response,
+    )
+
+    yield mocked_responses
+
+
+@pytest.fixture
+def mocked_polling_validation_responses_all_not_found(mocked_responses):
+    mocked_responses.add(
+        "POST",
+        f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate",
+        match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+        json={"id": "0be865fe-efb2-479d-99c2-c1b608155369"},
+    )
+    for _ in range(10):
         mocked_responses.add(
+            "GET",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/validate/0be865fe-efb2-479d-99c2-c1b608155369",
+            match=[responses.matchers.header_matcher({MCPM_JOB_ID_HEADER: JOB_NAME})],
+            json={"status": "NOT_FOUND"},
+        )
+
+    yield mocked_responses
+
+
+class TestMarketingCloudDeployTask:
+    def test_run_task__deploy_succeeds_with_custom_inputs(
+        self, task, mocked_validation_responses
+    ):
+        mocked_validation_responses.add(
             "POST",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments",
             json={"id": "JOBID", "status": "IN_PROGRESS"},
         )
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "GET",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
             json={"status": "DONE", "entities": {}},
         )
 
         task.logger = mock.Mock()
         task._run_task()
-        task.logger.info.assert_called_with("Deployment completed successfully.")
+        task.logger.info.assert_called_with(
+            "Deployment (JOBID) completed successfully."
+        )
         assert task.logger.error.call_count == 0
         assert task.logger.warn.call_count == 0
 
     def test_run_task__deploy_succeeds_without_custom_inputs(
-        self, task_without_custom_inputs, mocked_responses
+        self, task_without_custom_inputs, mocked_validation_responses
     ):
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "POST",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments",
             json={"id": "JOBID", "status": "IN_PROGRESS"},
         )
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "GET",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
             json={"status": "DONE", "entities": {}},
         )
         task = task_without_custom_inputs
         task.logger = mock.Mock()
         task._run_task()
-        task.logger.info.assert_called_with("Deployment completed successfully.")
+        task.logger.info.assert_called_with(
+            "Deployment (JOBID) completed successfully."
+        )
         assert task.logger.error.call_count == 0
         assert task.logger.warn.call_count == 0
 
-    def test_run_task__deploy_fails(self, task, mocked_responses):
-        mocked_responses.add(
+    def test_run_task__deploy_fails(self, task, mocked_validation_responses):
+        mocked_validation_responses.add(
             "POST",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments",
             json={"id": "JOBID", "status": "IN_PROGRESS"},
         )
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "GET",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
             json={
                 "status": "DONE",
                 "entities": {
@@ -160,15 +262,15 @@ class TestMarketingCloudDeployTask:
             == "Failed to deploy assets/1. Status: SKIPPED. Issues: ['A problem occurred']"
         )
 
-    def test_run_task__FATAL_ERROR_result(self, task, mocked_responses):
-        mocked_responses.add(
+    def test_run_task__FATAL_ERROR_result(self, task, mocked_validation_responses):
+        mocked_validation_responses.add(
             "POST",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments",
             json={"id": "JOBID", "status": "FATAL_ERROR"},
         )
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "GET",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
             json={
                 "status": "FATAL_ERROR",
                 "entities": {},
@@ -178,16 +280,18 @@ class TestMarketingCloudDeployTask:
         with pytest.raises(DeploymentException):
             task._run_task()
 
-    def test_run_task__unknown_deploy_status(self, task, mocked_responses, caplog):
+    def test_run_task__unknown_deploy_status(
+        self, task, mocked_validation_responses, caplog
+    ):
         unknown_status = "FOOBAR"
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "POST",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments",
             json={"id": "JOBID"},
         )
-        mocked_responses.add(
+        mocked_validation_responses.add(
             "GET",
-            f"{MCPM_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
             json={
                 "status": unknown_status,
                 "entities": {},
@@ -200,6 +304,54 @@ class TestMarketingCloudDeployTask:
         logged_messages = [log.message for log in caplog.records]
         assert UNKNOWN_STATUS_MESSAGE.format(unknown_status) in logged_messages
 
+    # NOTE: the x-mcpm-job-id header is validated implicitly by all use
+    # of Responses mocks.
+
+    def test_validate_package__validation_accepts_intermittent_not_found(
+        self, task, mocked_polling_validation_responses
+    ):
+        mocked_polling_validation_responses.add(
+            "POST",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments",
+            json={"id": "JOBID", "status": "IN_PROGRESS"},
+        )
+        mocked_polling_validation_responses.add(
+            "GET",
+            f"{MCPM_BASE_ENDPOINT.format(STACK_KEY)}/deployments/JOBID",
+            json={"status": "DONE", "entities": {}},
+        )
+        task._run_task()
+        assert task.validation_response["status"] == "DONE"
+
+    def test_validate_package__validation_fails_consistent_not_found(
+        self, task, mocked_polling_validation_responses_all_not_found
+    ):
+        with pytest.raises(DeploymentException):
+            task._run_task()
+
+    def test_update_payload_entities_with_actions(
+        self, task, expected_payload, validation_response
+    ):
+        task.validation_response = validation_response
+        validated_payload = task._update_payload_entities_with_actions(expected_payload)
+
+        assert validated_payload["entities"]["assets"]["0"]["action"] == {
+            "type": "create",
+            "available": True,
+            "issues": [],
+        }
+
+    def test_action_for_entity(self, task, validation_response):
+        task.validation_response = validation_response
+
+        # We should expect to get back the first action with available == True
+        assert task.action_for_entity("assets", "0") == {
+            "type": "create",
+            "available": True,
+            "issues": [],
+        }
+        assert task.action_for_entity("foo", "0") is None
+
     def test_zipfile_not_valid(self, task):
         task.options["package_zip_file"] = "not-a-valid-file.zip"
         task.logger = mock.Mock()
@@ -209,25 +361,19 @@ class TestMarketingCloudDeployTask:
             "Package zip file not valid: not-a-valid-file.zip"
         )
 
-    @mock.patch("cumulusci.tasks.marketing_cloud.deploy.uuid")
-    def test_construct_payload(self, uuid, task):
-        uuid.uuid4.return_value = "cci-deploy"
+    def test_construct_payload(self, task, expected_payload):
+        task.options["name"] = "cci-deploy"
         pkg_zip_file = Path(task.options["package_zip_file"])
         with temporary_dir() as temp_dir:
             with zipfile.ZipFile(pkg_zip_file) as zf:
                 zf.extractall(temp_dir)
                 actual_payload = task._construct_payload(Path(temp_dir))
 
-        expected_payload_file = Path(__file__).parent.absolute() / PAYLOAD_FILE
-        with open(expected_payload_file, "r") as f:
-            expected_payload = json.load(f)
-
         assert expected_payload == actual_payload
 
-    @mock.patch("cumulusci.tasks.marketing_cloud.deploy.uuid")
-    def test_construct_payload__file_not_found(self, uuid, task):
+    def test_construct_payload__file_not_found(self, task):
         """Ensure we state clearly state where expect files to be"""
-        uuid.uuid4.return_value = "cci-deploy"
+        task.options["name"] = "cci-deploy"
         pkg_zip_file = Path(task.options["package_zip_file"])
         with temporary_dir() as temp_dir:
             with zipfile.ZipFile(pkg_zip_file) as zf:
diff --git a/cumulusci/tasks/marketing_cloud/tests/validation-response.json b/cumulusci/tasks/marketing_cloud/tests/validation-response.json
new file mode 100644
index 000000000..ee14613b4
--- /dev/null
+++ b/cumulusci/tasks/marketing_cloud/tests/validation-response.json
@@ -0,0 +1,39 @@
+{
+    "status": "DONE",
+    "entities": {
+        "assets": {
+            "0": {
+                "issues": [],
+                "data": {},
+                "action": {
+                    "type": "create",
+                    "available": true,
+                    "issues": []
+                },
+                "actions": {
+                    "update": {
+                        "type": "update",
+                        "available": false,
+                        "issues": []
+                    },
+                    "create": {
+                        "type": "create",
+                        "available": true,
+                        "issues": []
+                    },
+                    "reuse": {
+                        "type": "reuse",
+                        "available": false,
+                        "issues": []
+                    },
+                    "skip": {
+                        "type": "skip",
+                        "available": false
+                    }
+                },
+                "renameable": true,
+                "status": "READY"
+            }
+        }
+    }
+}