Implement canary file generation functionality from contract test inp…

…uts files

src/rpdk/core/generate.py

@@ -4,6 +4,7 @@
 """
 import logging
 
+from .generate_canary import CanaryFileGenerator
 from .project import Project
 
 LOG = logging.getLogger(__name__)
@@ -20,7 +21,8 @@ def generate(args):
         args.profile,
     )
     project.generate_docs()
-
+    canary_file_generator = CanaryFileGenerator()
+    canary_file_generator.generate_canary_files()
     LOG.warning("Generated files for %s", project.type_name)
 
 

src/rpdk/core/generate_canary.py

@@ -0,0 +1,254 @@
+import json
+import logging
+import re
+import shutil
+from pathlib import Path
+from typing import Any, Dict
+from uuid import uuid4
+
+import yaml
+
+TARGET_SCHEMAS_FOLDER = "target-schemas"
+TARGET_CANARY_ROOT_FOLDER = "canary-bundle"
+TARGET_CANARY_FOLDER = "canary-bundle/canary"
+CONTRACT_TEST_FOLDER = "contract-tests-artifacts"
+RPDK_CONFIG_FILE = ".rpdk-config"
+CANARY_COUNT = "canary_count"
+CANARY_FILE_PREFIX = "canary"
+CONTRACT_TEST_INPUT_PREFIX = "inputs_*"
+CONTRACT_TEST_DEPENDENCY_FILE_NAME = "dependencies.yml"
+CANARY_DEPENDENCY_FILE_NAME = "bootstrap.yaml"
+FILE_GENERATION_ENABLED = "file_generation_enabled"
+CANARY_SETTINGS = "canarySettings"
+TYPE_NAME = "typeName"
+CONTRACT_TEST_FILE_NAMES = "contract_test_file_names"
+INPUT1_FILE_NAME = "inputs_1.json"
+FN_SUB = "Fn::Sub"
+FN_IMPORT_VALUE = "Fn::ImportValue"
+UUID = "uuid"
+DYNAMIC_VALUES_MAP = {
+    "region": "${AWS::Region}",
+    "partition": "${AWS::Partition}",
+    "account": "${AWS::AccountId}",
+}
+
+
+class CanaryFileGenerator:
+    def __init__(self, root=None):
+        self.root = Path(root) if root else Path.cwd()
+        self.load_rpdk_config()
+
+    def load_rpdk_config(self) -> None:
+        """
+        Load the RPDK configuration file.
+
+        This method loads the RPDK configuration file from the current working directory.
+        If the file is not found, it logs a warning and returns an empty dictionary.
+        """
+        config_file_path = self.rpdk_config
+
+        if not config_file_path.is_file():
+            logging.warning("RPDK configuration file not found")
+            self.canary_settings = {}
+            return
+
+        with open(config_file_path, "r", encoding="utf-8") as f:
+            rpdk_config = json.load(f)
+            self.canary_settings = rpdk_config.get(CANARY_SETTINGS, {})
+            self.type_name = rpdk_config.get(TYPE_NAME, "")
+            self.contract_test_file_names = self.canary_settings.get(
+                CONTRACT_TEST_FILE_NAMES, [INPUT1_FILE_NAME]
+            )
+
+    def generate_canary_files(self) -> None:
+        """
+        Generate canary files based on the contract test input files.
+
+        This method checks if file generation is enabled and if the target contract test folder exists.
+        If both conditions are met, it creates the canary folder, copies the contract test dependencies,
+        and generates canary files for each contract test input file up to the specified count.
+        """
+        if (
+            not self.is_file_generation_enabled()
+            or not self.contract_test_folder_exists()
+        ):
+            return
+
+        self._setup_canary_environment()
+        self._generate_canary_files()
+
+    def is_file_generation_enabled(self) -> bool:
+        return self.canary_settings.get(FILE_GENERATION_ENABLED, False)
+
+    def contract_test_folder_exists(self) -> bool:
+        return Path(self.target_contract_test_folder_path).exists()
+
+    def _setup_canary_environment(self) -> None:
+        canary_root = Path(self.target_canary_root_path)
+        canary_folder = Path(self.target_canary_folder_path)
+        self.clean_and_create_canary_folder(canary_root, canary_folder)
+        self.create_canary_bootstrap(
+            Path(self.target_contract_test_folder_path), canary_root
+        )
+
+    def _generate_canary_files(self) -> None:
+        resource_name = self.type_name
+        canary_file_name_prefix = CANARY_FILE_PREFIX
+        canary_folder = Path(self.target_canary_folder_path)
+        contract_test_files = self._get_sorted_contract_test_files()
+        for count, ct_file in enumerate(contract_test_files, start=1):
+            self.create_canary_file(
+                resource_name, ct_file, canary_folder, canary_file_name_prefix, count
+            )
+
+    def _get_sorted_contract_test_files(self) -> list:
+        contract_test_folder = Path(self.target_contract_test_folder_path)
+        contract_test_files = [
+            file
+            for file in contract_test_folder.glob(CONTRACT_TEST_INPUT_PREFIX)
+            if file.is_file() and file.name in self.contract_test_file_names
+        ]
+        return sorted(contract_test_files)
+
+    def clean_and_create_canary_folder(
+        self, canary_root: Path, canary_folder: Path
+    ) -> None:
+        """
+        Clean and create the canary folder.
+
+        This method removes the existing canary root folder and creates a new canary folder.
+
+        Args:
+            canary_root (Path): The path to the canary root folder.
+            canary_folder (Path): The path to the canary folder.
+        """
+        shutil.rmtree(canary_root, ignore_errors=True)
+        canary_folder.mkdir(parents=True, exist_ok=True)
+
+    def create_canary_bootstrap(self, file_location: Path, canary_root: Path) -> None:
+        """
+        Copy the contract test dependencies to the canary root folder.
+
+        This method copies the contract test dependency file to the canary root folder
+        as the canary bootstrap file.
+
+        Args:
+            file_location (Path): The path to the contract test folder.
+            canary_root (Path): The path to the canary root folder.
+        """
+        dependencies_file = file_location / CONTRACT_TEST_DEPENDENCY_FILE_NAME
+        bootstrap_file = canary_root / CANARY_DEPENDENCY_FILE_NAME
+
+        if dependencies_file.exists():
+            shutil.copy(dependencies_file, bootstrap_file)
+
+    def create_canary_file(
+        self,
+        resource_type: str,
+        ct_file: Path,
+        canary_folder: Path,
+        canary_file_name_prefix: str,
+        count: int,
+    ) -> None:
+        """
+        Create a canary file based on the contract test input file.
+
+        This method generates a canary file in YAML format based on the provided contract test input file.
+        The canary file contains the resource configuration with dynamic values replaced.
+
+        Args:
+            resource_type (str): The type of the resource being tested.
+            ct_file (Path): The path to the contract test input file.
+            canary_folder (Path): The path to the canary folder.
+            canary_file_name_prefix (str): The prefix for the canary file name.
+            count (int): The count of the canary file being generated.
+        """
+        with ct_file.open("r") as f:
+            json_data = json.load(f)
+        resource_name = resource_type.split("::")[2]
+        canary_data = {
+            "Description": f"Canary template for {resource_type}",
+            "Resources": {
+                f"{resource_name}Canary": {
+                    "Type": resource_type,
+                    "Properties": self.replace_dynamic_values(
+                        json_data["CreateInputs"]
+                    ),
+                }
+            },
+        }
+        canary_file_name = f"{canary_file_name_prefix}{count}_001.yaml"
+        canary_file_path = canary_folder / canary_file_name
+
+        with canary_file_path.open("w") as canary_file:
+            yaml.dump(canary_data, canary_file, indent=2)
+
+    def replace_dynamic_values(self, properties: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Replace dynamic values in the resource properties.
+
+        This method recursively replaces dynamic values in the resource properties dictionary.
+        It handles nested dictionaries, lists, and strings with dynamic value placeholders.
+
+        Args:
+            properties (Dict[str, Any]): The resource properties dictionary.
+
+        Returns:
+            Dict[str, Any]: The resource properties dictionary with dynamic values replaced.
+        """
+        for key, value in properties.items():
+            if isinstance(value, dict):
+                properties[key] = self.replace_dynamic_values(value)
+            elif isinstance(value, list):
+                properties[key] = [self._replace_dynamic_value(item) for item in value]
+            else:
+                return_value = self._replace_dynamic_value(value)
+                properties[key] = return_value
+        return properties
+
+    def _replace_dynamic_value(self, original_value: Any) -> Any:
+        """
+        Replace a dynamic value with its corresponding value.
+
+        This method replaces dynamic value placeholders in a string with their corresponding values.
+        It handles UUID generation, partition replacement, and Fn::ImportValue function.
+
+        Args:
+            original_value (Any): The value to be replaced.
+
+        Returns:
+            Any: The replaced value.
+        """
+        pattern = r"\{\{(.*?)\}\}"
+
+        def replace_token(match):
+            token = match.group(1)
+            if UUID in token:
+                return str(uuid4())
+            if token in DYNAMIC_VALUES_MAP:
+                return DYNAMIC_VALUES_MAP[token]
+            return f'{{"{FN_IMPORT_VALUE}": "{token.strip()}"}}'
+
+        replaced_value = re.sub(pattern, replace_token, str(original_value))
+
+        if any(value in replaced_value for value in DYNAMIC_VALUES_MAP.values()):
+            replaced_value = {FN_SUB: replaced_value}
+        if FN_IMPORT_VALUE in replaced_value:
+            replaced_value = json.loads(replaced_value)
+        return replaced_value
+
+    @property
+    def target_canary_root_path(self):
+        return self.root / TARGET_CANARY_ROOT_FOLDER
+
+    @property
+    def target_canary_folder_path(self):
+        return self.root / TARGET_CANARY_FOLDER
+
+    @property
+    def target_contract_test_folder_path(self):
+        return self.root / CONTRACT_TEST_FOLDER
+
+    @property
+    def rpdk_config(self):
+        return self.root / RPDK_CONFIG_FILE

