Implement canary file generation functionality from contract test PatchInputs

setup.cfg

@@ -29,7 +29,7 @@ include_trailing_comma = true
 combine_as_imports = True
 force_grid_wrap = 0
 known_first_party = rpdk
-known_third_party = boto3,botocore,cfn_tools,cfnlint,colorama,docker,hypothesis,jinja2,jsonschema,nested_lookup,ordered_set,pkg_resources,pytest,pytest_localserver,requests,setuptools,yaml
+known_third_party = boto3,botocore,cfn_tools,cfnlint,colorama,docker,hypothesis,jinja2,jsonpatch,jsonschema,nested_lookup,ordered_set,pkg_resources,pytest,pytest_localserver,requests,setuptools,yaml
 
 [tool:pytest]
 # can't do anything about 3rd part modules, so don't spam us

setup.py

@@ -41,6 +41,7 @@ def find_version(*file_paths):
  "boto3>=1.10.20",
  "Jinja2>=3.1.2",
  "markupsafe>=2.1.0",
+ "jsonpatch",
  "jsonschema>=3.0.0,<=4.17.3",
  "pytest>=4.5.0",
  "pytest-random-order>=1.0.4",

src/rpdk/core/project.py

@@ -11,6 +11,7 @@
 from typing import Any, Dict
 from uuid import uuid4
 
+import jsonpatch
 import yaml
 from botocore.exceptions import ClientError, WaiterError
 from jinja2 import Environment, PackageLoader, select_autoescape
@@ -63,6 +64,13 @@
 TARGET_CANARY_FOLDER = "canary-bundle/canary"
 RPDK_CONFIG_FILE = ".rpdk-config"
 CANARY_FILE_PREFIX = "canary"
+CANARY_FILE_CREATE_SUFFIX = "001"
+CANARY_FILE_UPDATE_SUFFIX = "002"
+CANARY_SUPPORTED_PATCH_INPUT_OPERATIONS = {"replace", "remove", "add"}
+CREATE_INPUTS_KEY = "CreateInputs"
+PATCH_INPUTS_KEY = "PatchInputs"
+PATCH_VALUE_KEY = "value"
+PATCH_OPERATION_KEY = "op"
 CONTRACT_TEST_DEPENDENCY_FILE_NAME = "dependencies.yml"
 CANARY_DEPENDENCY_FILE_NAME = "bootstrap.yaml"
 CANARY_SETTINGS = "canarySettings"
@@ -76,7 +84,6 @@
 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"
@@ -1345,21 +1352,68 @@ def _generate_stack_template_files(self) -> None:
  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)
+
+ self._save_stack_template_data(
+ resource_name,
+ count,
+ stack_template_folder,
+ self._replace_dynamic_values(
+ json_data[CREATE_INPUTS_KEY],
+ ),
+ CANARY_FILE_CREATE_SUFFIX,
+ )
+ if PATCH_INPUTS_KEY in json_data:
+ supported_patch_inputs = self._translate_supported_patch_inputs(
+ json_data[PATCH_INPUTS_KEY]
+ )
+ patch_data = jsonpatch.apply_patch(
+ json_data[CREATE_INPUTS_KEY], supported_patch_inputs, in_place=False
+ )
+ self._save_stack_template_data(
+ resource_name,
+ count,
+ stack_template_folder,
+ patch_data,
+ CANARY_FILE_UPDATE_SUFFIX,
+ )
+
+ def _save_stack_template_data(
+ self,
+ resource_name,
+ contract_test_input_count,
+ stack_template_folder,
+ properties_data,
+ suffix,
+ ):
+ stack_template_data = {
+ "Description": f"Template for {self.type_name}",
+ "Resources": {
+ f"{resource_name}": {
+ "Type": self.type_name,
+ "Properties": properties_data,
+ }
+ },
+ }
+ stack_template_file_name = (
+ f"{CANARY_FILE_PREFIX}{contract_test_input_count}_{suffix}.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 _translate_supported_patch_inputs(self, patch_inputs: Any) -> Any:
+ output = []
+ for patch_input in patch_inputs:
+ if (
+ patch_input.get(PATCH_OPERATION_KEY)
+ in CANARY_SUPPORTED_PATCH_INPUT_OPERATIONS
+ ):
+ if PATCH_VALUE_KEY in patch_input:
+ self._replace_dynamic_values_with_root_key(
+ patch_input, PATCH_VALUE_KEY
+ )
+ output.append(patch_input)
+ return output
 
  def _replace_dynamic_values(self, properties: Dict[str, Any]) -> Dict[str, Any]:
  for key, value in properties.items():
@@ -1372,6 +1426,19 @@ def _replace_dynamic_values(self, properties: Dict[str, Any]) -> Dict[str, Any]:
  properties[key] = return_value
  return properties
 
+ def _replace_dynamic_values_with_root_key(
+ self, properties: Dict[str, Any], root_key=None
+ ) -> Dict[str, Any]:
+ value = properties[root_key]
+ if isinstance(value, dict):
+ properties[root_key] = self._replace_dynamic_values(value)
+ elif isinstance(value, list):
+ properties[root_key] = [self._replace_dynamic_value(item) for item in value]
+ else:
+ return_value = self._replace_dynamic_value(value)
+ properties[root_key] = return_value
+ return properties
+
  def _replace_dynamic_value(self, original_value: Any) -> Any:
  pattern = r"\{\{(.*?)\}\}"