diff --git a/samcli/commands/build/build_context.py b/samcli/commands/build/build_context.py index 1c6355b030..d3a11ba4bc 100644 --- a/samcli/commands/build/build_context.py +++ b/samcli/commands/build/build_context.py @@ -504,7 +504,13 @@ def _update_original_template_paths(self, original_template: Dict, modified_temp elif isinstance(resource_value, dict) and resource_key in modified_resources: # Regular resource - copy updated paths from modified template modified_resource = modified_resources.get(resource_key, {}) - self._copy_artifact_paths(resource_value, modified_resource) + from samcli.lib.cfn_language_extensions.property_paths import copy_artifact_properties + + copy_artifact_properties( + resource_value.get("Properties", {}), + modified_resource.get("Properties", {}), + resource_value.get("Type", ""), + ) # Merge generated Mappings into the template if all_generated_mappings: @@ -552,13 +558,13 @@ def _update_foreach_artifact_paths( Generated Mappings section for dynamic artifact properties (empty dict if none) """ from samcli.lib.cfn_language_extensions.models import PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES - from samcli.lib.cfn_language_extensions.sam_integration import resolve_collection - from samcli.lib.package.language_extensions_packaging import ( - _get_prop_value, - _leaf_prop_name, - _resolve_property_paths, - _set_prop_value, + from samcli.lib.cfn_language_extensions.property_paths import ( + get_prop_value, + leaf_prop_name, + resolve_property_paths, + set_prop_value, ) + from samcli.lib.cfn_language_extensions.sam_integration import resolve_collection generated_mappings: Dict[str, Dict[str, Dict[str, str]]] = {} @@ -606,15 +612,15 @@ def _update_foreach_artifact_paths( continue candidate_paths = PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES.get(resource_type, []) - for prop_name in _resolve_property_paths(candidate_paths, properties): - prop_value = _get_prop_value(properties, prop_name) + for prop_name in resolve_property_paths(candidate_paths, properties): + prop_value = get_prop_value(properties, prop_name) if prop_value is None: continue # CloudFormation Mapping names and FindInMap third-args must be alphanumeric, # so dotted property names (e.g. "Command.ScriptLocation") use the leaf segment # for those identifiers while the dotted form is preserved for property addressing. - leaf_name = _leaf_prop_name(prop_name) + leaf_name = leaf_prop_name(prop_name) if contains_loop_variable(prop_value, loop_variable) and collection_values: # Determine which outer loop variables the property references @@ -653,7 +659,7 @@ def _update_foreach_artifact_paths( else: lookup_key = {"Ref": loop_variable} - _set_prop_value(properties, prop_name, {"Fn::FindInMap": [mapping_name, lookup_key, leaf_name]}) + set_prop_value(properties, prop_name, {"Fn::FindInMap": [mapping_name, lookup_key, leaf_name]}) else: expanded_key = self._build_expanded_key( resource_template_key, @@ -664,7 +670,7 @@ def _update_foreach_artifact_paths( if expanded_key: artifact_value = self._get_artifact_value(modified_resources, expanded_key, prop_name) if artifact_value is not None: - _set_prop_value(properties, prop_name, artifact_value) + set_prop_value(properties, prop_name, artifact_value) # Propagate auto dependency layer references from expanded functions # to the ForEach body. Each expanded function may have Layers entries @@ -723,10 +729,10 @@ def _count_dynamic_properties( share the same property name (e.g., two resources both with DefinitionUri). """ from samcli.lib.cfn_language_extensions.models import PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES - from samcli.lib.package.language_extensions_packaging import ( - _get_prop_value, - _leaf_prop_name, - _resolve_property_paths, + from samcli.lib.cfn_language_extensions.property_paths import ( + get_prop_value, + leaf_prop_name, + resolve_property_paths, ) count: Counter = Counter() @@ -740,12 +746,12 @@ def _count_dynamic_properties( if not isinstance(rprops, dict): continue candidate_paths = PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES.get(rtype, []) - for pname in _resolve_property_paths(candidate_paths, rprops): - pval = _get_prop_value(rprops, pname) + for pname in resolve_property_paths(candidate_paths, rprops): + pval = get_prop_value(rprops, pname) if pval is not None and contains_loop_variable(pval, loop_variable) and collection_values: # Count by leaf so collisions across resource types with # different dotted paths but the same leaf are detected. - count[_leaf_prop_name(pname)] += 1 + count[leaf_prop_name(pname)] += 1 return count @staticmethod @@ -788,10 +794,10 @@ def _collect_dynamic_mapping_entries( in the Mapping uses the leaf segment so it stays alphanumeric and matches the third argument of the generated Fn::FindInMap. """ - from samcli.lib.package.language_extensions_packaging import _leaf_prop_name + from samcli.lib.cfn_language_extensions.property_paths import leaf_prop_name mapping_entries: Dict[str, Dict[str, str]] = {} - leaf_name = _leaf_prop_name(prop_name) + leaf_name = leaf_prop_name(prop_name) for coll_value in collection_values: if outer_context: @@ -830,9 +836,9 @@ def _collect_nested_mapping_entry( """ import itertools - from samcli.lib.package.language_extensions_packaging import _leaf_prop_name + from samcli.lib.cfn_language_extensions.property_paths import leaf_prop_name - leaf_name = _leaf_prop_name(prop_name) + leaf_name = leaf_prop_name(prop_name) outer_collections = [oc[1] for oc in outer_context] outer_vars = [oc[0] for oc in outer_context] @@ -936,7 +942,7 @@ def _get_artifact_value(modified_resources: Dict, expanded_key: str, prop_name: ``prop_name`` may be a jmespath dotted path (e.g. "Command.ScriptLocation"). """ - from samcli.lib.package.language_extensions_packaging import _get_prop_value + from samcli.lib.cfn_language_extensions.property_paths import get_prop_value modified_resource = modified_resources.get(expanded_key, {}) if not isinstance(modified_resource, dict): @@ -944,37 +950,7 @@ def _get_artifact_value(modified_resources: Dict, expanded_key: str, prop_name: modified_props = modified_resource.get("Properties", {}) if not isinstance(modified_props, dict): return None - return _get_prop_value(modified_props, prop_name) - - def _copy_artifact_paths(self, original_resource: Dict, modified_resource: Dict) -> None: - """ - Copy artifact paths from modified resource to original resource. - - Uses PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES to determine which - properties to copy, avoiding a hardcoded elif chain. - - Parameters - ---------- - original_resource : Dict - The original resource (will be modified in place) - modified_resource : Dict - The modified resource with updated artifact paths - """ - from samcli.lib.cfn_language_extensions.models import PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES - from samcli.lib.package.language_extensions_packaging import _get_prop_value, _set_prop_value - - original_props = original_resource.get("Properties", {}) - modified_props = modified_resource.get("Properties", {}) - resource_type = original_resource.get("Type", "") - - prop_names = PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES.get(resource_type) - if not prop_names: - return - - for prop_name in prop_names: - value = _get_prop_value(modified_props, prop_name) - if value is not None: - _set_prop_value(original_props, prop_name, value) + return get_prop_value(modified_props, prop_name) def _gen_success_msg(self, artifacts_dir: str, output_template_path: str, is_default_build_dir: bool) -> str: """ diff --git a/samcli/lib/cfn_language_extensions/property_paths.py b/samcli/lib/cfn_language_extensions/property_paths.py new file mode 100644 index 0000000000..25dfd8b6f1 --- /dev/null +++ b/samcli/lib/cfn_language_extensions/property_paths.py @@ -0,0 +1,132 @@ +""" +Helpers for addressing CloudFormation resource artifact properties by jmespath. + +CloudFormation resource artifact properties are most-naturally addressed as +jmespath paths because some live at flat keys (``CodeUri``) and others at +nested locations (``Command.ScriptLocation`` on ``AWS::Glue::Job``, +``Code.ImageUri`` on a Lambda image function). The canonical packageable- +property registry (`PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES`) records each +in its addressing form, so consumers across SAM CLI must use jmespath get/set +to read and write them without clobbering siblings. + +This module provides those small, jmespath-aware helpers plus +``copy_artifact_properties``, the single helper that copies packaged +artifact values from an exported resource onto the corresponding original +resource for a given resource type. It is shared by the build-time and +package-time pipelines so the two cannot drift in their treatment of +property addressing. +""" + +from typing import Any, Dict, List, Optional, Set, Tuple + +import jmespath +from botocore.utils import set_value_from_jmespath + + +def get_prop_value(props: Dict[str, Any], path: str) -> Optional[Any]: + """Read a property by jmespath path. Supports flat keys ("CodeUri") and + dotted paths ("Command.ScriptLocation"). Returns None if missing. + """ + return jmespath.search(path, props) + + +def set_prop_value(props: Dict[str, Any], path: str, value: Any) -> None: + """Write a property by jmespath path. Creates intermediate dicts as + needed. Supports flat keys and dotted paths. + """ + set_value_from_jmespath(props, path, value) + + +def leaf_prop_name(path: str) -> str: + """Return the last segment of a jmespath property path. + + CloudFormation Mapping names must be alphanumeric, and the third argument + of ``Fn::FindInMap`` and the keys of Mapping value-dicts must match each + other as plain strings. Dotted property paths (e.g. ``Command.ScriptLocation``) + address the property *on* a resource; the *identifier* used in Mapping + names and FindInMap lookups must use only the leaf segment so the + generated CloudFormation is well-formed. + """ + return path.rsplit(".", 1)[-1] + + +def resolve_property_paths(paths: List[str], properties: Dict[str, Any]) -> List[str]: + """Filter ``paths`` so a parent path is dropped when a more specific + child path is present in ``properties``. + + Some resource types declare multiple alternative artifact paths in + ``PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES``. For example, + ``AWS::Lambda::Function`` lists ``Code`` (used for ZIP packages) and + ``Code.ImageUri`` (used for image packages). A given user template uses + only one of these shapes, but ``Code`` is a prefix of ``Code.ImageUri``, + so a naive iteration would address both paths and treat the same nested + value twice. Returning only the most specific resolved path picks the + correct interpretation for the user's actual property shape. + """ + # Sort longest-first so child paths win over their parents. + sorted_paths = sorted(paths, key=lambda p: -p.count(".")) + consumed_prefixes: Set[str] = set() + selected: List[str] = [] + for path in sorted_paths: + if get_prop_value(properties, path) is None: + continue + # If a more specific path under this prefix has already been selected, skip. + if any(other.startswith(path + ".") for other in consumed_prefixes): + continue + selected.append(path) + consumed_prefixes.add(path) + # Preserve original ordering for callers that care. + order = {p: i for i, p in enumerate(paths)} + selected.sort(key=lambda p: order.get(p, 0)) + return selected + + +def copy_artifact_properties( + original_props: Dict[str, Any], + exported_props: Dict[str, Any], + resource_type: str, + *, + foreach_key: Optional[str] = None, + dynamic_prop_keys: Optional[Set[Tuple[str, str]]] = None, +) -> bool: + """Copy packaged artifact property values from ``exported_props`` onto + ``original_props`` for the given resource type. + + Returns True if any property was copied. When called from the + language-extensions merge path, pass ``foreach_key`` and + ``dynamic_prop_keys`` so dynamic-loop properties (handled separately by + Mapping/FindInMap rewrites) are skipped. Build-time and non-ForEach + callers omit both kwargs. + + Both input dicts are addressed with jmespath, so dotted property paths + like ``Command.ScriptLocation`` or ``Code.ImageUri`` round-trip correctly + without creating literal flat keys. + """ + # Lazy import to avoid forcing the canonical-list module on every consumer + # of this small helper module at import time. + from samcli.lib.cfn_language_extensions.models import PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES + + paths = PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES.get(resource_type) + if not paths: + return False + + copied = False + for path in paths: + exported_value = get_prop_value(exported_props, path) + if exported_value is None: + continue + if dynamic_prop_keys and foreach_key and (foreach_key, path) in dynamic_prop_keys: + continue + set_prop_value(original_props, path, exported_value) + copied = True + + return copied + + +__all__ = [ + "copy_artifact_properties", + "get_prop_value", + "leaf_prop_name", + "resolve_property_paths", + "set_prop_value", +] diff --git a/samcli/lib/cfn_language_extensions/sam_integration.py b/samcli/lib/cfn_language_extensions/sam_integration.py index fd973eadb8..3838932f5e 100644 --- a/samcli/lib/cfn_language_extensions/sam_integration.py +++ b/samcli/lib/cfn_language_extensions/sam_integration.py @@ -412,10 +412,10 @@ def detect_foreach_dynamic_properties( if not isinstance(properties, dict): continue - from samcli.lib.package.language_extensions_packaging import _get_prop_value, _resolve_property_paths + from samcli.lib.cfn_language_extensions.property_paths import get_prop_value, resolve_property_paths - for prop_name in _resolve_property_paths(artifact_properties, properties): - prop_value = _get_prop_value(properties, prop_name) + for prop_name in resolve_property_paths(artifact_properties, properties): + prop_value = get_prop_value(properties, prop_name) if prop_value is not None: if contains_loop_variable(prop_value, loop_variable): dynamic_properties.append( diff --git a/samcli/lib/package/language_extensions_packaging.py b/samcli/lib/package/language_extensions_packaging.py index 9eb39366ae..7f24056384 100644 --- a/samcli/lib/package/language_extensions_packaging.py +++ b/samcli/lib/package/language_extensions_packaging.py @@ -14,12 +14,13 @@ from typing import Any, Dict, List, Optional, Tuple import click -import jmespath -from botocore.utils import set_value_from_jmespath -from samcli.lib.cfn_language_extensions.models import ( - PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES, - DynamicArtifactProperty, +from samcli.lib.cfn_language_extensions.models import DynamicArtifactProperty +from samcli.lib.cfn_language_extensions.property_paths import ( + copy_artifact_properties, + get_prop_value, + leaf_prop_name, + set_prop_value, ) from samcli.lib.cfn_language_extensions.sam_integration import ( contains_loop_variable, @@ -120,64 +121,6 @@ def generate_and_apply_artifact_mappings( return _apply_artifact_mappings_to_template(template, mappings, dynamic_properties, property_to_mapping) -def _get_prop_value(props: Dict[str, Any], prop_name: str) -> Optional[Any]: - """Read a property by jmespath path. Supports flat keys ("CodeUri") and - dotted paths ("Command.ScriptLocation"). Returns None if missing. - """ - return jmespath.search(prop_name, props) - - -def _set_prop_value(props: Dict[str, Any], prop_name: str, value: Any) -> None: - """Write a property by jmespath path. Creates intermediate dicts as needed. - Supports flat keys and dotted paths. - """ - set_value_from_jmespath(props, prop_name, value) - - -def _leaf_prop_name(prop_name: str) -> str: - """Return the last segment of a jmespath property name. - - CloudFormation Mapping names must be alphanumeric, and the third argument of - Fn::FindInMap and the keys of Mapping value-dicts must match each other as - plain strings. Dotted property names (e.g. "Command.ScriptLocation") are - used to *address* the property on a resource, but the *identifier* used in - Mapping names and FindInMap lookups must use only the leaf segment so the - generated CloudFormation is well-formed. - """ - return prop_name.rsplit(".", 1)[-1] - - -def _resolve_property_paths(prop_names: List[str], properties: Dict[str, Any]) -> List[str]: - """Filter ``prop_names`` so a parent path is dropped when a more specific - child path is present in ``properties``. - - Some resource types declare multiple alternative artifact paths in - ``PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES``. For example, - ``AWS::Lambda::Function`` lists ``Code`` (used for ZIP packages) and - ``Code.ImageUri`` (used for image packages). A given user template uses - only one of these shapes, but ``Code`` is a prefix of ``Code.ImageUri``, - so a naive iteration would address both paths and treat the same nested - value twice. Returning only the most specific resolved path picks the - correct interpretation for the user's actual property shape. - """ - # Sort longest-first so child paths win over their parents. - sorted_paths = sorted(prop_names, key=lambda p: -p.count(".")) - consumed_prefixes: set = set() - selected: List[str] = [] - for path in sorted_paths: - if _get_prop_value(properties, path) is None: - continue - # If a more specific path under this prefix has already been selected, skip. - if any(other.startswith(path + ".") for other in consumed_prefixes): - continue - selected.append(path) - consumed_prefixes.add(path) - # Preserve original ordering for callers that care. - order = {p: i for i, p in enumerate(prop_names)} - selected.sort(key=lambda p: order.get(p, 0)) - return selected - - # --------------------------------------------------------------------------- # Merge helpers # --------------------------------------------------------------------------- @@ -262,8 +205,12 @@ def _update_foreach_with_s3_uris( continue exported_props = exported_resource.get("Properties", {}) - _copy_artifact_uris_for_type( - properties, exported_props, resource_template.get("Type", ""), foreach_key, dynamic_prop_keys + copy_artifact_properties( + properties, + exported_props, + resource_template.get("Type", ""), + foreach_key=foreach_key, + dynamic_prop_keys=dynamic_prop_keys, ) @@ -288,40 +235,11 @@ def _build_expanded_key( def _copy_artifact_uris(original_resource: Dict, exported_resource: Dict) -> None: """Copy artifact URIs from exported resource to original resource.""" - original_props = original_resource.get("Properties", {}) - exported_props = exported_resource.get("Properties", {}) - resource_type = original_resource.get("Type", "") - _copy_artifact_uris_for_type(original_props, exported_props, resource_type) - - -def _copy_artifact_uris_for_type( - original_props: Dict, - exported_props: Dict, - resource_type: str, - foreach_key: Optional[str] = None, - dynamic_prop_keys: Optional[set] = None, -) -> bool: - """ - Copy artifact URIs based on resource type. - - Uses PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES to determine which - properties to copy, avoiding a long elif chain. - """ - prop_names = PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES.get(resource_type) - if not prop_names: - return False - - copied = False - for prop_name in prop_names: - exported_value = _get_prop_value(exported_props, prop_name) - if exported_value is None: - continue - if dynamic_prop_keys and foreach_key and (foreach_key, prop_name) in dynamic_prop_keys: - continue - _set_prop_value(original_props, prop_name, exported_value) - copied = True - - return copied + copy_artifact_properties( + original_resource.get("Properties", {}), + exported_resource.get("Properties", {}), + original_resource.get("Type", ""), + ) # --------------------------------------------------------------------------- @@ -376,7 +294,7 @@ def _compute_mapping_name( npath = _nesting_path(prop) # Use the leaf segment for the Mapping name so dotted property paths (e.g. # "Command.ScriptLocation") yield well-formed alphanumeric Mapping names. - leaf = _leaf_prop_name(prop.property_name) + leaf = leaf_prop_name(prop.property_name) base_name = f"SAM{leaf}{npath}" # Collision detection must also key on the leaf — two resources with # different dotted paths but the same leaf (e.g. "ImageUri" vs @@ -409,7 +327,7 @@ def _generate_artifact_mappings( # Keyed on the leaf so dotted paths that share a leaf (e.g. "ImageUri" vs # "Code.ImageUri") are detected as colliding and get a resource-key suffix. collision_groups: Dict[Tuple[str, str], int] = Counter( - (_nesting_path(p), _leaf_prop_name(p.property_name)) for p in dynamic_properties + (_nesting_path(p), leaf_prop_name(p.property_name)) for p in dynamic_properties ) for prop in dynamic_properties: @@ -419,7 +337,7 @@ def _generate_artifact_mappings( # The Mapping value-dict key and the third FindInMap argument must use # the leaf segment so dotted property paths produce well-formed # CloudFormation (Mappings can't contain dots in either name or keys). - leaf_name = _leaf_prop_name(prop.property_name) + leaf_name = leaf_prop_name(prop.property_name) if mapping_name not in mappings: mappings[mapping_name] = {} @@ -536,7 +454,7 @@ def _find_artifact_uri_for_resource( if not isinstance(properties, dict): return None - artifact_uri = _get_prop_value(properties, property_name) + artifact_uri = get_prop_value(properties, property_name) if isinstance(artifact_uri, str): return artifact_uri @@ -590,7 +508,7 @@ def _replace_dynamic_artifact_with_findmap( """ if mapping_name is None: # Use the leaf segment so dotted paths produce alphanumeric Mapping names. - mapping_name = f"SAM{_leaf_prop_name(prop.property_name)}{_nesting_path(prop)}" + mapping_name = f"SAM{leaf_prop_name(prop.property_name)}{_nesting_path(prop)}" current_scope = resources if prop.outer_loops: @@ -643,14 +561,14 @@ def _replace_dynamic_artifact_with_findmap( # Address the property at its dotted location (creating intermediate dicts # if needed) and use the leaf segment as the FindInMap third argument so # the lookup matches the Mapping value-dict key. - _set_prop_value( + set_prop_value( properties, prop.property_name, { "Fn::FindInMap": [ mapping_name, lookup_key, - _leaf_prop_name(prop.property_name), + leaf_prop_name(prop.property_name), ] }, ) diff --git a/tests/unit/commands/buildcmd/test_build_context_language_extensions.py b/tests/unit/commands/buildcmd/test_build_context_language_extensions.py index cf65a021a2..09aa87e22a 100644 --- a/tests/unit/commands/buildcmd/test_build_context_language_extensions.py +++ b/tests/unit/commands/buildcmd/test_build_context_language_extensions.py @@ -1,99 +1,93 @@ """ Tests for BuildContext language extensions support. -Covers _copy_artifact_paths for various resource types, -and bug condition exploration tests for auto dependency layer with ForEach templates. +Covers the build-time copy of packaged artifact properties via +``copy_artifact_properties`` for various resource types, plus bug-condition +exploration tests for auto dependency layer with ForEach templates. """ from unittest import TestCase from unittest.mock import patch, MagicMock, PropertyMock from samcli.commands.build.build_context import BuildContext +from samcli.lib.cfn_language_extensions.property_paths import copy_artifact_properties -class TestCopyArtifactPaths(TestCase): - """Tests for _copy_artifact_paths.""" +class TestCopyArtifactProperties(TestCase): + """Build-time call site of ``copy_artifact_properties`` (was: BuildContext._copy_artifact_paths).""" - def _make_context(self): - with patch.object(BuildContext, "__init__", lambda self: None): - return BuildContext() + def _copy(self, original, modified): + copy_artifact_properties( + original.get("Properties", {}), + modified.get("Properties", {}), + original.get("Type", ""), + ) def test_serverless_function_codeuri(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::Function", "Properties": {"CodeUri": "./src"}} modified = {"Type": "AWS::Serverless::Function", "Properties": {"CodeUri": ".aws-sam/build/Func"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["CodeUri"], ".aws-sam/build/Func") def test_serverless_function_imageuri(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::Function", "Properties": {"PackageType": "Image"}} modified = { "Type": "AWS::Serverless::Function", "Properties": {"ImageUri": "123.dkr.ecr.us-east-1.amazonaws.com/repo"}, } - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["ImageUri"], "123.dkr.ecr.us-east-1.amazonaws.com/repo") def test_lambda_function_code(self): - ctx = self._make_context() original = {"Type": "AWS::Lambda::Function", "Properties": {"Code": {"ZipFile": "code"}}} modified = {"Type": "AWS::Lambda::Function", "Properties": {"Code": {"S3Bucket": "b", "S3Key": "k"}}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["Code"]["S3Bucket"], "b") def test_serverless_layer_contenturi(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::LayerVersion", "Properties": {"ContentUri": "./layer"}} modified = {"Type": "AWS::Serverless::LayerVersion", "Properties": {"ContentUri": ".aws-sam/build/Layer"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["ContentUri"], ".aws-sam/build/Layer") def test_lambda_layer_content(self): - ctx = self._make_context() original = {"Type": "AWS::Lambda::LayerVersion", "Properties": {"Content": {"S3Bucket": "old"}}} modified = {"Type": "AWS::Lambda::LayerVersion", "Properties": {"Content": {"S3Bucket": "new"}}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["Content"]["S3Bucket"], "new") def test_serverless_api_definitionuri(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::Api", "Properties": {"DefinitionUri": "./api.yaml"}} modified = {"Type": "AWS::Serverless::Api", "Properties": {"DefinitionUri": ".aws-sam/build/api.yaml"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["DefinitionUri"], ".aws-sam/build/api.yaml") def test_serverless_httpapi_definitionuri(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::HttpApi", "Properties": {"DefinitionUri": "./api.yaml"}} modified = {"Type": "AWS::Serverless::HttpApi", "Properties": {"DefinitionUri": ".aws-sam/build/api.yaml"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["DefinitionUri"], ".aws-sam/build/api.yaml") def test_serverless_statemachine_definitionuri(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::StateMachine", "Properties": {"DefinitionUri": "./sm.json"}} modified = {"Type": "AWS::Serverless::StateMachine", "Properties": {"DefinitionUri": ".aws-sam/build/sm.json"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["DefinitionUri"], ".aws-sam/build/sm.json") def test_unknown_resource_type_no_copy(self): - ctx = self._make_context() original = {"Type": "AWS::SNS::Topic", "Properties": {"TopicName": "test"}} modified = {"Type": "AWS::SNS::Topic", "Properties": {"TopicName": "modified"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["TopicName"], "test") def test_no_matching_property_no_copy(self): - ctx = self._make_context() original = {"Type": "AWS::Serverless::Function", "Properties": {"Handler": "main.handler"}} modified = {"Type": "AWS::Serverless::Function", "Properties": {"Handler": "main.handler"}} - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertNotIn("CodeUri", original["Properties"]) def test_glue_job_dotted_path(self): """Regression: dotted property path (Command.ScriptLocation) must be copied.""" - ctx = self._make_context() original = { "Type": "AWS::Glue::Job", "Properties": {"Command": {"Name": "glueetl", "ScriptLocation": "./script.py"}}, @@ -102,7 +96,7 @@ def test_glue_job_dotted_path(self): "Type": "AWS::Glue::Job", "Properties": {"Command": {"Name": "glueetl", "ScriptLocation": "s3://b/k.py"}}, } - ctx._copy_artifact_paths(original, modified) + self._copy(original, modified) self.assertEqual(original["Properties"]["Command"]["ScriptLocation"], "s3://b/k.py") self.assertEqual(original["Properties"]["Command"]["Name"], "glueetl") diff --git a/tests/unit/commands/package/test_package_context.py b/tests/unit/commands/package/test_package_context.py index 76ba34b685..901cc6a612 100644 --- a/tests/unit/commands/package/test_package_context.py +++ b/tests/unit/commands/package/test_package_context.py @@ -12,13 +12,13 @@ contains_loop_variable, detect_dynamic_artifact_properties, ) +from samcli.lib.cfn_language_extensions.property_paths import copy_artifact_properties from samcli.lib.package.artifact_exporter import Template from samcli.lib.package.language_extensions_packaging import ( merge_language_extensions_s3_uris, warn_parameter_based_collections, _update_resources_with_s3_uris, _update_foreach_with_s3_uris, - _copy_artifact_uris_for_type, _copy_artifact_uris, _build_expanded_key, _generate_artifact_mappings, @@ -585,17 +585,17 @@ def test_update_original_template_with_s3_uris_regular_resources(self): "s3://bucket/xyz789.zip", ) - def test_copy_artifact_uris_for_type_serverless_function(self): + def test_copy_artifact_properties_serverless_function(self): """Test copying artifact URIs for AWS::Serverless::Function""" original_props = {"CodeUri": "./src", "Handler": "app.handler"} exported_props = {"CodeUri": "s3://bucket/code.zip", "Handler": "app.handler"} - result = _copy_artifact_uris_for_type(original_props, exported_props, "AWS::Serverless::Function") + result = copy_artifact_properties(original_props, exported_props, "AWS::Serverless::Function") self.assertTrue(result) self.assertEqual(original_props["CodeUri"], "s3://bucket/code.zip") - def test_copy_artifact_uris_for_type_lambda_function(self): + def test_copy_artifact_properties_lambda_function(self): """Test copying artifact URIs for AWS::Lambda::Function""" original_props = {"Code": "./src", "Handler": "app.handler"} exported_props = { @@ -603,37 +603,37 @@ def test_copy_artifact_uris_for_type_lambda_function(self): "Handler": "app.handler", } - result = _copy_artifact_uris_for_type(original_props, exported_props, "AWS::Lambda::Function") + result = copy_artifact_properties(original_props, exported_props, "AWS::Lambda::Function") self.assertTrue(result) self.assertEqual(original_props["Code"], {"S3Bucket": "bucket", "S3Key": "code.zip"}) - def test_copy_artifact_uris_for_type_serverless_layer(self): + def test_copy_artifact_properties_serverless_layer(self): """Test copying artifact URIs for AWS::Serverless::LayerVersion""" original_props = {"ContentUri": "./layer"} exported_props = {"ContentUri": "s3://bucket/layer.zip"} - result = _copy_artifact_uris_for_type(original_props, exported_props, "AWS::Serverless::LayerVersion") + result = copy_artifact_properties(original_props, exported_props, "AWS::Serverless::LayerVersion") self.assertTrue(result) self.assertEqual(original_props["ContentUri"], "s3://bucket/layer.zip") - def test_copy_artifact_uris_for_type_serverless_api(self): + def test_copy_artifact_properties_serverless_api(self): """Test copying artifact URIs for AWS::Serverless::Api""" original_props = {"DefinitionUri": "./api.yaml"} exported_props = {"DefinitionUri": "s3://bucket/api.yaml"} - result = _copy_artifact_uris_for_type(original_props, exported_props, "AWS::Serverless::Api") + result = copy_artifact_properties(original_props, exported_props, "AWS::Serverless::Api") self.assertTrue(result) self.assertEqual(original_props["DefinitionUri"], "s3://bucket/api.yaml") - def test_copy_artifact_uris_for_type_unknown_type(self): + def test_copy_artifact_properties_unknown_type(self): """Test that unknown resource types return False""" original_props = {"SomeProperty": "value"} exported_props = {"SomeProperty": "s3://bucket/value"} - result = _copy_artifact_uris_for_type(original_props, exported_props, "AWS::Unknown::Resource") + result = copy_artifact_properties(original_props, exported_props, "AWS::Unknown::Resource") self.assertFalse(result) @@ -975,85 +975,57 @@ def test_apply_artifact_mappings_to_template_integration(self): code_uri = body["${Name}Service"]["Properties"]["CodeUri"] self.assertEqual(code_uri, {"Fn::FindInMap": ["SAMCodeUriServices", {"Ref": "Name"}, "CodeUri"]}) - def test_get_prop_value_flat_key(self): - from samcli.lib.package.language_extensions_packaging import _get_prop_value + # NOTE: low-level get_prop_value / set_prop_value tests live in + # tests/unit/lib/cfn_language_extensions/test_property_paths.py — that's + # where the helpers themselves are now defined. - self.assertEqual(_get_prop_value({"CodeUri": "s3://b/k"}, "CodeUri"), "s3://b/k") - - def test_get_prop_value_dotted_key(self): - from samcli.lib.package.language_extensions_packaging import _get_prop_value - - props = {"Command": {"ScriptLocation": "s3://b/k.py"}} - self.assertEqual(_get_prop_value(props, "Command.ScriptLocation"), "s3://b/k.py") - - def test_get_prop_value_missing_returns_none(self): - from samcli.lib.package.language_extensions_packaging import _get_prop_value - - self.assertIsNone(_get_prop_value({"CodeUri": "x"}, "Command.ScriptLocation")) - - def test_set_prop_value_flat_key(self): - from samcli.lib.package.language_extensions_packaging import _set_prop_value - - props = {"CodeUri": "./src"} - _set_prop_value(props, "CodeUri", "s3://b/k") - self.assertEqual(props["CodeUri"], "s3://b/k") - - def test_set_prop_value_dotted_key_creates_intermediate(self): - from samcli.lib.package.language_extensions_packaging import _set_prop_value - - props = {"Command": {"Name": "glueetl"}} - _set_prop_value(props, "Command.ScriptLocation", "s3://b/k.py") - self.assertEqual(props["Command"]["ScriptLocation"], "s3://b/k.py") - # Existing keys preserved - self.assertEqual(props["Command"]["Name"], "glueetl") - - def test_copy_artifact_uris_for_type_glue_job_dotted_path(self): + def test_copy_artifact_properties_glue_job_dotted_path(self): """Regression: dotted property path (Command.ScriptLocation) must be copied.""" original_props = {"Command": {"Name": "glueetl", "ScriptLocation": "./script.py"}} exported_props = {"Command": {"Name": "glueetl", "ScriptLocation": "s3://b/k.py"}} - result = _copy_artifact_uris_for_type(original_props, exported_props, "AWS::Glue::Job") + result = copy_artifact_properties(original_props, exported_props, "AWS::Glue::Job") self.assertTrue(result) self.assertEqual(original_props["Command"]["ScriptLocation"], "s3://b/k.py") # Sibling keys preserved self.assertEqual(original_props["Command"]["Name"], "glueetl") - def test_copy_artifact_uris_for_type_serverless_application(self): + def test_copy_artifact_properties_serverless_application(self): """Regression for #9005: Location URI must be copied for AWS::Serverless::Application.""" original_props = {"Location": "./PublisherApi/template.yaml"} exported_props = {"Location": "https://s3.amazonaws.com/bucket/abc123.template"} - self.assertTrue(_copy_artifact_uris_for_type(original_props, exported_props, "AWS::Serverless::Application")) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::Serverless::Application")) self.assertEqual(original_props["Location"], "https://s3.amazonaws.com/bucket/abc123.template") - def test_copy_artifact_uris_for_type_cloudformation_stack(self): + def test_copy_artifact_properties_cloudformation_stack(self): """TemplateURL must be copied for AWS::CloudFormation::Stack.""" original_props = {"TemplateURL": "./child/template.yaml"} exported_props = {"TemplateURL": "https://s3.amazonaws.com/bucket/xyz789.template"} - self.assertTrue(_copy_artifact_uris_for_type(original_props, exported_props, "AWS::CloudFormation::Stack")) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::CloudFormation::Stack")) self.assertEqual(original_props["TemplateURL"], "https://s3.amazonaws.com/bucket/xyz789.template") - def test_copy_artifact_uris_for_type_cloudformation_stackset(self): + def test_copy_artifact_properties_cloudformation_stackset(self): original_props = {"TemplateURL": "./child.yaml"} exported_props = {"TemplateURL": "https://s3.amazonaws.com/b/k.template"} - self.assertTrue(_copy_artifact_uris_for_type(original_props, exported_props, "AWS::CloudFormation::StackSet")) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::CloudFormation::StackSet")) self.assertEqual(original_props["TemplateURL"], "https://s3.amazonaws.com/b/k.template") - def test_copy_artifact_uris_for_type_eb_application_version(self): + def test_copy_artifact_properties_eb_application_version(self): original_props = {"SourceBundle": "./bundle.zip"} exported_props = {"SourceBundle": {"S3Bucket": "b", "S3Key": "k"}} self.assertTrue( - _copy_artifact_uris_for_type(original_props, exported_props, "AWS::ElasticBeanstalk::ApplicationVersion") + copy_artifact_properties(original_props, exported_props, "AWS::ElasticBeanstalk::ApplicationVersion") ) self.assertEqual(original_props["SourceBundle"], {"S3Bucket": "b", "S3Key": "k"}) - def test_copy_artifact_uris_for_type_appsync_graphql_schema(self): + def test_copy_artifact_properties_appsync_graphql_schema(self): original_props = {"DefinitionS3Location": "./schema.graphql"} exported_props = {"DefinitionS3Location": "s3://b/schema.graphql"} - self.assertTrue(_copy_artifact_uris_for_type(original_props, exported_props, "AWS::AppSync::GraphQLSchema")) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::AppSync::GraphQLSchema")) self.assertEqual(original_props["DefinitionS3Location"], "s3://b/schema.graphql") - def test_copy_artifact_uris_for_type_appsync_resolver_all_three(self): + def test_copy_artifact_properties_appsync_resolver_all_three(self): original_props = { "RequestMappingTemplateS3Location": "./req.vtl", "ResponseMappingTemplateS3Location": "./res.vtl", @@ -1064,40 +1036,36 @@ def test_copy_artifact_uris_for_type_appsync_resolver_all_three(self): "ResponseMappingTemplateS3Location": "s3://b/res.vtl", "CodeS3Location": "s3://b/code.js", } - self.assertTrue(_copy_artifact_uris_for_type(original_props, exported_props, "AWS::AppSync::Resolver")) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::AppSync::Resolver")) self.assertEqual(original_props["RequestMappingTemplateS3Location"], "s3://b/req.vtl") self.assertEqual(original_props["ResponseMappingTemplateS3Location"], "s3://b/res.vtl") self.assertEqual(original_props["CodeS3Location"], "s3://b/code.js") - def test_copy_artifact_uris_for_type_appsync_function_configuration(self): + def test_copy_artifact_properties_appsync_function_configuration(self): original_props = {"CodeS3Location": "./code.js"} exported_props = {"CodeS3Location": "s3://b/code.js"} - self.assertTrue( - _copy_artifact_uris_for_type(original_props, exported_props, "AWS::AppSync::FunctionConfiguration") - ) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::AppSync::FunctionConfiguration")) self.assertEqual(original_props["CodeS3Location"], "s3://b/code.js") - def test_copy_artifact_uris_for_type_cloudformation_module_version(self): + def test_copy_artifact_properties_cloudformation_module_version(self): original_props = {"ModulePackage": "./module.zip"} exported_props = {"ModulePackage": "s3://b/module.zip"} - self.assertTrue( - _copy_artifact_uris_for_type(original_props, exported_props, "AWS::CloudFormation::ModuleVersion") - ) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::CloudFormation::ModuleVersion")) self.assertEqual(original_props["ModulePackage"], "s3://b/module.zip") - def test_copy_artifact_uris_for_type_cloudformation_resource_version(self): + def test_copy_artifact_properties_cloudformation_resource_version(self): original_props = {"SchemaHandlerPackage": "./pkg.zip"} exported_props = {"SchemaHandlerPackage": "s3://b/pkg.zip"} self.assertTrue( - _copy_artifact_uris_for_type(original_props, exported_props, "AWS::CloudFormation::ResourceVersion") + copy_artifact_properties(original_props, exported_props, "AWS::CloudFormation::ResourceVersion") ) self.assertEqual(original_props["SchemaHandlerPackage"], "s3://b/pkg.zip") - def test_copy_artifact_uris_for_type_lambda_function_image_dotted(self): + def test_copy_artifact_properties_lambda_function_image_dotted(self): # AWS::Lambda::Function has both Code and Code.ImageUri (dotted path) original_props = {"Code": {"ImageUri": "local-tag:latest"}} exported_props = {"Code": {"ImageUri": "111.dkr.ecr.us-east-1.amazonaws.com/r:tag"}} - self.assertTrue(_copy_artifact_uris_for_type(original_props, exported_props, "AWS::Lambda::Function")) + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::Lambda::Function")) self.assertEqual(original_props["Code"]["ImageUri"], "111.dkr.ecr.us-east-1.amazonaws.com/r:tag") # ========================================================================= diff --git a/tests/unit/commands/package/test_package_context_language_extensions.py b/tests/unit/commands/package/test_package_context_language_extensions.py index 3ae83ade7e..9baad00f65 100644 --- a/tests/unit/commands/package/test_package_context_language_extensions.py +++ b/tests/unit/commands/package/test_package_context_language_extensions.py @@ -1,17 +1,19 @@ """ Tests for language extensions packaging support. -Covers _copy_artifact_uris_for_type with various resource types and dynamic property skipping. -These functions now live in samcli.lib.package.language_extensions_packaging. +Covers ``copy_artifact_properties`` with various resource types and dynamic +property skipping. The helper now lives in +``samcli.lib.cfn_language_extensions.property_paths`` and is shared by +``samcli.lib.package.language_extensions_packaging`` and ``samcli.commands.build.build_context``. """ from unittest import TestCase from unittest.mock import patch, MagicMock from samcli.lib.cfn_language_extensions.exceptions import InvalidTemplateException +from samcli.lib.cfn_language_extensions.property_paths import copy_artifact_properties from samcli.lib.package.language_extensions_packaging import ( _compute_mapping_name, - _copy_artifact_uris_for_type, _nesting_path, _prop_identity, _update_foreach_with_s3_uris, @@ -21,110 +23,110 @@ ) -class TestCopyArtifactUrisForType(TestCase): - """Tests for _copy_artifact_uris_for_type.""" +class TestCopyArtifactProperties(TestCase): + """Per-resource-type behavior of the consolidated ``copy_artifact_properties`` helper.""" def test_serverless_function_codeuri(self): original = {} exported = {"CodeUri": "s3://bucket/code.zip"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::Function") + result = copy_artifact_properties(original, exported, "AWS::Serverless::Function") self.assertTrue(result) self.assertEqual(original["CodeUri"], "s3://bucket/code.zip") def test_serverless_function_imageuri(self): original = {} exported = {"ImageUri": "123456.dkr.ecr.us-east-1.amazonaws.com/repo:tag"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::Function") + result = copy_artifact_properties(original, exported, "AWS::Serverless::Function") self.assertTrue(result) self.assertEqual(original["ImageUri"], "123456.dkr.ecr.us-east-1.amazonaws.com/repo:tag") def test_lambda_function_code(self): original = {} exported = {"Code": {"S3Bucket": "bucket", "S3Key": "key"}} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Lambda::Function") + result = copy_artifact_properties(original, exported, "AWS::Lambda::Function") self.assertTrue(result) self.assertEqual(original["Code"]["S3Bucket"], "bucket") def test_serverless_layer_contenturi(self): original = {} exported = {"ContentUri": "s3://bucket/layer.zip"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::LayerVersion") + result = copy_artifact_properties(original, exported, "AWS::Serverless::LayerVersion") self.assertTrue(result) self.assertEqual(original["ContentUri"], "s3://bucket/layer.zip") def test_lambda_layer_content(self): original = {} exported = {"Content": {"S3Bucket": "bucket", "S3Key": "key"}} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Lambda::LayerVersion") + result = copy_artifact_properties(original, exported, "AWS::Lambda::LayerVersion") self.assertTrue(result) def test_serverless_api_definitionuri(self): original = {} exported = {"DefinitionUri": "s3://bucket/api.yaml"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::Api") + result = copy_artifact_properties(original, exported, "AWS::Serverless::Api") self.assertTrue(result) self.assertEqual(original["DefinitionUri"], "s3://bucket/api.yaml") def test_serverless_httpapi_definitionuri(self): original = {} exported = {"DefinitionUri": "s3://bucket/httpapi.yaml"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::HttpApi") + result = copy_artifact_properties(original, exported, "AWS::Serverless::HttpApi") self.assertTrue(result) def test_serverless_statemachine_definitionuri(self): original = {} exported = {"DefinitionUri": "s3://bucket/sm.json"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::StateMachine") + result = copy_artifact_properties(original, exported, "AWS::Serverless::StateMachine") self.assertTrue(result) def test_serverless_graphqlapi_schemauri(self): original = {} exported = {"SchemaUri": "s3://bucket/schema.graphql"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::GraphQLApi") + result = copy_artifact_properties(original, exported, "AWS::Serverless::GraphQLApi") self.assertTrue(result) self.assertEqual(original["SchemaUri"], "s3://bucket/schema.graphql") def test_serverless_graphqlapi_codeuri(self): original = {} exported = {"CodeUri": "s3://bucket/resolvers.zip"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::GraphQLApi") + result = copy_artifact_properties(original, exported, "AWS::Serverless::GraphQLApi") self.assertTrue(result) def test_apigateway_restapi_bodys3location(self): original = {} exported = {"BodyS3Location": "s3://bucket/body.yaml"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::ApiGateway::RestApi") + result = copy_artifact_properties(original, exported, "AWS::ApiGateway::RestApi") self.assertTrue(result) def test_apigatewayv2_api_bodys3location(self): original = {} exported = {"BodyS3Location": "s3://bucket/body.yaml"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::ApiGatewayV2::Api") + result = copy_artifact_properties(original, exported, "AWS::ApiGatewayV2::Api") self.assertTrue(result) def test_stepfunctions_statemachine_definitions3location(self): original = {} exported = {"DefinitionS3Location": "s3://bucket/sm.json"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::StepFunctions::StateMachine") + result = copy_artifact_properties(original, exported, "AWS::StepFunctions::StateMachine") self.assertTrue(result) def test_unknown_resource_type_returns_false(self): original = {} exported = {"SomeUri": "s3://bucket/thing"} - result = _copy_artifact_uris_for_type(original, exported, "AWS::SNS::Topic") + result = copy_artifact_properties(original, exported, "AWS::SNS::Topic") self.assertFalse(result) def test_no_matching_property_returns_false(self): original = {} exported = {"Handler": "index.handler"} # Not an artifact property - result = _copy_artifact_uris_for_type(original, exported, "AWS::Serverless::Function") + result = copy_artifact_properties(original, exported, "AWS::Serverless::Function") self.assertFalse(result) def test_dynamic_property_skipped(self): original = {} exported = {"CodeUri": "s3://bucket/code.zip"} dynamic_keys = {("Fn::ForEach::Funcs", "CodeUri")} - result = _copy_artifact_uris_for_type( + result = copy_artifact_properties( original, exported, "AWS::Serverless::Function", @@ -138,7 +140,7 @@ def test_non_dynamic_property_not_skipped(self): original = {} exported = {"CodeUri": "s3://bucket/code.zip"} dynamic_keys = {("Fn::ForEach::Other", "CodeUri")} - result = _copy_artifact_uris_for_type( + result = copy_artifact_properties( original, exported, "AWS::Serverless::Function", @@ -152,7 +154,7 @@ def test_no_foreach_key_skips_dynamic_check(self): original = {} exported = {"CodeUri": "s3://bucket/code.zip"} dynamic_keys = {("Fn::ForEach::Funcs", "CodeUri")} - result = _copy_artifact_uris_for_type( + result = copy_artifact_properties( original, exported, "AWS::Serverless::Function", diff --git a/tests/unit/lib/cfn_language_extensions/test_property_paths.py b/tests/unit/lib/cfn_language_extensions/test_property_paths.py new file mode 100644 index 0000000000..efcbff5f6d --- /dev/null +++ b/tests/unit/lib/cfn_language_extensions/test_property_paths.py @@ -0,0 +1,155 @@ +"""Tests for samcli.lib.cfn_language_extensions.property_paths. + +Verifies the small jmespath-aware helpers used across the language-extensions +build- and package-time pipelines, plus the contract that every entry in +``PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES`` round-trips through them. +""" + +from unittest import TestCase + +from samcli.lib.cfn_language_extensions.models import PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES +from samcli.lib.cfn_language_extensions.property_paths import ( + copy_artifact_properties, + get_prop_value, + leaf_prop_name, + resolve_property_paths, + set_prop_value, +) + + +class TestGetPropValue(TestCase): + def test_flat_key(self): + self.assertEqual(get_prop_value({"CodeUri": "s3://b/k"}, "CodeUri"), "s3://b/k") + + def test_dotted_key(self): + props = {"Command": {"ScriptLocation": "s3://b/k.py"}} + self.assertEqual(get_prop_value(props, "Command.ScriptLocation"), "s3://b/k.py") + + def test_missing_returns_none(self): + self.assertIsNone(get_prop_value({"CodeUri": "x"}, "Command.ScriptLocation")) + + +class TestSetPropValue(TestCase): + def test_flat_key(self): + props = {"CodeUri": "./src"} + set_prop_value(props, "CodeUri", "s3://b/k") + self.assertEqual(props["CodeUri"], "s3://b/k") + + def test_dotted_key_creates_intermediate(self): + props = {"Command": {"Name": "glueetl"}} + set_prop_value(props, "Command.ScriptLocation", "s3://b/k.py") + self.assertEqual(props["Command"]["ScriptLocation"], "s3://b/k.py") + # Sibling keys preserved. + self.assertEqual(props["Command"]["Name"], "glueetl") + + +class TestLeafPropName(TestCase): + def test_flat_path_returns_self(self): + self.assertEqual(leaf_prop_name("CodeUri"), "CodeUri") + + def test_dotted_path_returns_leaf(self): + self.assertEqual(leaf_prop_name("Command.ScriptLocation"), "ScriptLocation") + self.assertEqual(leaf_prop_name("Code.ImageUri"), "ImageUri") + + +class TestResolvePropertyPaths(TestCase): + def test_picks_dotted_child_when_user_uses_image_shape(self): + # AWS::Lambda::Function: ZIP uses Code, image uses Code.ImageUri. + # User template has the image shape; resolver must pick the dotted entry. + props = {"Code": {"ImageUri": "local-tag:latest"}} + self.assertEqual(resolve_property_paths(["Code", "Code.ImageUri"], props), ["Code.ImageUri"]) + + def test_picks_parent_when_user_uses_zip_shape(self): + # ZIP shape: Code is a string path or {S3Bucket, S3Key} dict — never has ImageUri sub-key. + props = {"Code": {"S3Bucket": "b", "S3Key": "k"}} + self.assertEqual(resolve_property_paths(["Code", "Code.ImageUri"], props), ["Code"]) + + def test_skips_paths_with_no_value(self): + self.assertEqual(resolve_property_paths(["CodeUri", "ImageUri"], {"CodeUri": "x"}), ["CodeUri"]) + + def test_preserves_input_order_for_unrelated_paths(self): + props = {"SchemaUri": "s.graphql", "CodeUri": "src/"} + # Both present; neither is a prefix of the other; order should match input. + self.assertEqual( + resolve_property_paths(["SchemaUri", "CodeUri"], props), + ["SchemaUri", "CodeUri"], + ) + + +class TestCopyArtifactProperties(TestCase): + """End-to-end behaviors of the consolidated copy helper. + + The full per-resource-type matrix (Function/Layer/Api/StateMachine/...) is + in tests/unit/commands/package/test_package_context.py — those tests + exercise the same call site against the canonical dict. + """ + + def test_copies_flat_property(self): + original_props = {"CodeUri": "./src"} + exported_props = {"CodeUri": "s3://b/k"} + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::Serverless::Function")) + self.assertEqual(original_props["CodeUri"], "s3://b/k") + + def test_copies_dotted_property_without_clobbering_siblings(self): + # Glue.Command.ScriptLocation: must land under Properties.Command.ScriptLocation, + # NOT at a literal "Command.ScriptLocation" flat key. + original_props = {"Command": {"Name": "glueetl", "ScriptLocation": "./script.py"}} + exported_props = {"Command": {"Name": "glueetl", "ScriptLocation": "s3://b/k.py"}} + self.assertTrue(copy_artifact_properties(original_props, exported_props, "AWS::Glue::Job")) + self.assertEqual(original_props["Command"]["ScriptLocation"], "s3://b/k.py") + self.assertEqual(original_props["Command"]["Name"], "glueetl") + self.assertNotIn("Command.ScriptLocation", original_props) + + def test_unknown_resource_type_is_noop(self): + original_props = {"X": "y"} + self.assertFalse(copy_artifact_properties(original_props, {"X": "z"}, "AWS::SNS::Topic")) + # Original was not touched. + self.assertEqual(original_props["X"], "y") + + def test_dynamic_prop_keys_skip(self): + # When called from the language-extensions merge path, dynamic + # (loop-templated) properties are skipped and handled by Mappings. + original_props = {"CodeUri": "./services/${Name}"} + exported_props = {"CodeUri": "s3://b/k"} + result = copy_artifact_properties( + original_props, + exported_props, + "AWS::Serverless::Function", + foreach_key="Fn::ForEach::Services", + dynamic_prop_keys={("Fn::ForEach::Services", "CodeUri")}, + ) + self.assertFalse(result) + # Original was not overwritten — Mapping/FindInMap will replace it later. + self.assertEqual(original_props["CodeUri"], "./services/${Name}") + + +class TestPackageablePropertyContract(TestCase): + """Lock in the structural contract every artifact property must satisfy. + + If a future addition to ``RESOURCES_WITH_LOCAL_PATHS`` / + ``RESOURCES_WITH_IMAGE_COMPONENT`` declares a property whose path doesn't + round-trip through ``set_prop_value`` + ``get_prop_value``, or whose leaf + isn't alphanumeric (it must be, to appear in a CloudFormation Mapping + name), this test fails loudly at the contract layer rather than at a + user-template execution path. + """ + + def test_every_packageable_property_round_trips_and_has_alphanumeric_leaf(self): + for resource_type, paths in PACKAGEABLE_RESOURCE_ARTIFACT_PROPERTIES.items(): + for path in paths: + with self.subTest(resource_type=resource_type, path=path): + # Round trip: write then read. + props: dict = {} + set_prop_value(props, path, "sentinel") + self.assertEqual( + get_prop_value(props, path), + "sentinel", + f"{resource_type}.{path}: set_prop_value/get_prop_value did not round-trip", + ) + # Leaf must be alphanumeric — Mapping names can't contain dots. + leaf = leaf_prop_name(path) + self.assertTrue( + leaf.isalnum(), + f"{resource_type}.{path}: leaf {leaf!r} must be alphanumeric " + f"(used as Mapping name suffix and FindInMap third arg)", + )