feat: validators for cdd,ead, and error pipeline

eox_nelp/pearson_vue/data_classes.py

@@ -0,0 +1,71 @@
+"""
+Module to add data_classes related Pearson Vue Integration
+"""
+# pylint: disable=missing-class-docstring
+from pydantic import BaseModel, Field
+
+
+class Phone(BaseModel):
+    phone_number: str = Field(alias="phoneNumber", min_length=1, max_length=20)
+    phone_country_code: str = Field(alias="phoneCountryCode", min_length=1, max_length=3)
+
+
+class Mobile(BaseModel):
+    mobile_number: str = Field(alias="mobileNumber", min_length=1, max_length=20)
+    mobile_country_code: str = Field(alias="mobileCountryCode", min_length=1, max_length=3)
+
+
+class NativeAddress(BaseModel):
+    language: str = Field(alias="language", min_length=3, max_length=3)
+    potential_mismatch: str = Field(alias="potentialMismatch", min_length=1)
+    first_name: str = Field(alias="firstName", min_length=1, max_length=30)
+    last_name: str = Field(alias="lastName", min_length=1, max_length=50)
+    address1: str = Field(alias="address1", min_length=1, max_length=40)
+    city: str = Field(alias="city", min_length=1, max_length=32)
+
+
+class Address(BaseModel):
+    address1: str = Field(alias="address1", min_length=1, max_length=40)
+    city: str = Field(alias="city", min_length=1, max_length=32)
+    country: str = Field(alias="country", min_length=1, max_length=3)
+    phone: Phone = Field(alias="phone")
+    mobile: Mobile = Field(alias="mobile")
+    native_address: NativeAddress = Field(alias="nativeAddress")
+
+
+class PrimaryAddress(Address):
+    pass
+
+
+class AlternateAddress(Address):
+    pass
+
+
+class CandidateName(BaseModel):
+    first_name: str = Field(alias="firstName", min_length=1, max_length=30)
+    last_name: str = Field(alias="lastName", min_length=1, max_length=50)
+
+
+class WebAccountInfo(BaseModel):
+    email: str = Field(alias="email", min_length=1, max_length=255)
+
+
+class CddRequest(BaseModel):
+    client_candidate_id: str = Field(alias="@clientCandidateID", min_length=1, max_length=50)
+    client_id: str = Field(alias="@clientID", min_length=1)
+    candidate_name: CandidateName = Field(alias="candidateName")
+    last_update: str = Field(alias="lastUpdate", min_length=1)
+    primary_address: PrimaryAddress = Field(alias="primaryAddress")
+    web_account_info: WebAccountInfo = Field(alias="webAccountInfo")
+
+
+class EadRequest(BaseModel):
+    client_id: str = Field(alias="@clientID", min_length=1)
+    authorization_transaction_type: str = Field(alias="@authorizationTransactionType", min_length=1)
+    client_authorization_id: str = Field(alias="@clientAuthorizationID", min_length=1, max_length=25)
+    client_candidate_id: str = Field(alias="clientCandidateID", min_length=1, max_length=50)
+    exam_authorization_count: int = Field(alias="examAuthorizationCount")
+    exam_series_code: str = Field(alias="examSeriesCode", min_length=1, max_length=20)
+    elegibility_appt_date_first: str = Field(alias="eligibilityApptDateFirst", min_length=1)
+    elegibility_appt_date_last: str = Field(alias="eligibilityApptDateLast", min_length=1)
+    last_update: str = Field(alias="lastUpdate", min_length=1)

eox_nelp/pearson_vue/exceptions.py

