Add smoke test for running gateway services

scripts/bash/package_release.sh

@@ -63,6 +63,7 @@ echo ""
 
 REQUIRED_FILES=("src/" "sample_images/" "scripts/python/database.py" \
     "scripts/powershell/maintenance.ps1" "scripts/powershell/debug_toolkit.ps1" \
+    "scripts/python/smoke_test.py" \
     "pyproject.toml" "uv.lock" "README.md" "LICENCE.md")
 
 for item in "${REQUIRED_FILES[@]}"; do

scripts/powershell/smoke_test.ps1

@@ -3,8 +3,10 @@
 .SYNOPSIS
     Smoke test for the gateway services.
 .DESCRIPTION
-    Checks all 4 Windows services are Running, then sends a DICOM C-ECHO to the
-    PACS server on localhost to verify it is accepting connections.
+    Checks all 4 Windows services are Running.
+    Seeds a test worklist item on the MWL server and verifies it can be found via C-FIND.
+    Sends a test DICOM image to the PACS server via C-STORE.
+    Verifies the DICOM image upload was attempted.
     Executed on the Arc-enabled VM via az connectedmachine run-command.
 #>
 
@@ -23,7 +25,7 @@ foreach ($name in $serviceNames) {
     Write-Output "OK: $name is Running"
 }
 
