[NDR-293] Use EICAR file in virus test

.github/workflows/ndr-e2e-test-sandbox.yml

@@ -32,7 +32,7 @@ jobs:
       sandbox: ${{ inputs.sandbox }}
     secrets:
       AWS_ASSUME_ROLE: ${{ secrets.AWS_ASSUME_ROLE }}
-  
+
   services-fhir-api-e2etest:
     uses: ./.github/workflows/base-e2e-fhir-backendtest.yml
     with:

lambdas/tests/e2e/api/fhir/test_upload_document_fhir_api.py

@@ -95,25 +95,25 @@ def test_create_document_virus(test_data):
         "ods": "H81109",
         "nhs_number": "9730154260",
     }
+
+    # Attach EICAR data
+    eicar_string = r"X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
+    record["data"] = base64.b64encode(eicar_string.encode()).decode()
     payload = pdm_data_helper.create_upload_payload(record)
 
     raw_upload_response = upload_document(payload)
     assert raw_upload_response.status_code == 200
-    record["id"] = raw_upload_response.json()["id"].split("~")[1]
-    test_data.append(record)
-
-    # Presigned upload
     upload_response = raw_upload_response.json()
-    presign_uri = upload_response["content"][0]["attachment"]["url"]
-    del upload_response["content"][0]["attachment"]["url"]
-    sample_pdf_path = os.path.join(os.path.dirname(__file__), "files", "dummy.pdf")
-    with open(sample_pdf_path, "rb") as f:
-        presign_response = requests.put(presign_uri, files={"file": f})
-        assert presign_response.status_code == 200
+    record["id"] = upload_response["id"].split("~")[1]
+    test_data.append(record)
 
+    # Poll until processing/scan completes
     def condition(response_json):
         logging.info(response_json)
-        return response_json.get("docStatus", False) == "cancelled"
+        return response_json.get("docStatus") in (
+        "cancelled",
+        "final",
+        )
 
     raw_retrieve_response = retrieve_document_with_retry(
         upload_response["id"], condition

lambdas/tests/e2e/api/test_upload_document_api.py

@@ -164,6 +164,10 @@ def test_create_document_virus(test_data, snapshot_json):
 
     lloyd_george_record["nhs_number"] = "9730154260"
 
+    # Attach EICAR data
+    eicar_string = r"X5O!P%@AP[4\PZX54(P^)7CC)7}$EICAR-STANDARD-ANTIVIRUS-TEST-FILE!$H+H*"
+    lloyd_george_record["data"] = base64.b64encode(eicar_string.encode()).decode()
+
     payload = create_upload_payload(lloyd_george_record)
     url = f"https://{API_ENDPOINT}/FhirDocumentReference"
     headers = {"Authorization": "Bearer 123", "X-Api-Key": API_KEY}
@@ -172,27 +176,23 @@ def test_create_document_virus(test_data, snapshot_json):
     upload_response = retrieve_response.json()
     lloyd_george_record["id"] = upload_response["id"].split("~")[1]
     test_data.append(lloyd_george_record)
-    presign_uri = upload_response["content"][0]["attachment"]["url"]
-    del upload_response["content"][0]["attachment"]["url"]
-
-    sample_pdf_path = os.path.join(os.path.dirname(__file__), "files", "dummy.pdf")
-    with open(sample_pdf_path, "rb") as f:
-        files = {"file": f}
-        presign_response = requests.put(presign_uri, files=files)
-        assert presign_response.status_code == 200
 
     retrieve_url = (
         f"https://{API_ENDPOINT}/FhirDocumentReference/{upload_response['id']}"
     )
 
+    # Poll until processing/scan completes
     def condition(response_json):
         logging.info(response_json)
-        return response_json.get("docStatus", False) == "cancelled"
+        return response_json.get("docStatus") in (
+        "cancelled",
+        "final",
+        )
 
     raw_retrieve_response = fetch_with_retry(retrieve_url, condition)
     retrieve_response = raw_retrieve_response.json()
 
-    assert upload_response == snapshot_json(exclude=paths("id", "date"))
+    assert upload_response == snapshot_json(exclude=paths("id", "date", "content.0.attachment.url"))
     assert retrieve_response == snapshot_json(exclude=paths("id", "date"))
 
 

lambdas/tests/e2e/conftest.py

@@ -15,7 +15,7 @@
 LG_UNSTITCHED_TABLE = os.environ.get("LG_UNSTITCHED_TABLE")
 BULK_REPORT_TABLE = os.environ.get("BULK_REPORT_TABLE")
 LLOYD_GEORGE_S3_BUCKET = data_helper.s3_bucket
-APIM_ENDPOINT = "internal-dev.api.service.nhs.uk"
+APIM_ENDPOINT = data_helper.apim_url
 PDM_SNOMED = 717391000000106
 MTLS_ENDPOINT = os.environ.get("MTLS_ENDPOINT")
 CLIENT_CERT_PATH = os.environ.get("CLIENT_CERT_PATH")

lambdas/tests/e2e/helpers/data_helper.py

@@ -24,6 +24,7 @@ def __init__(
         self.record_type = record_type
         self.dynamo_service = DynamoDBService()
         self.s3_service = S3Service()
+        self.apim_url = None
 
         self.build_env(table_name, bucket_name)
 
@@ -33,6 +34,14 @@ def build_env(self, table_name, bucket_name):
         self.dynamo_table = f"{self.workspace}_{table_name}"
         self.s3_bucket = f"{self.workspace}-{bucket_name}"
 
+        apim_map = {
+            "pre-prod": "int.api.service.nhs.uk",
+            "ndr-test": "internal-qa.api.service.nhs.uk",
+            "ndr-dev":  "internal-dev.api.service.nhs.uk",
+            }
+        self.apim_url = apim_map.get(str(self.workspace), "internal-dev.api.service.nhs.uk")
+
+
     def build_record(
         self, nhs_number="9912003071", data=None, doc_status=None, size=None
     ):