Mav 2831

mavis/test/constants.py

@@ -309,6 +309,24 @@ def programme(self) -> Programme:
         }
         return programme_mapping[self]
 
+    @property
+    def target_disease_code(self) -> str:
+        target_disease_mapping = {
+            Vaccine.FLUENZ: "6142004",
+            Vaccine.SEQUIRUS: "6142004",
+            Vaccine.GARDASIL_9: "240532009",
+        }
+        return target_disease_mapping[self]
+
+    @property
+    def target_disease_display(self) -> str:
+        target_disease_display_mapping = {
+            Vaccine.FLUENZ: "Influenza",
+            Vaccine.SEQUIRUS: "Influenza",
+            Vaccine.GARDASIL_9: "Human papillomavirus infection",
+        }
+        return target_disease_display_mapping[self]
+
 
 class DeliverySite(StrEnum):
     LEFT_ARM_UPPER = "Left arm (upper position)"
@@ -326,6 +344,18 @@ def from_imms_api_code(cls, code: str) -> "DeliverySite":
         }
         return sites[code]
 
+    @property
+    def imms_api_code(self) -> str:
+        """Get the IMMS API code for this delivery site."""
+        codes = {
+            DeliverySite.LEFT_ARM_UPPER: "368208006",
+            DeliverySite.RIGHT_ARM_UPPER: "368209003",
+            DeliverySite.LEFT_ARM_LOWER: "368208006",  # Use same as upper for lower
+            DeliverySite.RIGHT_ARM_LOWER: "368209003",  # Use same as upper for lower
+            DeliverySite.NOSE: "279549004",
+        }
+        return codes[self]
+
 
 class ConsentRefusalReason(StrEnum):
     VACCINE_ALREADY_RECEIVED = "Vaccine already received"
@@ -434,6 +464,7 @@ def generate_name(self) -> str:
 
 class ImmsEndpoints(StrEnum):
     READ = "/immunisation-fhir-api/FHIR/R4/Immunization"
+    CREATE = "/immunisation-fhir-api/FHIR/R4/Immunization"
 
     @property
     def to_url(self) -> str:

mavis/test/data/fhir_immunization_template.json.template

@@ -0,0 +1,123 @@
+{
+  "resourceType": "Immunization",
+  "identifier": [
+    {
+      "system": "https://tools.ietf.org/html/rfc4122",
+      "value": "<<IMMUNIZATION_ID>>"
+    }
+  ],
+  "extension": [
+    {
+      "url": "https://fhir.hl7.org.uk/StructureDefinition/Extension-UKCore-VaccinationProcedure",
+      "valueCodeableConcept": {
+        "coding": [
+          {
+            "system": "http://snomed.info/sct",
+            "code": "<<VACCINE_CODE>>",
+            "display": "<<VACCINE_NAME>>"
+          }
+        ]
+      }
+    }
+  ],
+  "contained": [
+    {
+      "resourceType": "Patient",
+      "id": "Patient1",
+      "identifier": [
+        {
+          "system": "https://fhir.nhs.uk/Id/nhs-number",
+          "value": "<<PATIENT_NHS_NUMBER>>"
+        }
+      ],
+      "name": [
+        {
+          "family": "<<PATIENT_FAMILY_NAME>>",
+          "given": [
+            "<<PATIENT_GIVEN_NAME>>"
+          ]
+        }
+      ],
+      "gender": "<<PATIENT_GENDER>>",
+      "birthDate": "<<PATIENT_BIRTH_DATE>>",
+      "address": [
+        {
+          "postalCode": "<<PATIENT_POSTAL_CODE>>"
+        }
+      ]
+    }
+  ],
+  "status": "completed",
+  "vaccineCode": {
+    "coding": [
+      {
+        "system": "http://snomed.info/sct",
+        "code": "<<VACCINE_CODE>>",
+        "display": "<<VACCINE_NAME>>"
+      }
+    ]
+  },
+  "patient": {
+    "reference": "#Patient1"
+  },
+  "occurrenceDateTime": "<<VACCINATION_TIME>>",
+  "recorded": "<<RECORDED_TIME>>",
+  "primarySource": true,
+  "location": {
+    "identifier": {
+      "system": "https://fhir.nhs.uk/Id/ods-organization-code",
+      "value": "<<SCHOOL_URN>>"
+    }
+  },
+  "site": {
+    "coding": [
+      {
+        "system": "http://snomed.info/sct",
+        "code": "<<DELIVERY_SITE_CODE>>",
+        "display": "<<DELIVERY_SITE_DISPLAY>>"
+      }
+    ]
+  },
+  "performer": [
+    {
+      "function": {
+        "coding": [
+          {
+            "system": "http://terminology.hl7.org/CodeSystem/v2-0443",
+            "code": "AP",
+            "display": "Administering Provider"
+          }
+        ]
+      },
+      "actor": {
+        "display": "Test Nurse"
+      }
+    },
+    {
+      "actor": {
+        "type": "Organization",
+        "identifier": {
+          "system": "https://fhir.nhs.uk/Id/ods-organization-code",
+          "value": "<<SCHOOL_URN>>"
+        },
+        "display": "School Organization"
+      }
+    }
+  ],
+  "protocolApplied": [
+    {
+      "targetDisease": [
+        {
+          "coding": [
+            {
+              "system": "http://snomed.info/sct",
+              "code": "<<TARGET_DISEASE_CODE>>",
+              "display": "<<TARGET_DISEASE_DISPLAY>>"
+            }
+          ]
+        }
+      ],
+      "doseNumberPositiveInt": 1
+    }
+  ]
+}

