Do not retry on certification errors

cloudpub/error.py

@@ -27,3 +27,11 @@ class ConflictError(RuntimeError):
 
 class Timeout(Exception):
     """Represent a missing resource."""
+
+
+class CertificationError(InvalidStateError):
+    """Report Azure Marketplace certification failure."""
+
+
+class InvalidSchema(RuntimeError):
+    """Report when an invalid schema is returned from cloud provider API."""

cloudpub/models/ms_azure.py

@@ -67,7 +67,7 @@ class ConfigureStatus(AttrsJSONDecodeMixin):
     resource_uri: Optional[str] = field(metadata={"alias": "resourceUri", "hide_unset": True})
     """The resource URI related to the configure job."""
 
-    errors: List[str]
+    errors: List[Dict[str, Any]]
     """List of errors when the ``job_result`` is ``failed``."""
 
 

cloudpub/ms_azure/service.py

@@ -8,12 +8,18 @@
 from deepdiff import DeepDiff
 from requests import HTTPError
 from tenacity import RetryError, Retrying, retry
-from tenacity.retry import retry_if_exception_type, retry_if_result
+from tenacity.retry import retry_if_exception_type, retry_if_not_exception_type, retry_if_result
 from tenacity.stop import stop_after_attempt, stop_after_delay
 from tenacity.wait import wait_fixed
 
 from cloudpub.common import BaseService
-from cloudpub.error import ConflictError, InvalidStateError, NotFoundError, Timeout
+from cloudpub.error import (
+    CertificationError,
+    ConflictError,
+    InvalidStateError,
+    NotFoundError,
+    Timeout,
+)
 from cloudpub.models.ms_azure import (
     RESOURCE_MAPING,
     AzureResource,
@@ -43,6 +49,7 @@
     TechnicalConfigLookUpData,
     create_disk_version_from_scratch,
     is_azure_job_not_complete,
+    is_certification_error,
     is_sas_present,
     logdiff,
     seek_disk_version,
@@ -190,11 +197,14 @@ def query_job_status(self, job_id: str) -> ConfigureStatus:
         Returns:
             ConfigureStatus: The ConfigureStatus from JobID
         Raises:
+            CertificationError: If the job failed due to Certifications issues.
             InvalidStateError: If the job has failed.
         """
         job_details = self._query_job_details(job_id=job_id)
         if job_details.job_result == "failed":
             error_message = f"Job {job_id} failed: \n{job_details.errors}"
+            if is_certification_error(job_details.errors):
+                self._raise_error(CertificationError, error_message)
             self._raise_error(InvalidStateError, error_message)
         elif job_details.job_result == "succeeded":
             log.debug("Job %s succeeded", job_id)
@@ -731,6 +741,7 @@ def _is_submission_in_preview(self, current: ProductSubmission) -> bool:
     @retry(
         wait=wait_fixed(wait=60),
         stop=stop_after_attempt(3),
+        retry=retry_if_not_exception_type(CertificationError),
         reraise=True,
     )
     def _publish_preview(
@@ -754,16 +765,18 @@ def _publish_preview(
         if res.job_result != 'succeeded' or not self.get_submission_state(
             product.id, state="preview"
         ):
-            errors = "\n".join(res.errors)
             failure_msg = (
                 f"Failed to submit the product {product_name} ({product.id}) to preview. "
-                f"Status: {res.job_result} Errors: {errors}"
+                f"Status: {res.job_result} Errors: {res.errors}"
             )
+            if is_certification_error(res.errors):
+                self._raise_error(CertificationError, failure_msg)
             raise RuntimeError(failure_msg)
 
     @retry(
         wait=wait_fixed(wait=60),
         stop=stop_after_attempt(3),
+        retry=retry_if_not_exception_type(CertificationError),
         reraise=True,
     )
     def _publish_live(self, product: Product, product_name: str) -> None:
@@ -781,11 +794,12 @@ def _publish_live(self, product: Product, product_name: str) -> None:
         res = self.submit_to_status(product_id=product.id, status='live')
 
         if res.job_result != 'succeeded' or not self.get_submission_state(product.id, state="live"):
-            errors = "\n".join(res.errors)
             failure_msg = (
                 f"Failed to submit the product {product_name} ({product.id}) to live. "
-                f"Status: {res.job_result} Errors: {errors}"
+                f"Status: {res.job_result} Errors: {res.errors}"
             )
+            if is_certification_error(res.errors):
+                self._raise_error(CertificationError, failure_msg)
             raise RuntimeError(failure_msg)
 
     def _overwrite_disk_version(

cloudpub/ms_azure/utils.py

@@ -6,6 +6,7 @@
 from deepdiff import DeepDiff
 
 from cloudpub.common import PublishingMetadata  # Cannot circular import AzurePublishingMetadata
+from cloudpub.error import InvalidSchema
 from cloudpub.models.ms_azure import (
     ConfigureStatus,
     DiskVersion,
@@ -544,3 +545,29 @@ def logdiff(diff: DeepDiff) -> None:
     """Log the offer diff if it exists."""
     if diff:
         log.warning("Found the following offer diff before publishing:\n%s", diff.pretty())
+
+
+def _contains_certification_error(item: Any) -> bool:
+    """Recursively inspect Azure error payloads for certification failures."""
+    if not isinstance(item, dict):
+        raise InvalidSchema(f"Invalid schema for error object: {item}")
+
+    code: str = item.get("code", "")
+    message: str = item.get("message", "")
+    if code == "invalidState" and "certification" in message.lower():
+        return True
+    if not isinstance(item.get('details'), list):
+        raise InvalidSchema(f"Invalid schema for 'details' inside error object: {item}")
+    for detail in item.get("details") or []:
+        if _contains_certification_error(detail):
+            return True
+    return False
+
+
+def is_certification_error(errors: List[Dict[str, Any]]) -> bool:
+    """Return True when Azure job errors indicate a certification failure.
+
+    Certification failures are permanent for a given submission and should not
+    be retried.
+    """
+    return any(_contains_certification_error(error) for error in errors)

tests/ms_azure/conftest.py

@@ -98,7 +98,20 @@ def errors() -> List[Dict[str, Any]]:
         {
             "code": "conflict",
             "message": "Error message",
-            "details": [{"code": "invalidResource", "message": "Failure for resource"}],
+            "details": [
+                {"code": "invalidResource", "message": "Failure for resource", "details": []}
+            ],
+        }
+    ]
+
+
+@pytest.fixture
+def cert_error_invalid_schema() -> List[Dict[str, Any]]:
+    return [
+        {
+            "code": "internalServerError",
+            "message": "Certification failed.",
+            "details": {},
         }
     ]
 
@@ -629,3 +642,26 @@ def job_details_completed_failure_obj(
     job_details_completed_failure: Dict[str, Any],
 ) -> ConfigureStatus:
     return ConfigureStatus.from_json(job_details_completed_failure)
+
+
+@pytest.fixture
+def cert_error_failure() -> List[Dict[str, Any]]:
+    return [
+        {
+            "resourceId": "submission/2d541119",
+            "code": "internalServerError",
+            "message": "Operation failed",
+            "details": [
+                {
+                    "code": "invalidState",
+                    "message": "Certification",
+                    "details": [
+                        {
+                            "code": "invalidState",
+                            "message": "Issues found during Certification.",
+                        }
+                    ],
+                }
+            ],
+        }
+    ]