6815841a-f2d0-4f5f-8a01-53ddff0de095

Committed 21 May 2025 07:32AM UTC coverage: 86.655% (-0.01%) from 86.669%

Build # 6815841a-f2d0-4f5f-8a01-53ddff0de095

Build Type

push

circleci

Committed by

web-flow

Commit Message

CloudFormation V2 Engine: Support for DependsOn Blocks (#12644)

Run Details

51 of 53 new or added lines in 4 files covered. (96.23%)

162 existing lines in 16 files now uncovered.

64543 of 74483 relevant lines covered (86.65%)

0.87 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

59.41

/localstack-core/localstack/testing/scenario/cdk_lambda_helper.py

import base64
import os
import shutil
import tempfile
import zipfile
from pathlib import Path
from typing import TYPE_CHECKING

from botocore.exceptions import ClientError

from localstack.utils.aws.resources import create_s3_bucket
from localstack.utils.docker_utils import DOCKER_CLIENT
from localstack.utils.run import LOG, run

if TYPE_CHECKING:
    from mypy_boto3_ecr import ECRClient
    from mypy_boto3_s3 import S3Client


def load_python_lambda_to_s3(
    s3_client: "S3Client",
    bucket_name: str,
    key_name: str,
    code_path: str,
    additional_python_packages: list[str] = None,
):
    """
    Helper function to setup Lambdas that need additional python libs.
    Will create a temp-zip and upload in the s3 bucket.
    Installs additional libs and package with the zip

    :param s3_client: client for S3
    :param bucket_name: bucket name (bucket will be created)
    :param key_name: key name for the uploaded zip file
    :param code_path: the path to the source code that should be included
    :param additional_python_packages: a list of strings with python packages that are required to run the lambda
    :return: None
    """
    try:
        temp_dir = tempfile.mkdtemp()
        tmp_zip_path = os.path.join(tempfile.gettempdir(), "helper.zip")
        # install python packages
        if additional_python_packages:
            try:
                run(f"cd {temp_dir} && pip install {' '.join(additional_python_packages)} -t .")
            except Exception as e:
                LOG.error(
                    "Could not install additional packages %s: %s", additional_python_packages, e
                )
        # add the lambda to the directory
        _zip_lambda_resources(
            lambda_code_path=code_path,
            handler_file_name="index.py",
            resources_dir=temp_dir,
            zip_path=tmp_zip_path,
        )
        _upload_to_s3(s3_client, bucket_name=bucket_name, key_name=key_name, file=tmp_zip_path)

    finally:
        if temp_dir:
            shutil.rmtree(temp_dir)
        if tmp_zip_path and os.path.exists(tmp_zip_path):
            os.remove(tmp_zip_path)


def load_nodejs_lambda_to_s3(
    s3_client: "S3Client",
    bucket_name: str,
    key_name: str,
    code_path: str,
    additional_nodjs_packages: list[str] = None,
    additional_nodejs_packages: list[str] = None,
    additional_resources: list[str] = None,
):
    """
    Helper function to setup nodeJS Lambdas that need additional libs.
    Will create a temp-zip and upload in the s3 bucket.
    Installs additional libs and package with the zip

    :param s3_client: client for S3
    :param bucket_name: bucket name (bucket will be created)
    :param key_name: key name for the uploaded zip file
    :param code_path: the path to the source code that should be included
    :param additional_nodjs_packages: a list of strings with nodeJS packages that are required to run the lambda
    :param additional_nodejs_packages: a list of strings with nodeJS packages that are required to run the lambda
    :param additional_resources: list of path-strings to resources or internal libs that should be packaged into the lambda
    :return: None
    """
    additional_resources = additional_resources or []

    if additional_nodjs_packages:
        additional_nodejs_packages = additional_nodejs_packages or []
        additional_nodejs_packages.extend(additional_nodjs_packages)

    try:
        temp_dir = tempfile.mkdtemp()
        tmp_zip_path = os.path.join(tempfile.gettempdir(), "helper.zip")

        # Install NodeJS packages
        if additional_nodejs_packages:
            try:
                os.mkdir(os.path.join(temp_dir, "node_modules"))
                run(f"cd {temp_dir} && npm install {' '.join(additional_nodejs_packages)} ")
            except Exception as e:
                LOG.error(
                    "Could not install additional packages %s: %s", additional_nodejs_packages, e
                )

        for r in additional_resources:
            try:
                path = Path(r)
                if path.is_dir():
                    dir_name = os.path.basename(path)
                    dest_dir = os.path.join(temp_dir, dir_name)
                    shutil.copytree(path, dest_dir)
                elif path.is_file():
                    new_resource_temp_path = os.path.join(temp_dir, os.path.basename(path))
                    shutil.copy2(path, new_resource_temp_path)
            except Exception as e:
                LOG.error("Could not copy additional resources %s: %s", r, e)

        _zip_lambda_resources(
            lambda_code_path=code_path,
            handler_file_name="index.js",
            resources_dir=temp_dir,
            zip_path=tmp_zip_path,
        )
        _upload_to_s3(s3_client, bucket_name=bucket_name, key_name=key_name, file=tmp_zip_path)
    finally:
        if temp_dir:
            shutil.rmtree(temp_dir)
        if tmp_zip_path and os.path.exists(tmp_zip_path):
            os.remove(tmp_zip_path)


def _zip_lambda_resources(
    lambda_code_path: str, handler_file_name: str, resources_dir: str, zip_path: str
):
    # add the lambda to the directory
    new_resource_temp_path = os.path.join(resources_dir, handler_file_name)
    shutil.copy2(lambda_code_path, new_resource_temp_path)

    with zipfile.ZipFile(zip_path, "w") as temp_zip:
        # Add the contents of the existing ZIP file
        for root, _, files in os.walk(resources_dir):
            for file in files:
                file_path = os.path.join(root, file)
                archive_name = os.path.relpath(file_path, resources_dir)
                temp_zip.write(file_path, archive_name)


def generate_ecr_image_from_dockerfile(
    ecr_client: "ECRClient",
    repository_name: str,
    file_path: str,
    build_in_place: bool = False,
):
    """
    Helper function to generate an ECR image from a dockerfile.

    :param ecr_client: client for ECR
    :param repository_name: name for the repository to be created
    :param file_path: path of the file to be used
    :param build_in_place: build the container image in place rather than copying to a temporary location.
                           This is useful if the build context has other files.
    :return: None
    """
    repository_uri = ecr_client.create_repository(
        repositoryName=repository_name,
    )["repository"]["repositoryUri"]

    auth_response = ecr_client.get_authorization_token()
    auth_token = auth_response["authorizationData"][0]["authorizationToken"].encode()
    username, password = base64.b64decode(auth_token).decode().split(":")
    registry = auth_response["authorizationData"][0]["proxyEndpoint"]
    DOCKER_CLIENT.login(username, password, registry=registry)

    if build_in_place:
        destination_file = file_path
    else:
        temp_dir = tempfile.mkdtemp()
        destination_file = os.path.join(temp_dir, "Dockerfile")
        shutil.copy2(file_path, destination_file)
    DOCKER_CLIENT.build_image(dockerfile_path=destination_file, image_name=repository_uri)
    DOCKER_CLIENT.push_image(repository_uri)


def generate_ecr_image_from_docker_image(
    ecr_client: "ECRClient", repository_name: str, image_name: str, platform: str = "linux/amd64"
):
    """
    Parameters
    ----------
    ecr_client
    repository_name
    image_name
    platform

    Returns
    -------

    """

    DOCKER_CLIENT.pull_image(image_name, platform=platform)

    repository_uri = ecr_client.create_repository(
        repositoryName=repository_name,
    )["repository"]["repositoryUri"]

    auth_response = ecr_client.get_authorization_token()
    auth_token = auth_response["authorizationData"][0]["authorizationToken"].encode()
    username, password = base64.b64decode(auth_token).decode().split(":")
    registry = auth_response["authorizationData"][0]["proxyEndpoint"]
    DOCKER_CLIENT.login(username, password, registry=registry)

    DOCKER_CLIENT.tag_image(image_name, repository_uri)
    DOCKER_CLIENT.push_image(repository_uri)


def _upload_to_s3(s3_client: "S3Client", bucket_name: str, key_name: str, file: str):
    try:
        create_s3_bucket(bucket_name, s3_client)
    except ClientError as exc:
        # when creating an already existing bucket, regions differ in their behavior:
        # us-east-1 will silently pass (idempotent)
        # any other region will return a `BucketAlreadyOwnedByYou` exception.
        if exc.response["Error"]["Code"] != "BucketAlreadyOwnedByYou":
            raise exc
    s3_client.upload_file(Filename=file, Bucket=bucket_name, Key=key_name)

1	import base64	1✔
2	import os	1✔
3	import shutil	1✔
4	import tempfile	1✔
5	import zipfile	1✔
6	from pathlib import Path	1✔
7	from typing import TYPE_CHECKING	1✔
8
9	from botocore.exceptions import ClientError	1✔
10
11	from localstack.utils.aws.resources import create_s3_bucket	1✔
12	from localstack.utils.docker_utils import DOCKER_CLIENT	1✔
13	from localstack.utils.run import LOG, run	1✔
14
15	if TYPE_CHECKING:	1✔
16	from mypy_boto3_ecr import ECRClient	×
17	from mypy_boto3_s3 import S3Client	×
18
19
20	def load_python_lambda_to_s3(	1✔
21	s3_client: "S3Client",
22	bucket_name: str,
23	key_name: str,
24	code_path: str,
25	additional_python_packages: list[str] = None,
26	):
27	"""
28	Helper function to setup Lambdas that need additional python libs.
29	Will create a temp-zip and upload in the s3 bucket.
30	Installs additional libs and package with the zip
31
32	:param s3_client: client for S3
33	:param bucket_name: bucket name (bucket will be created)
34	:param key_name: key name for the uploaded zip file
35	:param code_path: the path to the source code that should be included
36	:param additional_python_packages: a list of strings with python packages that are required to run the lambda
37	:return: None
38	"""
39	try:	1✔
40	temp_dir = tempfile.mkdtemp()	1✔
41	tmp_zip_path = os.path.join(tempfile.gettempdir(), "helper.zip")	1✔
42	# install python packages
43	if additional_python_packages:	1✔
44	try:	1✔
45	run(f"cd {temp_dir} && pip install {' '.join(additional_python_packages)} -t .")	1✔
46	except Exception as e:	×
47	LOG.error(	×
48	"Could not install additional packages %s: %s", additional_python_packages, e
49	)
50	# add the lambda to the directory
51	_zip_lambda_resources(	1✔
52	lambda_code_path=code_path,
53	handler_file_name="index.py",
54	resources_dir=temp_dir,
55	zip_path=tmp_zip_path,
56	)
57	_upload_to_s3(s3_client, bucket_name=bucket_name, key_name=key_name, file=tmp_zip_path)	1✔
58
59	finally:
60	if temp_dir:	1✔
61	shutil.rmtree(temp_dir)	1✔
62	if tmp_zip_path and os.path.exists(tmp_zip_path):	1✔
63	os.remove(tmp_zip_path)	1✔
64
65
66	def load_nodejs_lambda_to_s3(	1✔
67	s3_client: "S3Client",
68	bucket_name: str,
69	key_name: str,
70	code_path: str,
71	additional_nodjs_packages: list[str] = None,
72	additional_nodejs_packages: list[str] = None,
73	additional_resources: list[str] = None,
74	):
75	"""
76	Helper function to setup nodeJS Lambdas that need additional libs.
77	Will create a temp-zip and upload in the s3 bucket.
78	Installs additional libs and package with the zip
79
80	:param s3_client: client for S3
81	:param bucket_name: bucket name (bucket will be created)
82	:param key_name: key name for the uploaded zip file
83	:param code_path: the path to the source code that should be included
84	:param additional_nodjs_packages: a list of strings with nodeJS packages that are required to run the lambda
85	:param additional_nodejs_packages: a list of strings with nodeJS packages that are required to run the lambda
86	:param additional_resources: list of path-strings to resources or internal libs that should be packaged into the lambda
87	:return: None
88	"""
89	additional_resources = additional_resources or []	1✔
90
91	if additional_nodjs_packages:	1✔
92	additional_nodejs_packages = additional_nodejs_packages or []	×
93	additional_nodejs_packages.extend(additional_nodjs_packages)	×
94
95	try:	1✔
96	temp_dir = tempfile.mkdtemp()	1✔
97	tmp_zip_path = os.path.join(tempfile.gettempdir(), "helper.zip")	1✔
98
99	# Install NodeJS packages
100	if additional_nodejs_packages:	1✔
101	try:	×
102	os.mkdir(os.path.join(temp_dir, "node_modules"))	×
103	run(f"cd {temp_dir} && npm install {' '.join(additional_nodejs_packages)} ")	×
104	except Exception as e:	×
105	LOG.error(	×
106	"Could not install additional packages %s: %s", additional_nodejs_packages, e
107	)
108
109	for r in additional_resources:	1✔
110	try:	1✔
111	path = Path(r)	1✔
112	if path.is_dir():	1✔
113	dir_name = os.path.basename(path)	1✔
114	dest_dir = os.path.join(temp_dir, dir_name)	1✔
115	shutil.copytree(path, dest_dir)	1✔
116	elif path.is_file():	×
117	new_resource_temp_path = os.path.join(temp_dir, os.path.basename(path))	×
118	shutil.copy2(path, new_resource_temp_path)	×
119	except Exception as e:	×
120	LOG.error("Could not copy additional resources %s: %s", r, e)	×
121
122	_zip_lambda_resources(	1✔
123	lambda_code_path=code_path,
124	handler_file_name="index.js",
125	resources_dir=temp_dir,
126	zip_path=tmp_zip_path,
127	)
128	_upload_to_s3(s3_client, bucket_name=bucket_name, key_name=key_name, file=tmp_zip_path)	1✔
129	finally:
130	if temp_dir:	1✔
131	shutil.rmtree(temp_dir)	1✔
132	if tmp_zip_path and os.path.exists(tmp_zip_path):	1✔
133	os.remove(tmp_zip_path)	1✔
134
135
136	def _zip_lambda_resources(	1✔
137	lambda_code_path: str, handler_file_name: str, resources_dir: str, zip_path: str
138	):
139	# add the lambda to the directory
140	new_resource_temp_path = os.path.join(resources_dir, handler_file_name)	1✔
141	shutil.copy2(lambda_code_path, new_resource_temp_path)	1✔
142
143	with zipfile.ZipFile(zip_path, "w") as temp_zip:	1✔
144	# Add the contents of the existing ZIP file
145	for root, _, files in os.walk(resources_dir):	1✔
146	for file in files:	1✔
147	file_path = os.path.join(root, file)	1✔
148	archive_name = os.path.relpath(file_path, resources_dir)	1✔
149	temp_zip.write(file_path, archive_name)	1✔
150
151
152	def generate_ecr_image_from_dockerfile(	1✔
153	ecr_client: "ECRClient",
154	repository_name: str,
155	file_path: str,
156	build_in_place: bool = False,
157	):
158	"""
159	Helper function to generate an ECR image from a dockerfile.
160
161	:param ecr_client: client for ECR
162	:param repository_name: name for the repository to be created
163	:param file_path: path of the file to be used
164	:param build_in_place: build the container image in place rather than copying to a temporary location.
165	This is useful if the build context has other files.
166	:return: None
167	"""
168	repository_uri = ecr_client.create_repository(	×
169	repositoryName=repository_name,
170	)["repository"]["repositoryUri"]
171
172	auth_response = ecr_client.get_authorization_token()	×
173	auth_token = auth_response["authorizationData"][0]["authorizationToken"].encode()	×
174	username, password = base64.b64decode(auth_token).decode().split(":")	×
175	registry = auth_response["authorizationData"][0]["proxyEndpoint"]	×
176	DOCKER_CLIENT.login(username, password, registry=registry)	×
177
178	if build_in_place:	×
179	destination_file = file_path	×
180	else:
181	temp_dir = tempfile.mkdtemp()	×
182	destination_file = os.path.join(temp_dir, "Dockerfile")	×
183	shutil.copy2(file_path, destination_file)	×
184	DOCKER_CLIENT.build_image(dockerfile_path=destination_file, image_name=repository_uri)	×
185	DOCKER_CLIENT.push_image(repository_uri)	×
186
187
188	def generate_ecr_image_from_docker_image(	1✔
189	ecr_client: "ECRClient", repository_name: str, image_name: str, platform: str = "linux/amd64"
190	):
191	"""
192	Parameters
193	----------
194	ecr_client
195	repository_name
196	image_name
197	platform
198
199	Returns
200	-------
201
202	"""
203
204	DOCKER_CLIENT.pull_image(image_name, platform=platform)	×
205
206	repository_uri = ecr_client.create_repository(	×
207	repositoryName=repository_name,
208	)["repository"]["repositoryUri"]
209
210	auth_response = ecr_client.get_authorization_token()	×
211	auth_token = auth_response["authorizationData"][0]["authorizationToken"].encode()	×
212	username, password = base64.b64decode(auth_token).decode().split(":")	×
213	registry = auth_response["authorizationData"][0]["proxyEndpoint"]	×
214	DOCKER_CLIENT.login(username, password, registry=registry)	×
215
216	DOCKER_CLIENT.tag_image(image_name, repository_uri)	×
217	DOCKER_CLIENT.push_image(repository_uri)	×
218
219
220	def _upload_to_s3(s3_client: "S3Client", bucket_name: str, key_name: str, file: str):	1✔
221	try:	1✔
222	create_s3_bucket(bucket_name, s3_client)	1✔
UNCOV 223	except ClientError as exc:	×
224	# when creating an already existing bucket, regions differ in their behavior:
225	# us-east-1 will silently pass (idempotent)
226	# any other region will return a `BucketAlreadyOwnedByYou` exception.
UNCOV 227	if exc.response["Error"]["Code"] != "BucketAlreadyOwnedByYou":	×
228	raise exc	×
229	s3_client.upload_file(Filename=file, Bucket=bucket_name, Key=key_name)	1✔

localstack / localstack / 6815841a-f2d0-4f5f-8a01-53ddff0de095

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous