Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

feature/PI-603-update_spine_mhs_questionnaire #426

Closed
wants to merge 1 commit into from
Closed

feature/PI-603-update_spine_mhs_questionnaire #426

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
71 changes: 59 additions & 12 deletions src/api/createDeviceMessageHandlingSystem/src/v1/steps.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
from domain.core.error import ConfigurationError
from domain.core.product_team import ProductTeam
from domain.core.questionnaire import Questionnaire, QuestionnaireResponse
from domain.core.timestamp import now
from domain.repository.device_reference_data_repository import (
DeviceReferenceDataRepository,
)
Expand All @@ -39,6 +40,30 @@ def parse_mhs_device_payload(data, cache) -> CreateMhsDeviceIncomingParams:
return CreateMhsDeviceIncomingParams(**payload)


def read_spine_mhs_questionnaire(data, cache) -> Questionnaire:
return QuestionnaireRepository().read(QuestionnaireInstance.SPINE_MHS)


def check_expected_questionnaire_response_fields(data, cache):
spine_mhs_questionnaire: Questionnaire = data[read_spine_mhs_questionnaire]
expected_fields = sorted(spine_mhs_questionnaire.user_provided_fields)
payload: CreateMhsDeviceIncomingParams = data[parse_mhs_device_payload]
spine_mhs_questionnaire_response = payload.questionnaire_responses[
QuestionnaireInstance.SPINE_MHS
]
payload_fields = set(spine_mhs_questionnaire_response.__root__[0].keys())

# Find unexpected fields
unexpected_fields = payload_fields - set(expected_fields)
if unexpected_fields:
raise ConfigurationError(
f"Payload contains unexpected fields: {unexpected_fields}. "
f"Expected fields are: {expected_fields}."
)
# Return questionnaire responses
return spine_mhs_questionnaire_response.__root__[0]


def check_for_existing_mhs(data, cache):
product_team: ProductTeam = data[read_product_team]
product: CpmProduct = data[read_product]
Expand Down Expand Up @@ -82,21 +107,41 @@ def read_device_reference_data(data, cache) -> DeviceReferenceData:
return device_reference_data


def read_spine_mhs_questionnaire(data, cache) -> Questionnaire:
return QuestionnaireRepository().read(QuestionnaireInstance.SPINE_MHS)


def validate_spine_mhs_questionnaire_response(data, cache) -> QuestionnaireResponse:
spine_mhs_questionnaire: Questionnaire = data[read_spine_mhs_questionnaire]
payload: CreateMhsDeviceIncomingParams = data[parse_mhs_device_payload]
def generate_spine_mhs_responses(data, cache) -> list:
spine_mhs_questionnaire_responses = data[
check_expected_questionnaire_response_fields
]

spine_mhs_questionnaire_response = payload.questionnaire_responses[
QuestionnaireInstance.SPINE_MHS
missing_fields_used_for_system_generated_fields = [
field
for field in ["Binding", "MHS FQDN"]
if not spine_mhs_questionnaire_responses.get(field)
]

return spine_mhs_questionnaire.validate(
data=spine_mhs_questionnaire_response.__root__[0]
if missing_fields_used_for_system_generated_fields:
raise ConfigurationError(
f"The following required fields are missing in the response to spine_mhs: {', '.join(missing_fields_used_for_system_generated_fields)}"
)

# System generated fields:
spine_mhs_questionnaire_responses["Address"] = (
f"{spine_mhs_questionnaire_responses.get("Binding")}{spine_mhs_questionnaire_responses.get("MHS FQDN")}"
)
spine_mhs_questionnaire_responses["MHS Party key"] = data[get_party_key]
spine_mhs_questionnaire_responses["Managing Organization"] = (
spine_mhs_questionnaire_responses["MHS Party key"].split("-")[0]
)
spine_mhs_questionnaire_responses["Date Approved"] = str(now())
spine_mhs_questionnaire_responses["Date Requested"] = str(now())
spine_mhs_questionnaire_responses["Date DNS Approved"] = str(None)

return spine_mhs_questionnaire_responses


def validate_spine_mhs_questionnaire_response(data, cache) -> QuestionnaireResponse:
spine_mhs_questionnaire: Questionnaire = data[read_spine_mhs_questionnaire]
spine_mhs_response = data[generate_spine_mhs_responses]
return spine_mhs_questionnaire.validate(spine_mhs_response)


def create_mhs_device(data, cache) -> Device:
Expand Down Expand Up @@ -173,12 +218,14 @@ def set_http_status(data, cache) -> tuple[HTTPStatus, dict]:
parse_event_body,
parse_path_params,
parse_mhs_device_payload,
read_spine_mhs_questionnaire,
check_expected_questionnaire_response_fields,
read_product_team,
read_product,
get_party_key,
check_for_existing_mhs,
read_device_reference_data,
read_spine_mhs_questionnaire,
generate_spine_mhs_responses,
validate_spine_mhs_questionnaire_response,
create_mhs_device,
create_party_key_tag,
Expand Down
53 changes: 34 additions & 19 deletions src/api/createDeviceMessageHandlingSystem/tests/test_index.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,25 +35,26 @@
PRODUCT_TEAM_NAME = "My Product Team"
PRODUCT_NAME = "My Product"
VERSION = 1
PARTY_KEY = "ABC1234-987654"

QUESTIONNAIRE_DATA = {
"Address": "http://example.com",
"Unique Identifier": "123456",
"Managing Organization": "Example Org",
"MHS Party key": "party-key-001",
"MHS CPA ID": "cpa-id-001",
"Approver URP": "approver-123",
"Contract Property Template Key": "contract-key-001",
"Date Approved": "2024-01-01",
"Date DNS Approved": "2024-01-02",
"Date Requested": "2024-01-03",
"DNS Approver": "dns-approver-456",
"Interaction Type": "FHIR",
"Binding": "https://",
"MHS FQDN": "mhs.example.com",
"MHS Is Authenticated": "PERSISTENT",
"Product Key": "product-key-001",
"Requestor URP": "requestor-789",
"MHS Manufacturer Organisation": "AAA",
"MHS Service Description": "Example Description",
"MHS Manufacturer Organisation": "F5H1R",
"Product Name": "Product Name",
"Product Version": 1,
"Approver URP": "UI provided",
"DNS Approver": "UI provided",
"Requestor URP": "UI provided",
}
QUESTIONNAIRE_DATA_SYSTEM_GENERATED_FIELDS = {
"Address": "https://mhs.example.com",
"MHS Party key": PARTY_KEY,
"Date Approved": "datetime",
"Date DNS Approved": "None",
"Date Requested": "datetime",
"Managing Organization": "ABC1234",
}


Expand All @@ -77,7 +78,7 @@ def mock_epr_product_with_message_set_drd() -> (
product = product_team.create_cpm_product(
name=PRODUCT_NAME, product_id=PRODUCT_ID
)
product.add_key(key_type=ProductKeyType.PARTY_KEY, key_value="ABC1234-987654")
product.add_key(key_type=ProductKeyType.PARTY_KEY, key_value=PARTY_KEY)
product_repo = CpmProductRepository(
table_name=TABLE_NAME, dynamodb_client=client
)
Expand Down Expand Up @@ -215,7 +216,9 @@ def test_index() -> None:
questionnaire_responses = device.questionnaire_responses["spine_mhs/1"]
assert len(questionnaire_responses) == 1
questionnaire_response = questionnaire_responses[0]
assert questionnaire_response.data == QUESTIONNAIRE_DATA
assert len(questionnaire_response.data) == len(QUESTIONNAIRE_DATA) + len(
QUESTIONNAIRE_DATA_SYSTEM_GENERATED_FIELDS
)

# Retrieve the created resource
repo = DeviceRepository(
Expand Down Expand Up @@ -299,13 +302,25 @@ def test_incoming_errors(body, path_parameters, error_code, status_code):
(
{
"questionnaire_responses": {
"spine_mhs": [{"Address": "http://example.com"}]
"spine_mhs": [
{"Binding": "https://", "MHS FQDN": "mhs.example.com"}
]
}
},
"MISSING_VALUE",
"Failed to validate data against 'spine_mhs/1': 'MHS Manufacturer Organisation' is a required property",
400,
),
(
{
"questionnaire_responses": {
"spine_mhs": [{"MHS Manufacturer Organisation": "F5H1R"}]
}
},
"VALIDATION_ERROR",
"The following required fields are missing in the response to spine_mhs: Binding, MHS FQDN",
400,
),
],
)
def test_questionnaire_response_validation_errors(
Expand Down
9 changes: 2 additions & 7 deletions src/api/readDevice/tests/test_index.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -190,21 +190,16 @@ def test_index_mhs_device(version):
)
spine_mhs_questionnaire_response = mhs_message_set_questionnaire.validate(
data={
"Address": "http://example.com",
"Unique Identifier": "123456",
"Binding": "https://",
"Address": "https://example.com",
"Managing Organization": "Example Org",
"MHS Party key": "party-key-001",
"MHS CPA ID": "cpa-id-001",
"Approver URP": "approver-123",
"Contract Property Template Key": "contract-key-001",
"Date Approved": "2024-01-01",
"Date DNS Approved": "2024-01-02",
"Date Requested": "2024-01-03",
"DNS Approver": "dns-approver-456",
"Interaction Type": "FHIR",
"MHS FQDN": "mhs.example.com",
"MHS Is Authenticated": "PERSISTENT",
"Product Key": "product-key-001",
"Requestor URP": "requestor-789",
"MHS Manufacturer Organisation": "AAA",
}
Expand Down
2 changes: 1 addition & 1 deletion src/api/readQuestionnaire/tests/test_index.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,7 @@ def test_index():
"name": "spine_mhs",
"version": "1",
}
assert len(json_schema["properties"]) == 25
assert len(json_schema["properties"]) == 16


def test_index_no_such_questionnaire():
Expand Down
Loading
Loading