mavis/test/data/file_generator.py

@@ -8,13 +8,7 @@
 
 from mavis.test.constants import Programme
 from mavis.test.data.file_mappings import FileMapping
-from mavis.test.data_models import (
-    Child,
-    Clinic,
-    Organisation,
-    School,
-    User,
-)
+from mavis.test.data_models import Child, Clinic, Organisation, School, User
 from mavis.test.utils import (
     get_current_datetime_compact,
     get_current_time_hms_format,

mavis/test/data/file_utils.py

@@ -1,10 +1,13 @@
+import json
+import uuid
+from datetime import datetime
 from pathlib import Path
 
 import pandas as pd
 
-from mavis.test.utils import (
-    normalize_whitespace,
-)
+from mavis.test.constants import DeliverySite, Vaccine
+from mavis.test.data_models import Child, School
+from mavis.test.utils import get_current_datetime, normalize_whitespace
 
 
 def read_scenario_list_from_file(input_file_path: Path) -> str | None:
@@ -53,3 +56,52 @@ def increment_date_of_birth_for_records(file_path: Path) -> None:
         _file_df["CHILD_DATE_OF_BIRTH"],
     ) + pd.Timedelta(days=1)
     _file_df.to_csv(file_path, index=False)
+
+
+def create_fhir_immunization_payload(
+    vaccine: Vaccine,
+    child: Child,
+    school: School,
+    delivery_site: DeliverySite,
+    vaccination_time: datetime,
+) -> dict:
+    """Create a FHIR Immunization resource payload using FileGenerator pattern.
+
+    This loads a FHIR R4 Immunization template and substitutes placeholders
+    using the same pattern as FileGenerator for consistency.
+    """
+    # Load template file
+    template_path = Path(__file__).parent / "fhir_immunization_template.json.template"
+
+    with template_path.open("r") as f:
+        template_content = f.read()
+
+    # Generate unique IDs for this immunization
+    immunization_id = str(uuid.uuid4())
+
+    # Create replacements dictionary using FileGenerator pattern
+    replacements = {
+        "<<IMMUNIZATION_ID>>": immunization_id,
+        "<<VACCINE_CODE>>": vaccine.imms_api_code,
+        "<<VACCINE_NAME>>": vaccine.name,
+        "<<PATIENT_NHS_NUMBER>>": child.nhs_number,
+        "<<PATIENT_FAMILY_NAME>>": child.last_name,
+        "<<PATIENT_GIVEN_NAME>>": child.first_name,
+        "<<PATIENT_GENDER>>": "unknown",  # Child model doesn't have gender
+        "<<PATIENT_BIRTH_DATE>>": child.date_of_birth.strftime("%Y-%m-%d"),
+        "<<PATIENT_POSTAL_CODE>>": child.address[3],
+        "<<VACCINATION_TIME>>": vaccination_time.strftime("%Y-%m-%dT%H:%M:%S+00:00"),
+        "<<RECORDED_TIME>>": get_current_datetime().strftime("%Y-%m-%dT%H:%M:%S+00:00"),
+        "<<SCHOOL_URN>>": school.urn,
+        "<<DELIVERY_SITE_CODE>>": delivery_site.imms_api_code,
+        "<<DELIVERY_SITE_DISPLAY>>": delivery_site.value,
+        "<<TARGET_DISEASE_CODE>>": vaccine.target_disease_code,
+        "<<TARGET_DISEASE_DISPLAY>>": vaccine.target_disease_display,
+    }
+
+    # Apply replacements using FileGenerator pattern
+    payload_content = template_content
+    for placeholder, value in replacements.items():
+        payload_content = payload_content.replace(placeholder, value)
+
+    return json.loads(payload_content)

