Add more Pulp Exceptions.

  Assisted-by: Claude Sonnet 4.5 <noreply@anthropic.com>

Author: aKlimau

CHANGES/+add-pulp-exceptions.bugfix

@@ -0,0 +1 @@
+Add more Pulp Exceptions.

pulp_python/app/exceptions.py

@@ -0,0 +1,43 @@
+from gettext import gettext as _
+
+from pulpcore.plugin.exceptions import PulpException
+
+
+class ProvenanceVerificationError(PulpException):
+    """
+    Raised when provenance verification fails.
+    """
+
+    error_code = "PLPY0001"
+
+    def __init__(self, message):
+        """
+        :param message: Description of the provenance verification error
+        :type message: str
+        """
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Provenance verification failed: {message}").format(
+            message=self.message
+        )
+
+
+class AttestationVerificationError(PulpException):
+    """
+    Raised when attestation verification fails.
+    """
+
+    error_code = "PLPY0002"
+
+    def __init__(self, message):
+        """
+        :param message: Description of the attestation verification error
+        :type message: str
+        """
+        self.message = message
+
+    def __str__(self):
+        return f"[{self.error_code}] " + _("Attestation verification failed: {message}").format(
+            message=self.message
+        )

pulp_python/app/serializers.py

@@ -8,14 +8,16 @@
 from packaging.requirements import Requirement
 from rest_framework import serializers
 from pypi_attestations import AttestationError
-from pydantic import TypeAdapter, ValidationError
+from pydantic import TypeAdapter, ValidationError as PydanticValidationError
 from urllib.parse import urljoin
 
+from pulpcore.plugin.exceptions import DigestValidationError, ValidationError
 from pulpcore.plugin import models as core_models
 from pulpcore.plugin import serializers as core_serializers
 from pulpcore.plugin.util import get_domain, get_prn, get_current_authenticated_user
 
 from pulp_python.app import models as python_models
+from pulp_python.app.exceptions import AttestationVerificationError, ProvenanceVerificationError
 from pulp_python.app.provenance import (
     Attestation,
     Provenance,
@@ -374,7 +376,7 @@ def validate_attestations(self, value):
                 attestations = TypeAdapter(list[Attestation]).validate_json(value)
             else:
                 attestations = TypeAdapter(list[Attestation]).validate_python(value)
-        except ValidationError as e:
+        except PydanticValidationError as e:
             raise serializers.ValidationError(_("Invalid attestations: {}".format(e)))
         return attestations
 
@@ -387,9 +389,7 @@ def handle_attestations(self, filename, sha256, attestations, offline=True):
         try:
             verify_provenance(filename, sha256, provenance, offline=offline)
         except AttestationError as e:
-            raise serializers.ValidationError(
-                {"attestations": _("Attestations failed verification: {}".format(e))}
-            )
+            raise AttestationVerificationError(str(e))
         return provenance.model_dump(mode="json")
 
     def deferred_validate(self, data):
@@ -408,26 +408,23 @@ def deferred_validate(self, data):
         try:
             filename = data["relative_path"]
         except KeyError:
-            raise serializers.ValidationError(detail={"relative_path": _("This field is required")})
+            raise ValidationError(_("This field is required: relative_path"))
 
         artifact = data["artifact"]
         try:
             _data = artifact_to_python_content_data(filename, artifact, domain=get_domain())
         except ValueError:
-            raise serializers.ValidationError(
+            raise ValidationError(
                 _(
                     "Extension on {} is not a valid python extension "
                     "(.whl, .exe, .egg, .tar.gz, .tar.bz2, .zip)"
                 ).format(filename)
             )
 
         if data.get("sha256") and data["sha256"] != artifact.sha256:
-            raise serializers.ValidationError(
-                detail={
-                    "sha256": _(
-                        "The uploaded artifact's sha256 checksum does not match the one provided"
-                    )
-                }
+            raise DigestValidationError(
+                actual=artifact.sha256,
+                expected=data["sha256"],
             )
 
         data.update(_data)
@@ -641,15 +638,13 @@ def deferred_validate(self, data):
         try:
             provenance = Provenance.model_validate_json(data["file"].read())
             data["provenance"] = provenance.model_dump(mode="json")
-        except ValidationError as e:
-            raise serializers.ValidationError(
-                _("The uploaded provenance is not valid: {}".format(e))
-            )
+        except PydanticValidationError as e:
+            raise ValidationError(_("The uploaded provenance is not valid: {}".format(e)))
         if data.pop("verify"):
             try:
                 verify_provenance(data["package"].filename, data["package"].sha256, provenance)
             except AttestationError as e:
-                raise serializers.ValidationError(_("Provenance verification failed: {}".format(e)))
+                raise ProvenanceVerificationError(str(e))
         return data
 
     def retrieve(self, validated_data):

pulp_python/tests/functional/api/test_attestations.py

@@ -69,7 +69,7 @@ def test_verify_provenance(python_bindings, twine_package, python_content_factor
     with pytest.raises(PulpTaskError) as e:
         monitor_task(provenance.task)
     assert e.value.task.state == "failed"
-    assert "twine-6.2.0-py3-none-any.whl != twine-6.2.0.tar.gz" in e.value.task.error["description"]
+    assert "[PLPY0002]" in e.value.task.error["description"]
 
     # Test creating a provenance without verifying
     provenance = python_bindings.ContentProvenanceApi.create(
@@ -239,4 +239,4 @@ def test_bad_attestation_upload(python_bindings, twine_package, monitor_task):
     with pytest.raises(PulpTaskError) as e:
         monitor_task(task)
     assert e.value.task.state == "failed"
-    assert "Attestations failed verification" in e.value.task.error["description"]
+    assert "[PLPY0002]" in e.value.task.error["description"]