Add support for environment variable placeholders in hot-reloading pa…

…ths (#10857)

localstack/services/lambda_/invocation/lambda_models.py

@@ -5,6 +5,7 @@
 
 import dataclasses
 import logging
+import os.path
 import shutil
 import tempfile
 import threading
@@ -257,7 +258,8 @@ def generate_presigned_url(self, endpoint_url: str | None = None) -> str:
         return f"Code location: {self.host_path}"
 
     def get_unzipped_code_location(self) -> Path:
-        return Path(self.host_path)
+        path = os.path.expandvars(self.host_path)
+        return Path(path)
 
     def is_hot_reloading(self) -> bool:
         """

localstack/services/lambda_/invocation/lambda_service.py

@@ -3,6 +3,7 @@
 import dataclasses
 import io
 import logging
+import os.path
 import random
 import uuid
 from concurrent.futures import Executor, Future, ThreadPoolExecutor
@@ -549,12 +550,28 @@ def store_lambda_archive(
     )
 
 
-def create_hot_reloading_code(path: str) -> HotReloadingCode:
-    # TODO extract into other function
-    if not PurePosixPath(path).is_absolute() and not PureWindowsPath(path).is_absolute():
+def assert_hot_reloading_path_absolute(path: str) -> None:
+    """
+    Check whether a given path, after environment variable substitution, is an absolute path.
+    Accepts either posix or windows paths, with environment placeholders.
+    Example placeholders: $ENV_VAR, ${ENV_VAR}
+
+    :param path: Posix or windows path, potentially containing environment variable placeholders.
+        Example: `$ENV_VAR/lambda/src` with `ENV_VAR=/home/user/test-repo` set.
+    """
+    # expand variables in path before checking for an absolute path
+    expanded_path = os.path.expandvars(path)
+    if (
+        not PurePosixPath(expanded_path).is_absolute()
+        and not PureWindowsPath(expanded_path).is_absolute()
+    ):
         raise InvalidParameterValueException(
             f"When using hot reloading, the archive key has to be an absolute path! Your archive key: {path}",
         )
+
+
+def create_hot_reloading_code(path: str) -> HotReloadingCode:
+    assert_hot_reloading_path_absolute(path)
     return HotReloadingCode(host_path=path)
 
 

tests/aws/services/lambda_/test_lambda_developer_tools.py

@@ -135,6 +135,57 @@ def test_hot_reloading_publish_version(
         )
         aws_client.lambda_.publish_version(FunctionName=function_name, CodeSha256="zipfilehash")
 
+    @markers.aws.only_localstack
+    def test_hot_reloading_error_path_not_absolute(
+        self,
+        create_lambda_function_aws,
+        lambda_su_role,
+        cleanups,
+        aws_client,
+    ):
+        """Tests validation of hot reloading paths"""
+        function_name = f"test-hot-reloading-{short_uid()}"
+        hot_reloading_bucket = config.BUCKET_MARKER_LOCAL
+        with pytest.raises(Exception):
+            aws_client.lambda_.create_function(
+                FunctionName=function_name,
+                Handler="handler.handler",
+                Code={"S3Bucket": hot_reloading_bucket, "S3Key": "not/an/absolute/path"},
+                Role=lambda_su_role,
+                Runtime=Runtime.python3_12,
+            )
+
+    @markers.aws.only_localstack
+    def test_hot_reloading_environment_placeholder(
+        self, create_lambda_function_aws, lambda_su_role, cleanups, aws_client, monkeypatch
+    ):
+        """Test hot reloading of lambda code when the S3Key containers an environment variable placeholder like $DIR"""
+        function_name = f"test-hot-reloading-{short_uid()}"
+        hot_reloading_bucket = config.BUCKET_MARKER_LOCAL
+        tmp_path = config.dirs.mounted_tmp
+        hot_reloading_dir_path = os.path.join(tmp_path, f"hot-reload-{short_uid()}")
+        mkdir(hot_reloading_dir_path)
+        cleanups.append(lambda: rm_rf(hot_reloading_dir_path))
+        function_content = load_file(HOT_RELOADING_PYTHON_HANDLER)
+        with open(os.path.join(hot_reloading_dir_path, "handler.py"), mode="wt") as f:
+            f.write(function_content)
+
+        mount_path = get_host_path_for_path_in_docker(hot_reloading_dir_path)
+        head, tail = os.path.split(mount_path)
+        monkeypatch.setenv("HEAD_DIR", head)
+
+        create_lambda_function_aws(
+            FunctionName=function_name,
+            Handler="handler.handler",
+            Code={"S3Bucket": hot_reloading_bucket, "S3Key": f"$HEAD_DIR/{tail}"},
+            Role=lambda_su_role,
+            Runtime=Runtime.python3_12,
+        )
+        response = aws_client.lambda_.invoke(FunctionName=function_name, Payload=b"{}")
+        response_dict = json.load(response["Payload"])
+        assert response_dict["counter"] == 1
+        assert response_dict["constant"] == "value1"
+
 
 class TestDockerFlags:
     @markers.aws.only_localstack