AS Device read unit tests

AS Device read feature tests

Author: megan-bower4

infrastructure/swagger/05_paths.yaml

@@ -385,6 +385,32 @@ paths:
         - ${authoriser_name}: []
         - app-level0: []
 
+  /ProductTeam/{product_team_id}/Product/{product_id}/Device/AccreditedSystem:
+    post:
+      operationId: createDeviceAccreditedSystem
+      summary: Create a Accredited System Device resource (POST)
+      parameters:
+        - $ref: "#/components/parameters/ProductTeamId"
+        - $ref: "#/components/parameters/ProductId"
+        - $ref: "#/components/parameters/HeaderVersion"
+        - $ref: "#/components/parameters/HeaderRequestId"
+        - $ref: "#/components/parameters/HeaderCorrelationId"
+      requestBody:
+        $ref: "#/components/requestBodies/AccreditedSystemDeviceCreateRequestBody"
+      responses:
+        "201":
+          $ref: "#/components/responses/AsDeviceCreate"
+        "400":
+          $ref: "#/components/responses/BadRequest"
+        "404":
+          $ref: "#/components/responses/NotFound"
+      x-amazon-apigateway-integration:
+        <<: *ApiGatewayIntegration
+        uri: ${method_createDeviceAccreditedSystem}
+      security:
+        - ${authoriser_name}: []
+        - app-level0: []
+
   /searchSdsDevice:
     get:
       operationId: searchsdsdevice

infrastructure/swagger/07_components--schemas--domain.yaml

@@ -167,6 +167,32 @@ components:
         questionnaire_responses:
           type: object
 
+    AsDeviceResponse:
+      type: object
+      properties:
+        id:
+          type: string
+        name:
+          type: string
+        product_id:
+          type: string
+        product_team_id:
+          type: string
+        ods_code:
+          type: string
+        status:
+          type: string
+        created_on:
+          type: string
+        updated_on:
+          type: string
+          nullable: true
+        deleted_on:
+          type: string
+          nullable: true
+        questionnaire_responses:
+          type: object
+
     MhsDeviceResponse:
       type: object
       properties:

infrastructure/swagger/11_components--requestBodies.yaml

@@ -144,3 +144,25 @@ components:
               - questionnaire_responses
           example:
             questionnaire_responses: { "spine_mhs": [{ "question": "answer" }] }
+    AccreditedSystemDeviceCreateRequestBody:
+      description: Request body to create a Accredited System Device object
+      required: true
+      content:
+        application/json:
+          schema:
+            type: object
+            properties:
+              questionnaire_responses:
+                type: object
+                description: Questionnaire Responses for AS Device
+                properties:
+                  spine_as:
+                    type: array
+                    description: spine_as questionnaire associated with the as device
+                    items:
+                      type: object
+                      description: spine_as questionnaire response
+            required:
+              - questionnaire_responses
+          example:
+            questionnaire_responses: { "spine_as": [{ "question": "answer" }] }

infrastructure/swagger/12_components--responses.yaml

@@ -165,6 +165,12 @@ components:
         application/json:
           schema:
             $ref: "#/components/schemas/DeviceResponse"
+    AsDeviceCreate:
+      description: Create Accredited System Device operation successful
+      content:
+        application/json:
+          schema:
+            $ref: "#/components/schemas/AsDeviceResponse"
     MhsDeviceCreate:
       description: Create Message Handling System Device operation successful
       content:

src/api/createDeviceAccreditedSystem/index.py

@@ -0,0 +1,26 @@
+from api_utils.api_step_chain import execute_step_chain
+from event.aws.client import dynamodb_client
+from event.environment import BaseEnvironment
+from event.logging.logger import setup_logger
+
+from .src.v1.steps import steps as v1_steps
+
+
+class Environment(BaseEnvironment):
+    DYNAMODB_TABLE: str
+
+
+versioned_steps = {"1": v1_steps}
+cache = {
+    **Environment.build().dict(),
+    "DYNAMODB_CLIENT": dynamodb_client(),
+}
+
+
+def handler(event: dict, context=None):
+    setup_logger(service_name=__file__)
+    return execute_step_chain(
+        event=event,
+        cache=cache,
+        versioned_steps=versioned_steps,
+    )

src/api/createDeviceAccreditedSystem/make/make.py

@@ -0,0 +1,4 @@
+from builder.lambda_build import build
+
+if __name__ == "__main__":
+    build(__file__)

src/api/createDeviceAccreditedSystem/policies/dynamodb.json

@@ -0,0 +1,6 @@
+[
+  "dynamodb:Query",
+  "dynamodb:PutItem",
+  "dynamodb:UpdateItem",
+  "dynamodb:GetItem"
+]

src/api/createDeviceAccreditedSystem/policies/kms.json

@@ -0,0 +1 @@
+["kms:Decrypt"]

src/api/createDeviceAccreditedSystem/src/v1/steps.py

