Finalised manual upload tests

Author: robbailiff2

scripts/manual_uploads/manual_s3_dynamo_upload.py

@@ -43,7 +43,6 @@ def upload_to_s3(
 ) -> None:
 
     filename = os.path.basename(filepath)
-    print(f"Filepath: {filepath}")
     s3_key = f"manual-uploads/{filename}"
 
     if dry_run:
@@ -98,11 +97,23 @@ def run_upload(args: Optional[List[str]] = None) -> None:
     dynamo = session.client("dynamodb", region_name=parsed_args.region)
 
     if parsed_args.upload_s3:
-        for filepath in parsed_args.upload_s3.glob("*.json"):
+        if parsed_args.upload_s3.is_dir():
+            files = parsed_args.upload_s3.glob("*.json")
+        else:
+            files = [parsed_args.upload_s3]
+
+        for filepath in files:
+            print(f"Uploading to S3 from {filepath}")
             upload_to_s3(s3, parsed_args.s3_bucket, str(filepath), parsed_args.dry_run)
 
     if parsed_args.upload_dynamo:
-        for filepath in parsed_args.upload_dynamo.glob("*.json"):
+        if parsed_args.upload_dynamo.is_dir():
+            paths = parsed_args.upload_dynamo.glob("*.json")
+        else:
+            paths = [parsed_args.upload_dynamo]
+
+        for filepath in paths:
+            print(f"Uploading to DynamoDB from {filepath}")
             upload_to_dynamo(dynamo, parsed_args.dynamo_table, str(filepath))
 
 

tests/test_data/test_config/test_config.json

@@ -0,0 +1,56 @@
+{
+    "CampaignConfig": {
+        "ID": "id_1",
+        "Version": 1,
+        "Name": "Test Config",
+        "Type": "V",
+        "Target": "COVID",
+        "Manager": "person@test.com",
+        "Approver": "person@test.com",
+        "Reviewer": "person@test.com",
+        "IterationFrequency": "X",
+        "IterationType": "M",
+        "IterationTime": "07:00:00",
+        "DefaultCommsRouting": "Default_Comms_1",
+        "Iterations": [
+            {
+                "ID": "id_1",
+                "DefaultCommsRouting": "Default_Comms_1",
+                "IterationCohorts": [
+                    {
+                        "CohortLabel": "cohort_1",
+                        "Priority": 10
+					},
+                    {
+                        "CohortLabel": "cohort_2",
+                        "Priority": 20
+					}
+                ],
+                "IterationRules": [
+                    {
+                        "Type": "F",
+                        "Name": "Test Rule",
+                        "Description": "Test Rule Desc",
+                        "Priority": 20,
+                        "AttributeLevel": "PERSON",
+                        "AttributeName": "DATE_OF_BIRTH",
+                        "Operator": ">",
+                        "Comparator": "19000101"
+					}
+                ],
+                "Version": 1,
+                "Name": "Test Config",
+                "Type": "M",
+                "IterationDate": "20250101",
+                "IterationNumber": 1,
+                "CommsType": "R",
+                "ApprovalMinimum": 1,
+                "ApprovalMaximum": 5000000
+            }
+        ],
+        "StartDate": "20250101",
+        "EndDate": "20250101",
+        "ApprovalMinimum": 1,
+        "ApprovalMaximum": 5000000
+    }
+}

tests/test_data/test_records/test_cohort_records.json

@@ -0,0 +1,2 @@
+{"NHS_NUMBER": "1234567890", "ATTRIBUTE_TYPE": "COHORTS", "COHORT_MEMBERSHIPS": [{"COHORT_LABEL": "under_75", "DATE_JOINED": "2025-01-01"},{"COHORT_LABEL": "over_75", "DATE_JOINED": "2025-01-01"}]}
+{"NHS_NUMBER": "2345678901", "ATTRIBUTE_TYPE": "COHORTS", "COHORT_MEMBERSHIPS": [{"COHORT_LABEL": "under_75", "DATE_JOINED": "2025-01-01"},{"COHORT_LABEL": "over_75", "DATE_JOINED": "2025-01-01"}]}

tests/utils/manual_s3_dynamo_upload_tests.py

@@ -78,7 +78,6 @@ def test_script_cli_end_to_end(test_data_dir):
         }
         response = dynamodb.get_item(TableName=dynamo_table, Key=key)
         item = response.get("Item")
-        assert item is not None, f"Missing item for key {key}"
         dynamo_items.append(item)
 
     expected_dynamo_items = [{k: map_dynamo_type(v) for k, v in item.items()} for item in expected_data]