+generate DynamoDB data generate_dynamo_data.py

Author: ivma1-nhs

tests/e2e/data/dynamoDB/AUTO_RSV_SB_006.json

@@ -39,7 +39,7 @@
     {
       "NHS_NUMBER": "5000000006",
       "ATTRIBUTE_TYPE": "RSV",
-      "LAST_SUCCESSFUL_DATE": "<<DATE_LAST_WEEK>>"
+      "LAST_SUCCESSFUL_DATE": "<<DATE_DAY_-7>>"
     }
   ]
 }

tests/e2e/data/dynamoDB/AUTO_RSV_SB_007.json

@@ -39,7 +39,7 @@
     {
       "NHS_NUMBER": "5000000007",
       "ATTRIBUTE_TYPE": "RSV",
-      "LAST_SUCCESSFUL_DATE": "<<DATE_LAST_WEEK>>"
+      "LAST_SUCCESSFUL_DATE": "<<DATE_DAY_-7>>"
     }
   ]
 }

tests/e2e/data/dynamoDB/AUTO_RSV_SB_012.json

@@ -29,7 +29,7 @@
     {
       "NHS_NUMBER": "5000000012",
       "ATTRIBUTE_TYPE": "RSV",
-      "LAST_SUCCESSFUL_DATE": "<<DATE_LAST_WEEK>>"
+      "LAST_SUCCESSFUL_DATE": "<<DATE_DAY_-7>>"
     }
   ]
 }

tests/e2e/data/dynamoDB/AUTO_RSV_SB_013.json

@@ -39,7 +39,7 @@
     {
       "NHS_NUMBER": "50000000014",
       "ATTRIBUTE_TYPE": "RSV",
-      "LAST_SUCCESSFUL_DATE": "<<DATE_LAST_WEEK>>"
+      "LAST_SUCCESSFUL_DATE": "<<DATE_DAY_-7>>"
     }
   ]
 }

tests/e2e/data/generate_dynamo_data.py

@@ -0,0 +1,126 @@
+import json
+import logging
+import os
+import re
+from datetime import datetime, timedelta
+from pathlib import Path
+from typing import Any
+
+# Constants
+OUTPUT_ROOT = "out"
+DATE_FORMAT = "%Y%m%d"
+VAR_PATTERN = re.compile(r"<<([^<>]+)>>")
+
+# Configure logging
+logging.basicConfig(level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s")
+
+
+class DateVariableResolver:
+    """Handles the logic for parsing and evaluating date-based variables."""
+
+    def __init__(self, today: datetime = None):
+        self.today = today or datetime.today()
+
+    def resolve(self, token: str) -> str:
+        logging.debug(f"Resolving variable: {token}")
+        parts = token.split("_")
+        if len(parts) < 3 or parts[0].upper() != "DATE":
+            raise ValueError(f"Unsupported variable format: {token}")
+
+        _, unit, value = parts[0], parts[1].lower(), parts[2]
+
+        try:
+            offset = int(value)
+        except ValueError:
+            raise ValueError(f"Invalid offset value: {value}")
+
+        if unit == "day":
+            return (self.today + timedelta(days=offset)).strftime(DATE_FORMAT)
+        if unit == "week":
+            return (self.today + timedelta(weeks=offset)).strftime(DATE_FORMAT)
+        if unit == "year":
+            return (self.today.replace(year=self.today.year + offset)).strftime(DATE_FORMAT)
+        if unit == "age":
+            try:
+                birth_date = self.today.replace(year=self.today.year - offset)
+            except ValueError:
+                # Handle February 29th
+                birth_date = self.today.replace(month=2, day=28, year=self.today.year - offset)
+            return birth_date.strftime(DATE_FORMAT)
+        raise ValueError(f"Unsupported calculation unit: {unit}")
+
+
+class JsonTestDataProcessor:
+    """Processes JSON test files by resolving placeholders in 'data' arrays."""
+
+    def __init__(self, input_dir: Path, output_dir: Path, resolver: DateVariableResolver):
+        self.input_dir = input_dir
+        self.output_dir = output_dir
+        self.resolver = resolver
+
+    def resolve_placeholders(self, obj: Any) -> Any:
+        if isinstance(obj, dict):
+            return {k: self.resolve_placeholders(v) for k, v in obj.items()}
+        if isinstance(obj, list):
+            return [self.resolve_placeholders(item) for item in obj]
+        if isinstance(obj, str):
+            return VAR_PATTERN.sub(self._replace_token, obj)
+        return obj
+
+    def _replace_token(self, match: re.Match) -> str:
+        token = match.group(1)
+        try:
+            return self.resolver.resolve(token)
+        except Exception as e:
+            logging.warning(f"Failed to resolve variable {token}: {e}")
+            return match.group(0)
+
+    def process_file(self, file_path: Path):
+        logging.info(f"Processing file: {file_path}")
+        try:
+            with open(file_path) as f:
+                content = json.load(f)
+        except Exception as e:
+            logging.exception(f"Failed to read {file_path}: {e}")
+            return
+
+        try:
+            resolved = self.resolve_placeholders(content)
+        except Exception as e:
+            logging.exception(f"Failed to resolve placeholders: {e}")
+            return
+
+        if "data" not in resolved:
+            logging.error(f"Missing 'data' key in {file_path}")
+            return
+
+        relative_path = file_path.relative_to(self.input_dir)
+        output_path = self.output_dir / relative_path
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+
+        try:
+            with open(output_path, "w") as f:
+                json.dump(resolved["data"], f, indent=2)
+            logging.info(f"Written resolved file: {output_path}")
+        except Exception as e:
+            logging.exception(f"Failed to write output: {e}")
+
+
+def main():
+    input_dir = Path()
+    output_dir = Path(OUTPUT_ROOT)
+    resolver = DateVariableResolver()
+
+    processor = JsonTestDataProcessor(input_dir, output_dir, resolver)
+
+    logging.info(f"Scanning for JSON files in {input_dir}")
+    for root, _, files in os.walk(input_dir):
+        for file in files:
+            if file.endswith(".json"):
+                processor.process_file(Path(root) / file)
+            else:
+                logging.debug(f"Skipping non-JSON file: {file}")
+
+
+if __name__ == "__main__":
+    main()