@@ -0,0 +1,79 @@
+"""
+Module to add managed exceptions related Pearson Vue Integration
+"""
+import inspect
+
+
+class PearsonBaseError(Exception):
+    """Pearson Base error class
+    Most classes that inherit from this class must have exception_type.
+    """
+    exception_type = "base-error"
+
+    def __init__(self, *args, pipe_frame=inspect.currentframe(), exception_reason="", exception_dict=None):
+        """Init pearson exception using frame and exception_reason(str)
+        Or init using exception_dict representation.
+        That representation should have the following shape:
+         exception_dict  = {
+                'exception_type': 'validation-error',
+                'pipe_args_dict': {
+                    "cdd_request": {}
+                },
+                'pipe_function': 'validate_cdd_request',
+                'exception_reason': "error: ['String to short.']"
+            }
+        """
+
+        if exception_dict:
+            _ = [setattr(self, key, value) for key, value in exception_dict.items()]
+            super().__init__(exception_dict, *args)
+        else:
+            self.exception_type = self.exception_type
+            arg_info = inspect.getargvalues(pipe_frame)
+            self.pipe_args_dict = {arg: arg_info.locals[arg] for arg in arg_info.args}
+            self.pipe_function = pipe_frame.f_code.co_name
+            self.exception_reason = exception_reason
+            super().__init__(self.to_dict(), *args)
+
+    def to_dict(self):
+        """
+        Returns a dictionary representation of the class instance.
+
+        Returns:
+            A dictionary containing the instance's attributes as key-value pairs.
+        """
+
+        return {key: value for key, value in self.__dict__.items() if not key.startswith('_')}
+
+    @classmethod
+    def from_dict(cls, exception_dict: dict):
+        """Create an instance of Person or its subclass from a dictionary.
+            Returns:
+            Matched instance of pearson exception subclass initialized.
+            If not matched returns the base class initialized by default.
+        """
+        exception_type = exception_dict.get('exception_type')
+        for subclass in cls.__subclasses__():
+            if subclass.exception_type == exception_type:
+                return subclass(exception_dict=exception_dict)
+
+        # Default to Person if no matching subclass is found
+        return cls()
+
+
+class PearsonKeyError(PearsonBaseError):
+    """Pearson Key error class
+    """
+    exception_type = "key-error"
+
+
+class PearsonAttributeError(PearsonBaseError):
+    """Pearson Attribute error class
+    """
+    exception_type = "attribute-error"
+
+
+class PearsonValidationError(PearsonBaseError):
+    """Pearson Validation error class
+    """
+    exception_type = "validation-error"

eox_nelp/pearson_vue/pipeline.py

@@ -12,16 +12,25 @@
     import_exam_authorization: Imports exam authorization data.
     get_exam_data: Retrieves exam data.
 """
+import inspect
 import logging
 
 import phonenumbers
 from django.conf import settings
 from django.contrib.auth import get_user_model
 from django.utils import timezone
+from pydantic import ValidationError
 
 from eox_nelp.api_clients.pearson_rti import PearsonRTIApiClient
 from eox_nelp.edxapp_wrapper.student import anonymous_id_for_user
 from eox_nelp.pearson_vue.constants import PAYLOAD_CDD, PAYLOAD_EAD, PAYLOAD_PING_DATABASE
+from eox_nelp.pearson_vue.data_classes import CddRequest, EadRequest
+from eox_nelp.pearson_vue.exceptions import (
+    PearsonAttributeError,
+    PearsonBaseError,
+    PearsonKeyError,
+    PearsonValidationError,
+)
 from eox_nelp.pearson_vue.utils import generate_client_authorization_id, update_xml_with_dict
 from eox_nelp.signals.utils import get_completed_and_graded
 
@@ -32,6 +41,7 @@ def audit_method(action):  # pylint: disable=unused-argument
         """Identity audit_method"""
         return lambda x: x
 
+
 logger = logging.getLogger(__name__)
 
 User = get_user_model()
@@ -300,8 +310,13 @@ def get_exam_data(user_id, course_id, **kwargs):  # pylint: disable=unused-argum
     Raises:
         Exception: If the Pearson VUE RTI service does not accept the import request.
     """
