Unit test now passes

Author: robbailiff2

pyproject.toml

@@ -35,6 +35,7 @@ python-json-logger = "^3.3.0"
 fhir-resources = "^8.0.0"
 python-dateutil = "^2.9.0"
 pyhamcrest = "^2.1.0"
+moto = "^5.1.5"
 
 [tool.poetry.group.dev.dependencies]
 ruff = "^0.11.0"

scripts/manual_uploads/manual_s3_dynamo_upload.py

@@ -4,6 +4,9 @@
 import os
 import argparse
 from pathlib import Path
+from typing import Any, Dict, List, Optional
+from decimal import Decimal
+
 
 
 def map_dynamo_type(value: Any):
@@ -28,7 +31,8 @@ def map_dynamo_type(value: Any):
 
 def upload_to_s3(s3_client, bucket, filepath, dry_run=False):
     filename = os.path.basename(filepath)
-    s3_key = f"manual-uploads/{filename}.json"
+    print(f"Filepath: {filepath}")
+    s3_key = f"manual-uploads/{filename}"
 
     if dry_run:
         print(f"[DRY RUN] Would upload {filepath} to s3://{bucket}/{s3_key}")
@@ -54,7 +58,8 @@ def upload_to_dynamo(dynamo_client, table_name, filepath):
         print(f"Failed to upload {filepath}: {e}")
 
 
-def main():
+def run_upload(args=None):
+    print("\n\n\n***** We are in main *****\n\n\n")
     parser = argparse.ArgumentParser()
     parser.add_argument("--env")
     parser.add_argument("--upload-s3", type=Path)
@@ -63,25 +68,29 @@ def main():
     parser.add_argument("--s3-bucket")
     parser.add_argument("--dynamo-table")
     parser.add_argument("--dry-run", action="store_true")
-    args = parser.parse_args()
 
-    if not args.s3_bucket:
-        args.s3_bucket = f"eligibility-signposting-api-{args.env}-eli-rules"
-    if not args.dynamo_table:
-        args.dynamo_table = f"eligibility-signposting-api-{args.env}-eligibility_datastore"
+    if args is None:
+        parsed_args = parser.parse_args()
+    else:
+        parsed_args = parser.parse_args(args)
+
+    if not parsed_args.s3_bucket:
+        parsed_args.s3_bucket = f"eligibility-signposting-api-{parsed_args.env}-eli-rules"
+    if not parsed_args.dynamo_table:
+        parsed_args.dynamo_table = f"eligibility-signposting-api-{parsed_args.env}-eligibility_datastore"
 
     session = boto3.Session()
-    s3 = session.client("s3", region_name=args.region)
-    dynamo = session.client("dynamodb", region_name=args.region)
+    s3 = session.client("s3", region_name=parsed_args.region)
+    dynamo = session.client("dynamodb", region_name=parsed_args.region)
 
-    if args.upload_s3:
-        for filepath in args.upload_s3.glob("*.json"):
-            upload_to_s3(s3, args.s3_bucket, str(filepath), args.dry_run)
+    if parsed_args.upload_s3:
+        for filepath in parsed_args.upload_s3.glob("*.json"):
+            upload_to_s3(s3, parsed_args.s3_bucket, str(filepath), parsed_args.dry_run)
 
-    if args.upload_dynamo:
-        for filepath in args.upload_dynamo.glob("*.json"):
-            upload_to_dynamo(dynamo, args.dynamo_table, str(filepath))
+    if parsed_args.upload_dynamo:
+        for filepath in parsed_args.upload_dynamo.glob("*.json"):
+            upload_to_dynamo(dynamo, parsed_args.dynamo_table, str(filepath))
 
 
 if __name__ == "__main__":
-    main()
+    run_upload()

tests/utils/manual_s3_dynamo_upload_tests.py

@@ -1,11 +1,14 @@
+import os
 import json
 import tempfile
 import subprocess
-import os
 from pathlib import Path
 import pytest
-from moto import mock_s3, mock_dynamodb
+
 import boto3
+from moto import mock_aws
+
+from scripts.manual_uploads.manual_s3_dynamo_upload import run_upload
 
 
 @pytest.fixture
@@ -17,9 +20,8 @@ def test_data_dir(tmp_path):
     return tmp_path, data
 
 
-@mock_s3
-@mock_dynamodb
-def test_script_cli_end_to_end(test_data_dir):
+@mock_aws
+def test_script_cli_end_to_end(test_data_dir, capsys):
     # Arrange
     data_dir, expected_data = test_data_dir
     env = "test"
@@ -28,7 +30,10 @@ def test_script_cli_end_to_end(test_data_dir):
     dynamo_table = f"api-{env}-datastore"
 
     s3 = boto3.client("s3", region_name=region)
-    s3.create_bucket(Bucket=s3_bucket)
+    s3.create_bucket(
+        Bucket=s3_bucket,
+        CreateBucketConfiguration={"LocationConstraint": region}
+    )
 
     dynamodb = boto3.client("dynamodb", region_name=region)
     dynamodb.create_table(
@@ -45,20 +50,18 @@ def test_script_cli_end_to_end(test_data_dir):
     )
 
     # Act
-    result = subprocess.run(
-        [
-            "python", "your_script.py",
-            "--env", env,
-            "--upload-s3", str(data_dir),
-            "--upload-dynamo", str(data_dir),
-            "--region", region
-        ],
-        capture_output=True,
-        text=True
-    )
+    return_code = run_upload([
+        "--env", env,
+        "--upload-s3", str(data_dir),
+        "--upload-dynamo", str(data_dir),
+        "--region", region,
+        "--s3-bucket", s3_bucket,
+        "--dynamo-table", dynamo_table
+    ])
+    captured = capsys.readouterr()
 
     # Assert
-    key = f"manual-uploads/test.json.json"
+    key = f"manual-uploads/test.json"
     obj = s3.get_object(Bucket=s3_bucket, Key=key)
     uploaded_s3_data = json.load(obj["Body"])
     assert uploaded_s3_data == expected_data
@@ -72,5 +75,5 @@ def test_script_cli_end_to_end(test_data_dir):
     )["Item"]
     assert item["value"]["N"] == "99"
 
-    assert result.returncode == 0
-    assert "Uploaded" in result.stdout
+    assert "Uploaded" in captured.out
+    assert "Error" not in captured.err