VED-79: refactor authentication, test and pipeline run

Author: Akol125

azure/templates/post-deploy.yml

@@ -77,6 +77,10 @@ steps:
     displayName: Apply Terraform
     workingDirectory: "$(Pipeline.Workspace)/s/$(SERVICE_NAME)/$(SERVICE_ARTIFACT_NAME)"
     retryCountOnTaskFailure: 2
+  - bash: |
+      echo "Creating MNS subscription..."
+      python3 mns_subscription/src/subscribe_mns.py
+      displayName: "Subscribe SQS to MNS signal"
 
   - bash: |
       set -ex

mns_subscription/models/errors.py

@@ -25,20 +25,7 @@ class Code(str, Enum):
 class UnauthorizedError(RuntimeError):
     @staticmethod
     def to_operation_outcome() -> dict:
-        msg = f"Unauthorized request"
-        return create_operation_outcome(
-            resource_id=str(uuid.uuid4()),
-            severity=Severity.error,
-            code=Code.forbidden,
-            diagnostics=msg,
-        )
-
-
-@dataclass
-class UnauthorizedVaxError(RuntimeError):
-    @staticmethod
-    def to_operation_outcome() -> dict:
-        msg = "Unauthorized request for vaccine type"
+        msg = "Unauthorized request"
         return create_operation_outcome(
             resource_id=str(uuid.uuid4()),
             severity=Severity.error,
@@ -60,25 +47,6 @@ def to_operation_outcome() -> dict:
         )
 
 
-@dataclass
-class ResourceNotFoundError(RuntimeError):
-    """Return this error when the requested FHIR resource does not exist"""
-
-    resource_type: str
-    resource_id: str
-
-    def __str__(self):
-        return f"{self.resource_type} resource does not exist. ID: {self.resource_id}"
-
-    def to_operation_outcome(self) -> dict:
-        return create_operation_outcome(
-            resource_id=str(uuid.uuid4()),
-            severity=Severity.error,
-            code=Code.not_found,
-            diagnostics=self.__str__(),
-        )
-
-
 @dataclass
 class ResourceFoundError(RuntimeError):
     """Return this error when the requested FHIR resource does exist"""
@@ -117,71 +85,6 @@ def to_operation_outcome(self) -> dict:
         )
 
 
-class MandatoryError(Exception):
-    def __init__(self, message=None):
-        self.message = message
-
-
-class ValidationError(RuntimeError):
-    def to_operation_outcome(self) -> dict:
-        pass
-
-
-@dataclass
-class InvalidPatientId(ValidationError):
-    """Use this when NHS Number is invalid or doesn't exist"""
-
-    patient_identifier: str
-
-    def __str__(self):
-        return f"NHS Number: {self.patient_identifier} is invalid or it doesn't exist."
-
-    def to_operation_outcome(self) -> dict:
-        return create_operation_outcome(
-            resource_id=str(uuid.uuid4()),
-            severity=Severity.error,
-            code=Code.server_error,
-            diagnostics=self.__str__(),
-        )
-
-
-@dataclass
-class InconsistentIdError(ValidationError):
-    """Use this when the specified id in the message is inconsistent with the path
-    see: http://hl7.org/fhir/R4/http.html#update"""
-
-    imms_id: str
-
-    def __str__(self):
-        return f"The provided id:{self.imms_id} doesn't match with the content of the message"
-
-    def to_operation_outcome(self) -> dict:
-        return create_operation_outcome(
-            resource_id=str(uuid.uuid4()),
-            severity=Severity.error,
-            code=Code.server_error,
-            diagnostics=self.__str__(),
-        )
-
-
-@dataclass
-class CustomValidationError(ValidationError):
-    """Custom validation error"""
-
-    message: str
-
-    def __str__(self):
-        return self.message
-
-    def to_operation_outcome(self) -> dict:
-        return create_operation_outcome(
-            resource_id=str(uuid.uuid4()),
-            severity=Severity.error,
-            code=Code.invariant,
-            diagnostics=self.__str__(),
-        )
-
-
 @dataclass
 class IdentifierDuplicationError(RuntimeError):
     """Fine grain validation"""
