wip

Author: Harriethw

manage_breast_screening/notifications/tests/integration/test_load_test.py

@@ -0,0 +1,48 @@
+import os
+from datetime import datetime
+
+import pytest
+
+from manage_breast_screening.notifications.management.commands.create_appointments import (
+    Command,
+)
+from manage_breast_screening.notifications.services.blob_storage import BlobStorage
+from manage_breast_screening.notifications.tests.integration.helpers import Helpers
+from scripts.python.load_test.notifications_load_test import generate_load_test_data
+
+
+# TODO: DELETE - setup as easy way to validate data locally
+@pytest.mark.integration
+class TestLoadTest:
+    @pytest.fixture(autouse=True)
+    def setup(self, monkeypatch):
+        monkeypatch.setenv(
+            "BLOB_STORAGE_CONNECTION_STRING", Helpers().azurite_connection_string()
+        )
+        monkeypatch.setenv("BLOB_CONTAINER_NAME", "nbss-appoinments-data")
+
+    @pytest.fixture
+    def helpers(self):
+        return Helpers()
+
+    @pytest.mark.django_db
+    def test_generate_test_data(self, helpers):
+        for number in range(10):
+            with open(
+                f"{os.path.dirname(os.path.realpath(__file__))}/../fixtures/load_data_{number}.dat",
+                "w",
+            ) as file:
+                data_string = generate_load_test_data(number)
+                file.write(data_string)
+
+        for number in range(10):
+            today_dirname = datetime.today().strftime("%Y-%m-%d")
+            test_file_name = f"load_data_{number}.dat"
+            blob_name = f"{today_dirname}/{test_file_name}"
+
+            with open(
+                f"{os.path.dirname(os.path.realpath(__file__))}/../fixtures/load_data_{number}.dat"
+            ) as test_file:
+                BlobStorage().add(blob_name, test_file.read())
+
+            Command().handle(**{"date_str": today_dirname})

scripts/python/load_test/load_test_data.dat

@@ -0,0 +1,4 @@
+"NBSSAPPT_HDR"|"88888888"|"20250101"|"170922"|"000003"
+"NBSSAPPT_FLDS"|"Sequence"|"BSO"|"Action"|"Clinic Code"|"Holding Clinic"|"Status"|"Attended Not Scr"|"Appointment ID"|"NHS Num"|"Epsiode Type"|"Episode Start"|"BatchID"|"Screen or Asses"|"Screen Appt num"|"Booked By"|"Cancelled By"|"Appt Date"|"Appt Time"|"Location"|"Clinic Name"|"Clinic Name (Let)"|"Clinic Address 1"|"Clinic Address 2"|"Clinic Address 3"|"Clinic Address 4"|"Clinic Address 5"|"Postcode"|"Action Timestamp"
+"NBSSAPPT_DATA"|"000001"|"SM0K3"|"B"|"SM0K3"|"N"|"B"|"N"|"SM0K3-0000000000"|"9449306621"|"F"|"20250101"|"SM0K0000"|"S"|"1"|"H"|""|"20250101"|"1445"|"SMOK"|"SMOKE TEST CLINIC"|"SMOKE TEST CLINIC"|"SMOKE TEST UNIT"|"NOT A REAL HOSPITAL"|"SMOKE TEST LANE"|"LONDON"|"E5 0AB"|"E5 0AB"|"20250101-154004"
+"NBSSAPPT_END"|"00000013"|"20250101"|"17:09:22"|"000003"

scripts/python/load_test/notifications_load_test.py

@@ -0,0 +1,71 @@
+import logging
+import os
+import subprocess
+import time
+from datetime import datetime
+
+from scripts.python.smoke_test.notifications_smoke_test import (
+    mesh_client,
+    populate_mesh_env_vars,
+)
+
+WORK_DIR = os.path.dirname(os.path.realpath(__file__))
+STARTUP_SCRIPT_PATH = f"{WORK_DIR}/../../bash/run_container_app_job.sh"
+NUMBER_OF_ROWS = os.getenv("NUMBER_OF_ROWS", "1000")
+NUMBER_OF_FILES = os.getenv("NUMBER_OF_ROWS", "1000")
+
+
+def test_load():
+    logging.info("Running notifications load test")
+
+    environment = os.getenv("ENVIRONMENT")
+    resource_group_name = f"rg-manbrs-{environment}-container-app-uks"
+
+    if environment == "prod":
+        return
+
+    add_load_test_data_to_mesh(environment, resource_group_name)
+
+    for job in ["smm", "cap"]:
+        logging.info(
+            "Starting notifications container app job manbrs-%s-%s", job, environment
+        )
+
+        job_result = subprocess.run(
+            [STARTUP_SCRIPT_PATH, environment, job],
+            check=True,
+            capture_output=True,
+            text=True,
+        )
+        assert job_result.returncode == 0
+
+    logging.info("Finished notifications load test")
+
+
+def add_load_test_data_to_mesh(environment: str, resource_group_name: str):
+    populate_mesh_env_vars(environment, resource_group_name)
+    for number in range(int(NUMBER_OF_FILES)):
+        mesh_client().send_message(
+            os.getenv("NBSS_MESH_INBOX_NAME"),
+            generate_load_test_data(number).encode("ASCII"),
+            subject="Load test data",
+        )
+
+
+def generate_load_test_data(file_sequence_number: int) -> str:
+    data = open(f"{WORK_DIR}/load_test_data.dat").read()
+    data = data.replace("20250101", datetime.now().strftime("%Y%m%d"))
+    data = data.replace("88888888", f"{file_sequence_number:06d}")
+    data = data.replace("SM0K3-0000000000", f"SM0K3-{time.time_ns()}")
+
+    rows = data.split("\n")
+    _header = rows[0]
+    _footer = rows[3]
+    first_data_row = rows[2]
+
+    for number in range(2, int(NUMBER_OF_ROWS)):
+        new_data = first_data_row.replace("000001", f"{number:06d}")
+        new_data = new_data.replace("SM0K3-0000000000", f"SM0K3-{time.time_ns()}")
+        rows.insert((number + 1), new_data)
+
+    return "\n".join(rows)