mavis/test/fixtures/helpers.py

@@ -227,6 +227,16 @@ def _setup(programme_group):
                 if vaccine.programme.group == programme_group
             }
             VaccinesPage(page).header.click_mavis_header()
+            DashboardPage(page).click_sessions()
+            SessionsSearchPage(page).click_session_for_programme_group(
+                school, programme_group
+            )
+            if not SessionsOverviewPage(page).is_date_scheduled(get_offset_date(0)):
+                SessionsOverviewPage(page).schedule_or_edit_session()
+                SessionsEditPage(page).schedule_a_valid_session(
+                    offset_days=0, skip_weekends=False
+                )
+            SessionsOverviewPage(page).header.click_mavis_header()
             DashboardPage(page).click_schools()
             SchoolsSearchPage(page).click_school(school)
             SchoolsChildrenPage(page).click_import_class_lists()

mavis/test/helpers/imms_api_helper.py

@@ -6,11 +6,8 @@
 import dateutil.parser
 import requests
 
-from mavis.test.constants import (
-    DeliverySite,
-    ImmsEndpoints,
-    Vaccine,
-)
+from mavis.test.constants import DeliverySite, ImmsEndpoints, Vaccine
+from mavis.test.data.file_utils import create_fhir_immunization_payload
 from mavis.test.data_models import Child, School
 
 
@@ -196,3 +193,56 @@ def _get_imms_api_record_for_child(
         response.raise_for_status()
 
         return ImmsApiVaccinationRecord.from_response(response)
+
+    def create_vaccination_record(
+        self,
+        vaccine: Vaccine,
+        child: Child,
+        school: School,
+        delivery_site: DeliverySite,
+        vaccination_time: datetime,
+    ) -> None:
+        """Create a vaccination record in the IMMS API.
+
+        Args:
+            vaccine: The vaccine that was administered
+            child: The child who received the vaccination
+            school: The school where vaccination took place
+            delivery_site: Anatomical site of delivery
+            vaccination_time: When the vaccination occurred
+
+        Returns:
+            True if record was created successfully
+
+        Raises:
+            requests.HTTPError: If the API request fails
+        """
+        # Create FHIR Immunization resource payload
+        immunization_payload = create_fhir_immunization_payload(
+            vaccine=vaccine,
+            child=child,
+            school=school,
+            delivery_site=delivery_site,
+            vaccination_time=vaccination_time,
+        )
+
+        # Create headers for POST request
+        create_headers = self.headers.copy()
+        create_headers["content-type"] = "application/fhir+json"
+
+        response = requests.post(
+            url=ImmsEndpoints.CREATE.to_url,
+            headers=create_headers,
+            json=immunization_payload,
+            timeout=30,
+        )
+        response.raise_for_status()
+
+        # If creation was successful, verify the record exists in the API
+        self.check_record_in_imms_api(
+            vaccine=vaccine,
+            child=child,
+            school=school,
+            delivery_site=delivery_site,
+            vaccination_time=vaccination_time,
+        )