@@ -223,46 +126,3 @@ def create_operation_outcome(resource_id: str, severity: Severity, code: Code, d
             }
         ],
     }
-
-
-@dataclass
-class ParameterException(RuntimeError):
-    message: str
-
-    def __str__(self):
-        return self.message
-
-
-class UnauthorizedSystemError(RuntimeError):
-    def __init__(self, message="Unauthorized system"):
-        super().__init__(message)
-        self.message = message
-    
-    def to_operation_outcome(self) -> dict:
-        return create_operation_outcome(
-            resource_id=str(uuid.uuid4()),
-            severity=Severity.error,
-            code=Code.forbidden,
-            diagnostics=self.message,
-        )
-
-
-class MessageNotSuccessfulError(Exception):
-    """
-    Generic error message for any scenario which either prevents sending to the Imms API, or which results in a
-    non-successful response from the Imms API
-    """
-
-    def __init__(self, message=None):
-        self.message = message
-
-
-class RecordProcessorError(Exception):
-    """
-    Exception for re-raising exceptions which have already occurred in the Record Processor.
-    The diagnostics dictionary received from the Record Processor is passed to the exception as an argument
-    and is stored as an attribute.
-    """
-
-    def __init__(self, diagnostics_dictionary: dict):
-        self.diagnostics_dictionary = diagnostics_dictionary

mns_subscription/poetry.lock

@@ -4,11 +4,11 @@ version     = "0.1.0"
 description = "subscription to nhs number change event"
 authors     = ["Your Name <[email protected]>"]
 license     = "MIT"
-readme      = "README.md"
 packages    = [{include = "src"}]
 
 [tool.poetry.dependencies]
 python                  = "^3.11"
+boto3                   = "~1.38.42"
 pyjwt                   = "~2.10.1"
 moto                    = "^5.1.6"
 requests                = "~2.32.4"

mns_subscription/pyproject.toml

@@ -11,7 +11,7 @@
 
 
 class Service(Enum):
-    PDS = "pds"
+    MNS = "mns"
     IMMUNIZATION = "imms"
 
 
@@ -22,7 +22,7 @@ def __init__(self, service: Service, secret_manager_client, environment, cache:
         self.cache_key = f"{service.value}_access_token"
 
         self.expiry = 30
-        self.secret_name = f"imms/pds/{environment}/jwt-secrets" if service == Service.PDS else \
+        self.secret_name = f"imms/mns/{environment}/jwt-secrets" if service == Service.MNS else \
             f"imms/immunization/{environment}/jwt-secrets"
 
         self.token_url = f"https://{environment}.api.service.nhs.uk/oauth2/token" \

mns_subscription/src/authentication.py

@@ -1,8 +1,27 @@
 from authentication import AppRestrictedAuth
+import os
+from cache import Cache
 from mns_service import MnsService
+import boto3
+from authentication import Service
+from botocore.config import Config
+
+
+def run_subscription():
+    mns_env: str = os.getenv("MNS_ENV", "int")
+
+    boto_config = Config(region_name="eu-west-2")
+    cache = Cache(directory="/tmp")
+    authenticator = AppRestrictedAuth(
+        service=Service.MNS,
+        secret_manager_client=boto3.client("secretsmanager", config=boto_config),
+        environment=mns_env,
+        cache=cache,
+    )
+    mns = MnsService(authenticator, mns_env)
+    return mns.subscribeNotification()
+
 
 if __name__ == "__main__":
-    auth = AppRestrictedAuth()
-    mns = MnsService(authenticator=auth)
-    result = mns.subscribeNotification()
+    result = run_subscription()
     print("Subscription Result:", result)