Wip

Author: nhsdevws

e2e_batch/scenarios.json

@@ -0,0 +1,75 @@
+from vax_suppliers import TestPair
+from constants import ActionFlag
+from utils import BusRowResult, InfResult
+from utils import poll_destination
+
+
+class TestAction:
+    def __init__(self, action: ActionFlag, unique_id: str = None,
+                 expected_result: BusRowResult = BusRowResult.SUCCESS):
+        self.action = action
+        self.unique_id = unique_id
+        self.expected_result = expected_result
+
+    def poll_destination(self, pending: bool) -> bool:
+        # loop through keys in test (inf and bus)
+        for ack_key in self.ack_keys.keys():
+            if not self.ack_keys[ack_key]:
+                found_ack_key = poll_destination(self.file_name, ack_key)
+                if found_ack_key:
+                    self.ack_keys[ack_key] = found_ack_key
+                else:
+                    pending = True
+        return pending
+
+
+scenarios = {
+    "dev": [
+        {
+            "name": "Successful Create",
+            "ods_vax": TestPair.E8HA94_COVID19_CUD,
+            "expected_result": InfResult.SUCCESS,
+            "actions": [TestAction(ActionFlag.CREATE)],
+            "description": "Successful Create"
+        },
+        {
+            "name": "Successful Update",
+            "description": "Successful Create,Update",
+            "ods_vax": TestPair.DPSFULL_COVID19_CRUDS,
+            "expected_result": InfResult.SUCCESS,
+            "actions": [TestAction(ActionFlag.CREATE), TestAction(ActionFlag.UPDATE)]
+        },
+        {
+            "name": "Successful Delete",
+            "description": "Successful Create,Update, Delete",
+            "ods_vax": TestPair.V0V8L_FLU_CRUDS,
+            "expected_result": InfResult.SUCCESS,
+            "actions": [TestAction(ActionFlag.CREATE), TestAction(ActionFlag.DELETE_LOGICAL)]
+        },
+        {
+            "name": "Failed Update",
+            "description": "Failed Update - resource does not exist",
+            "ods_vax": TestPair.V0V8L_3IN1_CRUDS,
+            "expected_result": InfResult.FATAL_ERROR,
+            "actions": [TestAction(ActionFlag.UPDATE, expected_result=BusRowResult.FATAL_ERROR)],
+            "operation_outcome": "Immunization resource does not exist"
+        },
+        {
+            "name": "Failed Delete",
+            "description": "Failed Delete - resource does not exist",
+            "ods_vax": TestPair.X26_MMR_CRUDS,
+            "expected_result": InfResult.FATAL_ERROR,
+            "actions": [TestAction(ActionFlag.DELETE_LOGICAL, expected_result=BusRowResult.FATAL_ERROR)],
+            "operation_outcome": "Immunization resource does not exist"
+        },
+        {
+            "name": "Create with 1252 char",
+            "description": "Create with 1252 char",
+            "ods_vax": TestPair.YGA_MENACWY_CRUDS,
+            "expected_result": InfResult.SUCCESS,
+            "actions": [TestAction(ActionFlag.CREATE)],
+            "test_encoding": True
+        }
+      ],
+    "ref": []
+  }

e2e_batch/scenarios.py

@@ -7,23 +7,25 @@
     check_inf_file_content,
     validate_row_count,
     generate_csv_files_good,
-    OpMsgs,
-    TestScenario as Scenario,
+    # OpMsgs,
+    TestCase,
     DestinationType,
-    BusRowResult,
+    # BusRowResult,
     cleanup,
-    check_final_success_action
+    check_final_success_action,
+    TestCases,
 )
-from vax_suppliers import TestPair
+# from vax_suppliers import TestPair
 from clients import logger
+from scenarios import scenarios
 
 
 from constants import (
     SOURCE_BUCKET,
     INPUT_PREFIX,
     ACK_BUCKET,
     environment,
-    TestSet,
+    # TestSet,
 )
 
 NEW = "NEW"
@@ -34,16 +36,7 @@
 class TestE2EBatch(unittest.TestCase):
 
     def setUp(self):
-        self.test_data = [
-            Scenario(TestPair.E8HA94_COVID19_CUD, TestSet.CREATE_OK),
-            Scenario(TestPair.DPSFULL_COVID19_CRUDS, TestSet.UPDATE_OK),
-            Scenario(TestPair.V0V8L_FLU_CRUDS, TestSet.DELETE_OK),
-            Scenario(TestPair.V0V8L_3IN1_CRUDS, TestSet.UPDATE_FAIL, expected=BusRowResult.FATAL_ERROR,
-                     operation_outcome=OpMsgs.IMM_NOT_EXIST),
-            Scenario(TestPair.X26_MMR_CRUDS, TestSet.DELETE_FAIL, expected=BusRowResult.FATAL_ERROR,
-                     operation_outcome=OpMsgs.IMM_NOT_EXIST),
-            Scenario(TestPair.YGA_MENACWY_CRUDS, TestSet.CREATE_OK, inject_char=True, name="Create with 1252 char"),
-        ]
+        self.test_data = TestCases(scenarios["dev"])
 
     def tearDown(self):
         cleanup(self.test_data)
@@ -54,7 +47,7 @@ def test_batch_submission(self):
         start_time = time.time()
         max_timeout = 1200  # seconds
 
-        tests: list[Scenario] = generate_csv_files_good(self.test_data)
+        tests:  list[TestCase] = generate_csv_files_good(self.test_data.test_cases)
 
         send_files(tests)
 
@@ -66,7 +59,7 @@ def test_batch_submission(self):
         logger.info(f"Tests Completed. Time: {time.time() - start_time:.1f} seconds")
 
 
-def send_files(tests: list[Scenario]):
+def send_files(tests: list[TestCase]):
     start_time = time.time()
     for test in tests:
         logger.info(f"Upload for Test: {test.name} ")
@@ -75,7 +68,7 @@ def send_files(tests: list[Scenario]):
     logger.info(f"Files uploaded. Time: {time.time() - start_time:.1f} seconds")
 
 
-def poll_for_responses(tests: list[Scenario], max_timeout=1200) -> bool:
+def poll_for_responses(tests: list[TestCase], max_timeout=1200) -> bool:
     logger.info("Waiting while processing...")
     start_time = time.time()
     # while there are still pending files, poll for acks and forwarded files
@@ -93,7 +86,7 @@ def poll_for_responses(tests: list[Scenario], max_timeout=1200) -> bool:
     return True
 
 
-def validate_responses(tests: list[Scenario]):
+def validate_responses(tests: TestCases):
     start_time = time.time()
     count = 0
     expected_count = len(tests) * 2
@@ -104,7 +97,7 @@ def validate_responses(tests: list[Scenario]):
             if test.ack_keys[DestinationType.INF]:
                 count += 1
                 inf_ack_content = get_file_content_from_s3(ACK_BUCKET, test.ack_keys[DestinationType.INF])
-                check_ack_file_content(test.name, inf_ack_content, "Success", None, test.action_sequence.actions)
+                check_ack_file_content(test.name, inf_ack_content, "Success", None, test.expected_result)
             else:
                 logger.error(f"INF ACK file not found for test: {test.name}")