-    courses_data = getattr(settings, "PEARSON_RTI_COURSES_DATA")
-    exam_metadata = courses_data[course_id]
+    try:
+        courses_data = getattr(settings, "PEARSON_RTI_COURSES_DATA")
+        exam_metadata = courses_data[course_id]
+    except KeyError as exc:
+        raise PearsonKeyError(pipe_frame=inspect.currentframe(), exception_reason=str(exc)) from exc
+    except AttributeError as a_exc:
+        raise PearsonAttributeError(pipe_frame=inspect.currentframe(), exception_reason=str(a_exc)) from a_exc
 
     # This generates the clientAuthorizationID based on the user_id and course_id
     exam_metadata["client_authorization_id"] = generate_client_authorization_id(
@@ -337,39 +352,45 @@ def build_cdd_request(profile_metadata, **kwargs):  # pylint: disable=unused-arg
     Returns:
        dict: dict with ead_request dict.
     """
-    cdd_request = {
-        "@clientCandidateID": f'NELC{profile_metadata["anonymous_user_id"]}',
-        "@clientID": getattr(settings, "PEARSON_RTI_WSDL_CLIENT_ID"),
-        "candidateName": {
-            "firstName": profile_metadata["first_name"],
-            "lastName": profile_metadata["last_name"],
-        },
-        "webAccountInfo": {
-            "email": profile_metadata["email"],
-        },
-        "lastUpdate": timezone.now().strftime("%Y/%m/%d %H:%M:%S GMT"),
-        "primaryAddress": {
-            "address1": profile_metadata["address"],
-            "city": profile_metadata["city"],
-            "country": profile_metadata["country"],
-            "phone": {
-                "phoneNumber": profile_metadata["phone_number"],
-                "phoneCountryCode": profile_metadata["phone_country_code"],
+    try:
+        cdd_request = {
+            "@clientCandidateID": f'NELC{profile_metadata["anonymous_user_id"]}',
+            "@clientID": getattr(settings, "PEARSON_RTI_WSDL_CLIENT_ID"),
+            "candidateName": {
+                "firstName": profile_metadata["first_name"],
+                "lastName": profile_metadata["last_name"],
             },
-            "mobile": {
-                "mobileNumber": profile_metadata["mobile_number"],
-                "mobileCountryCode": profile_metadata["mobile_country_code"],
+            "webAccountInfo": {
+                "email": profile_metadata["email"],
             },
-            "nativeAddress": {
-                "language": getattr(settings, "PEARSON_RTI_NATIVE_ADDRESS_LANGUAGE", "UKN"),
-                "potentialMismatch": "false",
-                "firstName": profile_metadata["arabic_name"],
-                "lastName": profile_metadata["arabic_name"],
+            "lastUpdate": timezone.now().strftime("%Y/%m/%d %H:%M:%S GMT"),
+            "primaryAddress": {
                 "address1": profile_metadata["address"],
                 "city": profile_metadata["city"],
-            },
+                "country": profile_metadata["country"],
+                "phone": {
+                    "phoneNumber": profile_metadata["phone_number"],
+                    "phoneCountryCode": profile_metadata["phone_country_code"],
+                },
+                "mobile": {
+                    "mobileNumber": profile_metadata["mobile_number"],
+                    "mobileCountryCode": profile_metadata["mobile_country_code"],
+                },
+                "nativeAddress": {
+                    "language": getattr(settings, "PEARSON_RTI_NATIVE_ADDRESS_LANGUAGE", "UKN"),
+                    "potentialMismatch": "false",
+                    "firstName": profile_metadata["arabic_name"],
+                    "lastName": profile_metadata["arabic_name"],
+                    "address1": profile_metadata["address"],
+                    "city": profile_metadata["city"],
+                },
+            }
         }
-    }
+    except KeyError as exc:
+        raise PearsonKeyError(pipe_frame=inspect.currentframe(), exception_reason=str(exc)) from exc
+    except AttributeError as a_exc:
+        raise PearsonAttributeError(pipe_frame=inspect.currentframe(), exception_reason=str(a_exc)) from a_exc
+
     return {
         "cdd_request": cdd_request
     }
@@ -392,18 +413,88 @@ def build_ead_request(
     Returns:
         dict: dict with ead_request dict
     """
-    ead_request = {
-        "@clientAuthorizationID": exam_metadata["client_authorization_id"],
-        "@clientID": getattr(settings, "PEARSON_RTI_WSDL_CLIENT_ID"),
-        "@authorizationTransactionType": transaction_type,
-        "clientCandidateID": f'NELC{profile_metadata["anonymous_user_id"]}',
-        "examAuthorizationCount": exam_metadata["exam_authorization_count"],
-        "examSeriesCode": exam_metadata["exam_series_code"],
-        "eligibilityApptDateFirst": exam_metadata["eligibility_appt_date_first"],
-        "eligibilityApptDateLast": exam_metadata["eligibility_appt_date_last"],
-        "lastUpdate": timezone.now().strftime("%Y/%m/%d %H:%M:%S GMT"),
-    }
+    try:
+        ead_request = {
+            "@clientAuthorizationID": exam_metadata["client_authorization_id"],
+            "@clientID": getattr(settings, "PEARSON_RTI_WSDL_CLIENT_ID"),
+            "@authorizationTransactionType": transaction_type,
+            "clientCandidateID": f'NELC{profile_metadata["anonymous_user_id"]}',
+            "examAuthorizationCount": exam_metadata["exam_authorization_count"],
+            "examSeriesCode": exam_metadata["exam_series_code"],
+            "eligibilityApptDateFirst": exam_metadata["eligibility_appt_date_first"],
+            "eligibilityApptDateLast": exam_metadata["eligibility_appt_date_last"],
+            "lastUpdate": timezone.now().strftime("%Y/%m/%d %H:%M:%S GMT"),
+        }
+    except KeyError as exc:
+        raise PearsonKeyError(pipe_frame=inspect.currentframe(), exception_reason=str(exc)) from exc
+    except AttributeError as a_exc:
+        raise PearsonAttributeError(pipe_frame=inspect.currentframe(), exception_reason=str(a_exc)) from a_exc
 
     return {
         "ead_request": ead_request
     }
+
+
+def validate_cdd_request(cdd_request, **kwargs):  # pylint: disable=unused-argument):
+    """
+    Validates a CDD request dictionary using a Pydantic model.
+
+    This function attempts to create a Pydantic model instance (likely named `class CddRequest`:
+`)
+    from the provided `cdd_request` dictionary. It performs data validation based on the
+    model's data type definitions.
+    Then if there is an error then that error is raised using audit. PearsonValidationError
+
+    Args:
+        cdd_request (dict): The dictionary containing the CDD request data.
+    """
+    try:
+        CddRequest(**cdd_request)
+    except ValidationError as exc:
+        raise PearsonValidationError(pipe_frame=inspect.currentframe(), exception_reason=str(exc)) from exc
+
+
+def validate_ead_request(ead_request, **kwargs):  # pylint: disable=unused-argument
+    """
+    Validates an EAD request dictionary using a Pydantic model.
+
+    This function attempts to create a Pydantic model instance (likely named `EadRequest`)
+    from the provided `ead_request` dictionary. It performs data validation based on the
+    model's data type definitions.
+    Then if there is an error then that error is raised using PearsonValidationError
+
+    Args:
+        ead_request (dict): The dictionary containing the EAD request data.
+    """
+    try:
+        EadRequest(**ead_request)
+    except ValidationError as exc:
+        raise PearsonValidationError(pipe_frame=inspect.currentframe(), exception_reason=str(exc)) from exc
+
+
+def audit_pearson_error(failed_step_pipeline="", exception_dict=None, **kwargs):  # pylint: disable=unused-argument
+    """
+    Method to save an error with eox-audit.
+    Args:
+        exception_dict(dict): dict presentation of the dict.
+        failed_step_pipeline(str): Name of function of step failed.
+        the audit model and the logger error.
+        **kwargs
+    Logs:
+        LogError: log pearsonsubclass error subclass matched.
+    Returns:
+        None
+    """
+    if not exception_dict or not failed_step_pipeline:
+        return
+
+    audit_action = f"Pearson Vue Exception~{exception_dict['exception_type']}"
+
+    @audit_method(action=audit_action)
+    def raise_audit_pearson_exception(exception_dict, *args, **kwargs):
+        raise PearsonBaseError.from_dict(exception_dict)
+
+    try:
+        raise_audit_pearson_exception(failed_step_pipeline=failed_step_pipeline, exception_dict=exception_dict)
+    except PearsonBaseError as exc:
+        logger.error(exc)