@@ -0,0 +1,178 @@
+from http import HTTPStatus
+
+from domain.api.common_steps.general import parse_event_body
+from domain.api.common_steps.read_product import (
+    parse_path_params,
+    read_product,
+    read_product_team,
+)
+
+# from domain.core import device
+from domain.core.cpm_product import CpmProduct
+from domain.core.device import (  # DeviceTagAddedEvent,;; DeviceKeyAddedEvent,; DeviceReferenceDataIdAddedEvent,
+    Device,
+    DeviceTagAddedEvent,
+    QuestionnaireResponseUpdatedEvent,
+)
+
+# from domain.core.device_key.v1 import DeviceKeyType
+from domain.core.device_reference_data import DeviceReferenceData
+from domain.core.error import (  # InvalidSpineAsResponse,
+    AccreditedSystemFatalError,
+    ConfigurationError,
+    NotEprProductError,
+)
+from domain.core.product_key import ProductKeyType
+from domain.core.questionnaire import Questionnaire, QuestionnaireResponse
+from domain.repository.device_reference_data_repository import (
+    DeviceReferenceDataRepository,
+)
+from domain.repository.device_repository import DeviceRepository
+from domain.repository.questionnaire_repository import (
+    QuestionnaireInstance,
+    QuestionnaireRepository,
+)
+from domain.request_models import CpmProductPathParams, CreateAsDeviceIncomingParams
+from domain.response.validation_errors import mark_validation_errors_as_inbound
+
+
+@mark_validation_errors_as_inbound
+def parse_as_device_payload(data, cache) -> CreateAsDeviceIncomingParams:
+    payload: dict = data[parse_event_body]
+    return CreateAsDeviceIncomingParams(**payload)
+
+
+def read_device_reference_data(data, cache) -> list[DeviceReferenceData]:
+    path_params: CpmProductPathParams = data[parse_path_params]
+    product_repo = DeviceReferenceDataRepository(
+        table_name=cache["DYNAMODB_TABLE"], dynamodb_client=cache["DYNAMODB_CLIENT"]
+    )
+    device_reference_data = product_repo.search(
+        product_id=path_params.product_id,
+        product_team_id=path_params.product_team_id,
+    )
+
+    # Only 1 AS DRD and only 1 MHS DRD
+
+    if not device_reference_data or len(device_reference_data) < 2:
+        raise ConfigurationError(
+            "You must configure the AS and MessageSet Device Reference Data before creating an AS Device"
+        )
+
+    if len(device_reference_data) > 2:
+        raise AccreditedSystemFatalError(
+            "More that 2 MessageSet Device Reference Data resources were found. This is not allowed"
+        )
+
+    exists = set()
+    if any(drd.name in exists or exists.add(drd.name) for drd in device_reference_data):
+        raise ConfigurationError(
+            "Only one AS and one MessageSet Device Reference Data is allowed before creating an AS Device"
+        )
+
+    return device_reference_data
+
+
+def read_spine_as_questionnaire(data, cache) -> Questionnaire:
+    return QuestionnaireRepository().read(QuestionnaireInstance.SPINE_AS)
+
+
+def validate_spine_as_questionnaire_response(data, cache) -> QuestionnaireResponse:
+    spine_as_questionnaire: Questionnaire = data[read_spine_as_questionnaire]
+    payload: CreateAsDeviceIncomingParams = data[parse_as_device_payload]
+    spine_as_questionnaire_response = payload.questionnaire_responses[
+        QuestionnaireInstance.SPINE_AS
+    ]
+    return spine_as_questionnaire.validate(
+        data=spine_as_questionnaire_response.__root__[0]
+    )
+
+
+def create_as_device(data, cache) -> Device:
+    product: CpmProduct = data[read_product]
+    payload: CreateAsDeviceIncomingParams = data[parse_as_device_payload]
+
+    # Create a new Device dictionary excluding 'questionnaire_responses'
+    device_payload = payload.dict(exclude={"questionnaire_responses"})
+    return product.create_device(**device_payload)
+
+
+def create_party_key_tag(data, cache) -> DeviceTagAddedEvent:
+    as_device: Device = data[create_as_device]
+    return as_device.add_tag(party_key=data[get_party_key])
+
+
+def create_device_keys(data, cache) -> Device:
+    # We will need to add some keys in the future, ASID?
+    as_device: Device = data[create_as_device]
+    return as_device
+
+
+def add_device_reference_data_id(data, cache) -> Device:
+    as_device: Device = data[create_device_keys]
+    device_reference_data: DeviceReferenceData = data[read_device_reference_data]
+    for drd in device_reference_data:
+        as_device.add_device_reference_data_id(
+            device_reference_data_id=str(drd.id), path_to_data=["*.interaction_ids"]
+        )
+    return as_device
+
+
+def add_spine_as_questionnaire_response(
+    data, cache
+) -> QuestionnaireResponseUpdatedEvent:
+    spine_as_questionnaire_response: QuestionnaireResponse = data[
+        validate_spine_as_questionnaire_response
+    ]
+    as_device: Device = data[create_as_device]
+
+    return as_device.add_questionnaire_response(spine_as_questionnaire_response)
+
+
+def write_device(data: dict[str, Device], cache) -> Device:
+    as_device: Device = data[create_as_device]
+    repo = DeviceRepository(
+        table_name=cache["DYNAMODB_TABLE"], dynamodb_client=cache["DYNAMODB_CLIENT"]
+    )
+    return repo.write(as_device)
+
+
+def set_http_status(data, cache) -> tuple[HTTPStatus, dict]:
+    as_device: Device = data[create_as_device]
+    return HTTPStatus.CREATED, as_device.state_exclude_tags()
+
+
+def get_party_key(data, cache) -> str:
+    product: CpmProduct = data[read_product]
+    party_keys = (
+        key.key_value
+        for key in product.keys
+        if key.key_type is ProductKeyType.PARTY_KEY
+    )
+    try:
+        (party_key,) = party_keys
+    except ValueError:
+        raise NotEprProductError(
+            "Not an EPR Product: Cannot create AS device for product without exactly one Party Key"
+        )
+    return party_key
+
+
+steps = [
+    parse_event_body,
+    parse_path_params,
+    parse_as_device_payload,
+    read_product_team,
+    read_product,
+    get_party_key,
+    read_device_reference_data,
+    read_spine_as_questionnaire,
+    validate_spine_as_questionnaire_response,
+    create_as_device,
+    create_party_key_tag,
+    create_device_keys,
+    add_device_reference_data_id,
+    add_spine_as_questionnaire_response,
+    write_device,
+    set_http_status,
+]