-# -- DICOM C-ECHO -------------------------------------------------------------
+# -- DICOM C-FIND and C-STORE -------------------------------------------------
 
 $installPath = 'C:\Program Files\NHS\ManageBreastScreeningGateway\current'
 if (-not (Test-Path $installPath)) {
@@ -33,30 +35,9 @@ if (-not (Test-Path $installPath)) {
 Set-Location $installPath
 $env:PYTHONPATH = 'src'
 
-$echoScript = @'
-from pynetdicom import AE
-from pynetdicom.sop_class import Verification
-ae = AE()
-ae.add_requested_context(Verification)
-assoc = ae.associate("127.0.0.1", 4244)
-if not assoc.is_established:
-    raise SystemExit("PACS association failed — check Gateway-PACS service logs")
-status = assoc.send_c_echo()
-assoc.release()
-if not status or status.Status != 0x0000:
-    raise SystemExit("C-ECHO returned unexpected status: " + hex(status.Status if status else 0))
-print("C-ECHO OK")
-'@
-
-$tempPy = Join-Path $env:TEMP 'gw_smoke_echo.py'
-try {
-    [System.IO.File]::WriteAllText($tempPy, $echoScript, [System.Text.Encoding]::UTF8)
-    & '.venv\Scripts\python.exe' $tempPy
-    if ($LASTEXITCODE -ne 0) {
-        throw "DICOM C-ECHO failed (exit code: $LASTEXITCODE)"
-    }
-} finally {
-    Remove-Item $tempPy -ErrorAction SilentlyContinue
+& '.venv\Scripts\pytest.exe' 'scripts\python\smoke_test.py'
+if ($LASTEXITCODE -ne 0) {
+    throw "DICOM C-FIND and C-STORE smoke test failed (exit code: $LASTEXITCODE)"
 }
 
 Write-Output "=== Smoke test passed ==="

scripts/python/smoke_test.py

@@ -0,0 +1,183 @@
+import datetime
+import logging
+import os
+import random
+import time
+
+import numpy as np
+from dotenv import load_dotenv
+from PIL import Image
+from pydicom.dataset import Dataset, FileMetaDataset
+from pydicom.uid import ExplicitVRLittleEndian, generate_uid
+from pynetdicom import AE
+from pynetdicom.sop_class import (
+    DigitalMammographyXRayImageStorageForPresentation,
+    ModalityWorklistInformationFind,
+)
+
+from services.dicom import PENDING, SUCCESS
+from services.mwl.create_worklist_item import CreateWorklistItem
+from services.storage import MWLStorage, PACSStorage
+
+logger = logging.getLogger(__name__)
+
+load_dotenv()
+
+if os.getenv("ENVIRONMENT", "").lower() == "prod":
+    raise Exception("Smoke tests are not intended to be run in production environments.")
+
+seed = random.randint(1000000, 9999999)
+SAMPLE_IMAGES_PATH = os.getenv("SAMPLE_IMAGES_PATH", "sample_images")
+TEST_ACCESSION_NUMBER = f"SMOKE-{seed}"  # gitleaks:allow
+TEST_PATIENT_ID = f"999{seed}"  # gitleaks:allow
+TEST_PATIENT_NAME = f"TEST^{seed}"  # gitleaks:allow
+TEST_PATIENT_BIRTH_DATE = "19900101"  # gitleaks:allow
+TEST_SCHEDULED_DATE = datetime.date.today().strftime("%Y%m%d")
+TEST_SCHEDULED_TIME = datetime.datetime.now().strftime("%H%M%S")
+TEST_STUDY_ID = f"STUDY{seed}"  # gitleaks:allow
+
+
+# This smoke test is designed to verify the end-to-end functionality
+# of the Relay Listener, MWL and PACS services on a deployed,
+# Managed Identity secured Gateway deployed to Azure.
+
+
+def generate_dicom_dataset():
+    img_path = f"{SAMPLE_IMAGES_PATH}/LCC.jpg"
+    img = Image.open(img_path).convert("L")
+    columns, rows = img.size
+    pixel_array = np.array(img, dtype=np.uint8)
+    pixel_bytes = pixel_array.tobytes()
+    if len(pixel_bytes) % 2 != 0:
+        pixel_bytes += b"\x00"
+
+    file_meta = FileMetaDataset()
+    file_meta.MediaStorageSOPClassUID = DigitalMammographyXRayImageStorageForPresentation
+    file_meta.MediaStorageSOPInstanceUID = generate_uid()
+    file_meta.ImplementationClassUID = generate_uid()
+    file_meta.TransferSyntaxUID = ExplicitVRLittleEndian
+    ds = Dataset()
+    ds.SOPClassUID = file_meta.MediaStorageSOPClassUID
+    ds.SOPInstanceUID = file_meta.MediaStorageSOPInstanceUID
+    ds.PatientName = TEST_PATIENT_NAME
+    ds.PatientID = TEST_PATIENT_ID
+    ds.PatientBirthDate = TEST_PATIENT_BIRTH_DATE
+    ds.PatientSex = "F"
+    ds.StudyDate = TEST_SCHEDULED_DATE
+    ds.StudyTime = TEST_SCHEDULED_TIME
+    ds.StudyInstanceUID = generate_uid()
+    ds.StudyID = TEST_STUDY_ID
+    ds.AccessionNumber = TEST_ACCESSION_NUMBER
+    ds.SeriesInstanceUID = generate_uid()
+    ds.SeriesNumber = 1
+    ds.InstanceNumber = 1
+    ds.Modality = "MG"
+    ds.PhotometricInterpretation = "MONOCHROME2"
+    ds.Rows = rows
+    ds.Columns = columns
+    ds.BitsAllocated = 8
+    ds.BitsStored = 8
+    ds.HighBit = 7
+    ds.PixelRepresentation = 0
+    ds.PixelData = pixel_bytes
+    ds.ImageLaterality = "L"
+    ds.ViewPosition = "CC"
+    ds.file_meta = file_meta
+
+    return ds
+
+
+def test_create_worklist_item():
+    payload = {
+        "action_id": "action-12345",
+        "action_type": "worklist.create_item",
+        "parameters": {
+            "worklist_item": {
+                "participant": {
+                    "nhs_number": TEST_PATIENT_ID,
+                    "name": TEST_PATIENT_NAME,
+                    "birth_date": TEST_PATIENT_BIRTH_DATE,
+                    "sex": "F",
+                },
+                "scheduled": {
+                    "date": "20240615",
+                    "time": "101500",
+                },
+                "procedure": {
+                    "modality": "MG",
+                    "study_description": "MAMMOGRAPHY",
+                },
+                "accession_number": TEST_ACCESSION_NUMBER,
+            }
+        },
+    }
+
+    response = CreateWorklistItem(MWLStorage(os.environ["MWL_DB_PATH"])).call(payload)
+    assert response == {"action_id": "action-12345", "status": "created"}
+
+
+def test_c_find_worklist_item():
+    mwl_host = os.getenv("MWL_HOST", "127.0.0.1")
+    mwl_port = int(os.getenv("MWL_PORT", "4243"))
+    ae = AE(ae_title="SMOKE_TEST_AET")
+    ae.add_requested_context(ModalityWorklistInformationFind)
+
+    logger.info("Associating with MWL server %s at %s:%s", os.environ["MWL_AET"], mwl_host, mwl_port)
+
+    assoc = ae.associate(mwl_host, mwl_port, ae_title=os.environ["MWL_AET"])
+    assert assoc.is_established, "Failed to establish C-FIND association"
+
+    query = Dataset()
+    query.PatientID = TEST_PATIENT_ID
+
+    responses = list(
+        assoc.send_c_find(
+            query,
+            query_model=ModalityWorklistInformationFind,
+        )
+    )
+    assoc.release()
+
+    assert len(responses) == 2, "Unexpected number of C-FIND responses"
+
+    status, ds = responses[0]
+    assert status.Status == PENDING, "C-FIND response status is not PENDING"
+    assert ds.PatientID == TEST_PATIENT_ID, "C-FIND response PatientID does not match"
+
+
+def test_c_store_dicom_image():
+    pacs_host = os.getenv("PACS_HOST", "127.0.0.1")
+    pacs_port = int(os.getenv("PACS_PORT", "4244"))
+    ae = AE(ae_title="SMOKE_TEST_AET")
+    ae.add_requested_context(DigitalMammographyXRayImageStorageForPresentation)
+    pacs_assoc = ae.associate(pacs_host, pacs_port, ae_title=os.environ["PACS_AET"])
+
+    assert pacs_assoc.is_established, "Failed to establish C-STORE association"
+
+    ds = generate_dicom_dataset()
+    response = pacs_assoc.send_c_store(ds)
+    assert response.Status == SUCCESS, f"C-STORE failed with status: 0x{response.Status:04X}"
+
+
+def test_image_stored_in_pacs():
+    storage = PACSStorage(os.environ["PACS_DB_PATH"], os.environ["PACS_STORAGE_PATH"])
+    stored_image = storage.get_instance_by_accession(TEST_ACCESSION_NUMBER)
+    assert stored_image["patient_id"] == TEST_PATIENT_ID, "Stored image PatientID does not match"
+    assert stored_image["accession_number"] == TEST_ACCESSION_NUMBER, "Stored image AccessionNumber does not match"
+    assert stored_image["storage_path"] is not None, "Stored image storage_path is None"
+
+
+def test_upload_attempted_for_stored_image():
+    storage = PACSStorage(os.environ["PACS_DB_PATH"], os.environ["PACS_STORAGE_PATH"])
+    max_wait_time = 10
+    upload_attempted = False
+    for _ in range(max_wait_time):
+        stored_image = storage.get_instance_by_accession(TEST_ACCESSION_NUMBER)
+        # We expect a FAILED upload as the smoke test action id won't match anything in Rubie
+        # or Rubie won't be available to receive the upload.
+        if stored_image["upload_status"] == "FAILED":
+            upload_attempted = True
+            break
+        time.sleep(1)
+
+    assert upload_attempted, f"No upload attempt detected within {max_wait_time} seconds"