src/rpdk/core/project.py

@@ -29,6 +29,11 @@
     SpecValidationError,
 )
 from .fragment.module_fragment_reader import _get_fragment_file
+from .generate_canary import (
+    CONTRACT_TEST_FILE_NAMES,
+    FILE_GENERATION_ENABLED,
+    INPUT1_FILE_NAME,
+)
 from .jsonutils.pointer import fragment_decode, fragment_encode
 from .jsonutils.utils import traverse
 from .plugin_registry import load_plugin
@@ -145,6 +150,7 @@ def __init__(self, overwrite_enabled=False, root=None):
         self.test_entrypoint = None
         self.executable_entrypoint = None
         self.fragment_dir = None
+        self.canary_settings = {}
         self.target_info = {}
 
         self.env = Environment(
@@ -277,6 +283,7 @@ def validate_and_load_resource_settings(self, raw_settings):
         self.executable_entrypoint = raw_settings.get("executableEntrypoint")
         self._plugin = load_plugin(raw_settings["language"])
         self.settings = raw_settings.get("settings", {})
+        self.canary_settings = raw_settings.get("canarySettings", {})
 
     def _write_example_schema(self):
         self.schema = resource_json(
@@ -338,6 +345,7 @@ def _write_resource_settings(f):
                     "testEntrypoint": self.test_entrypoint,
                     "settings": self.settings,
                     **executable_entrypoint_dict,
+                    "canarySettings": self.canary_settings,
                 },
                 f,
                 indent=4,
@@ -391,6 +399,10 @@ def init(self, type_name, language, settings=None):
         self.language = language
         self._plugin = load_plugin(language)
         self.settings = settings or {}
+        self.canary_settings = {
+            FILE_GENERATION_ENABLED: True,
+            CONTRACT_TEST_FILE_NAMES: [INPUT1_FILE_NAME],
+        }
         self._write_example_schema()
         self._write_example_inputs()
         self._plugin.init(self)