refactor(cfn-lang-ext): consolidate copy helpers and relocate prop-name helpers

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,45 +942,15 @@ 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):
             return None
         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:
         """

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",
+]

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(