Implement canary file generation functionality from contract test inp… (

#1069)

src/rpdk/core/generate.py

@@ -20,7 +20,7 @@ def generate(args):
  args.profile,
  )
  project.generate_docs()
-
+ project.generate_canary_files()
  LOG.warning("Generated files for %s", project.type_name)
 
 

src/rpdk/core/project.py

@@ -2,13 +2,16 @@
 import json
 import logging
 import os
+import re
 import shutil
 import sys
 import zipfile
 from pathlib import Path
 from tempfile import TemporaryFile
+from typing import Any, Dict
 from uuid import uuid4
 
+import yaml
 from botocore.exceptions import ClientError, WaiterError
 from jinja2 import Environment, PackageLoader, select_autoescape
 from jsonschema import Draft7Validator
@@ -56,7 +59,32 @@
 ARTIFACT_TYPE_RESOURCE = "RESOURCE"
 ARTIFACT_TYPE_MODULE = "MODULE"
 ARTIFACT_TYPE_HOOK = "HOOK"
-
+TARGET_CANARY_ROOT_FOLDER = "canary-bundle"
+TARGET_CANARY_FOLDER = "canary-bundle/canary"
+RPDK_CONFIG_FILE = ".rpdk-config"
+CANARY_FILE_PREFIX = "canary"
+CONTRACT_TEST_DEPENDENCY_FILE_NAME = "dependencies.yml"
+CANARY_DEPENDENCY_FILE_NAME = "bootstrap.yaml"
+CANARY_SETTINGS = "canarySettings"
+TYPE_NAME = "typeName"
+CONTRACT_TEST_FILE_NAMES = "contract_test_file_names"
+INPUT1_FILE_NAME = "inputs_1.json"
+FILE_GENERATION_ENABLED = "file_generation_enabled"
+CONTRACT_TEST_FOLDER = "contract-tests-artifacts"
+CONTRACT_TEST_INPUT_PREFIX = "inputs_*"
+CONTRACT_TEST_DEPENDENCY_FILE_NAME = "dependencies.yml"
+FILE_GENERATION_ENABLED = "file_generation_enabled"
+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}",
+}
 DEFAULT_ROLE_TIMEOUT_MINUTES = 120 # 2 hours
 # min and max are according to CreateRole API restrictions
 # https://docs.aws.amazon.com/IAM/latest/APIReference/API_CreateRole.html
@@ -145,6 +173,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(
@@ -207,6 +236,30 @@ def target_schemas_path(self):
  def target_info_path(self):
  return self.root / TARGET_INFO_FILENAME
 
+ @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 rpdk_config(self):
+ return self.root / RPDK_CONFIG_FILE
+
+ @property
+ def file_generation_enabled(self):
+ return self.canary_settings.get(FILE_GENERATION_ENABLED, False)
+
+ @property
+ def contract_test_file_names(self):
+ return self.canary_settings.get(CONTRACT_TEST_FILE_NAMES, [INPUT1_FILE_NAME])
+
+ @property
+ def target_contract_test_folder_path(self):
+ return self.root / CONTRACT_TEST_FOLDER
+
  @staticmethod
  def _raise_invalid_project(msg, e):
  LOG.debug(msg, exc_info=e)
@@ -277,6 +330,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 +392,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 +446,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)
@@ -1251,3 +1310,83 @@ def _load_target_info(
  )
 
  return type_info
+
+ def generate_canary_files(self) -> None:
+ if (
+ not self.file_generation_enabled
+ or not Path(self.target_contract_test_folder_path).exists()
+ ):
+ return
+ self._setup_stack_template_environment()
+ self._generate_stack_template_files()
+
+ def _setup_stack_template_environment(self) -> None:
+ stack_template_root = Path(self.target_canary_root_path)
+ stack_template_folder = Path(self.target_canary_folder_path)
+ stack_template_folder.mkdir(parents=True, exist_ok=True)
+ dependencies_file = (
+ Path(self.target_contract_test_folder_path)
+ / CONTRACT_TEST_DEPENDENCY_FILE_NAME
+ )
+ bootstrap_file = stack_template_root / CANARY_DEPENDENCY_FILE_NAME
+ if dependencies_file.exists():
+ shutil.copy(str(dependencies_file), str(bootstrap_file))
+
+ def _generate_stack_template_files(self) -> None:
+ stack_template_folder = Path(self.target_canary_folder_path)
+ 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
+ ]
+ contract_test_files = sorted(contract_test_files)
+ for count, ct_file in enumerate(contract_test_files, start=1):
+ with ct_file.open("r") as f:
+ json_data = json.load(f)
+ resource_name = self.type_info[2]
+ stack_template_data = {
+ "Description": f"Template for {self.type_name}",
+ "Resources": {
+ f"{resource_name}": {
+ "Type": self.type_name,
+ "Properties": self._replace_dynamic_values(
+ json_data["CreateInputs"]
+ ),
+ }
+ },
+ }
+ stack_template_file_name = f"{CANARY_FILE_PREFIX}{count}_001.yaml"
+ stack_template_file_path = stack_template_folder / stack_template_file_name
+ with stack_template_file_path.open("w") as stack_template_file:
+ yaml.dump(stack_template_data, stack_template_file, indent=2)
+
+ def _replace_dynamic_values(self, properties: Dict[str, Any]) -> Dict[str, Any]:
+ 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:
+ 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