4e27dc30-df7d-47cf-9ddb-0b539d612501

Committed 17 Apr 2025 08:11PM UTC coverage: 86.279% (-0.02%) from 86.294%

Build # 4e27dc30-df7d-47cf-9ddb-0b539d612501

Build Type

push

circleci

Committed by

web-flow

Commit Message

Step Functions: Surface Support for Mocked Responses (#12525)

Coverage Stats

200 of 245 new or added lines in 9 files covered. (81.63%)

201 existing lines in 15 files now uncovered.

63889 of 74049 relevant lines covered (86.28%)

0.86 hits per line

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

78.72

/localstack-core/localstack/utils/archives.py

import io
import tarfile
import zipfile
from subprocess import Popen
from typing import IO, Optional

try:
    from typing import Literal
except ImportError:
    from typing_extensions import Literal

import glob
import logging
import os
import re
import tempfile
import time
from typing import Union

from localstack.constants import MAVEN_REPO_URL
from localstack.utils.files import load_file, mkdir, new_tmp_file, rm_rf, save_file
from localstack.utils.http import download
from localstack.utils.run import run

from .run import is_command_available
from .strings import truncate

LOG = logging.getLogger(__name__)


StrPath = Union[str, os.PathLike]


def is_zip_file(content):
    stream = io.BytesIO(content)
    return zipfile.is_zipfile(stream)


def get_unzipped_size(zip_file: Union[str, IO[bytes]]):
    """Returns the size of the unzipped file."""
    with zipfile.ZipFile(zip_file, "r") as zip_ref:
        return sum(f.file_size for f in zip_ref.infolist())


def unzip(path: str, target_dir: str, overwrite: bool = True) -> Optional[Union[str, Popen]]:
    from localstack.utils.platform import is_debian

    use_native_cmd = is_debian() or is_command_available("unzip")
    if use_native_cmd:
        # Running the native command can be an order of magnitude faster in the container. Also, `unzip`
        #  is capable of extracting zip files with incorrect CRC codes (sometimes happens, e.g., with some
        #  Node.js/Serverless versions), which can fail with Python's `zipfile` (extracting empty files).
        flags = ["-o"] if overwrite else []
        flags += ["-q"]
        try:
            cmd = ["unzip"] + flags + [path]
            return run(cmd, cwd=target_dir, print_error=False)
        except Exception as e:
            error_str = truncate(str(e), max_length=200)
            LOG.info(
                'Unable to use native "unzip" command (using fallback mechanism): %s', error_str
            )

    try:
        zip_ref = zipfile.ZipFile(path, "r")
    except Exception as e:
        LOG.warning("Unable to open zip file: %s: %s", path, e)
        raise e

    def _unzip_file_entry(zip_ref, file_entry, target_dir):
        """Extracts a Zipfile entry and preserves permissions"""
        out_path = os.path.join(target_dir, file_entry.filename)
        if use_native_cmd and os.path.exists(out_path) and os.path.getsize(out_path) > 0:
            # this can happen under certain circumstances if the native "unzip" command
            # fails with a non-zero exit code, yet manages to extract parts of the zip file
            return
        zip_ref.extract(file_entry.filename, path=target_dir)
        perm = file_entry.external_attr >> 16
        # Make sure to preserve file permissions in the zip file
        # https://www.burgundywall.com/post/preserving-file-perms-with-python-zipfile-module
        os.chmod(out_path, perm or 0o777)

    try:
        for file_entry in zip_ref.infolist():
            _unzip_file_entry(zip_ref, file_entry, target_dir)
    finally:
        zip_ref.close()


def untar(path: str, target_dir: str):
    mode = "r:gz" if path.endswith("gz") else "r"
    with tarfile.open(path, mode) as tar:
        tar.extractall(path=target_dir)


def create_zip_file_cli(source_path: StrPath, base_dir: StrPath, zip_file: StrPath):
    """
    Creates a zip archive by using the native zip command. The native command can be an order of magnitude faster in CI
    """
    source = "." if source_path == base_dir else os.path.basename(source_path)
    run(["zip", "-r", zip_file, source], cwd=base_dir)


def create_zip_file_python(
    base_dir: StrPath,
    zip_file: StrPath,
    mode: Literal["r", "w", "x", "a"] = "w",
    content_root: Optional[str] = None,
):
    with zipfile.ZipFile(zip_file, mode) as zip_file:
        for root, dirs, files in os.walk(base_dir):
            for name in files:
                full_name = os.path.join(root, name)
                relative = os.path.relpath(root, start=base_dir)
                if content_root:
                    dest = os.path.join(content_root, relative, name)
                else:
                    dest = os.path.join(relative, name)
                zip_file.write(full_name, dest)


def add_file_to_jar(class_file, class_url, target_jar, base_dir=None):
    base_dir = base_dir or os.path.dirname(target_jar)
    patch_class_file = os.path.join(base_dir, class_file)
    if not os.path.exists(patch_class_file):
        download(class_url, patch_class_file)
        run(["zip", target_jar, class_file], cwd=base_dir)


def update_jar_manifest(
    jar_file_name: str, parent_dir: str, search: Union[str, re.Pattern], replace: str
):
    manifest_file_path = "META-INF/MANIFEST.MF"
    jar_path = os.path.join(parent_dir, jar_file_name)
    with tempfile.TemporaryDirectory() as tmp_dir:
        tmp_manifest_file = os.path.join(tmp_dir, manifest_file_path)
        run(["unzip", "-o", jar_path, manifest_file_path], cwd=tmp_dir)
        manifest = load_file(tmp_manifest_file)

    # return if the search pattern does not match (for idempotence, to avoid file permission issues further below)
    if isinstance(search, re.Pattern):
        if not search.search(manifest):
            return
        manifest = search.sub(replace, manifest, 1)
    else:
        if search not in manifest:
            return
        manifest = manifest.replace(search, replace, 1)

    manifest_file = os.path.join(parent_dir, manifest_file_path)
    save_file(manifest_file, manifest)
    run(["zip", jar_file_name, manifest_file_path], cwd=parent_dir)


def upgrade_jar_file(base_dir: str, file_glob: str, maven_asset: str):
    """
    Upgrade the matching Java JAR file in a local directory with the given Maven asset
    :param base_dir: base directory to search the JAR file to replace in
    :param file_glob: glob pattern for the JAR file to replace
    :param maven_asset: name of Maven asset to download, in the form "<qualified_name>:<version>"
    """

    local_path = os.path.join(base_dir, file_glob)
    parent_dir = os.path.dirname(local_path)
    maven_asset = maven_asset.replace(":", "/")
    parts = maven_asset.split("/")
    maven_asset_url = f"{MAVEN_REPO_URL}/{maven_asset}/{parts[-2]}-{parts[-1]}.jar"
    target_file = os.path.join(parent_dir, os.path.basename(maven_asset_url))
    if os.path.exists(target_file):
        # avoid re-downloading the newer JAR version if it already exists locally
        return
    matches = glob.glob(local_path)
    if not matches:
        return
    for match in matches:
        os.remove(match)
    download(maven_asset_url, target_file)


def download_and_extract(archive_url, target_dir, retries=0, sleep=3, tmp_archive=None):
    mkdir(target_dir)

    _, ext = os.path.splitext(tmp_archive or archive_url)
    tmp_archive = tmp_archive or new_tmp_file()
    if not os.path.exists(tmp_archive) or os.path.getsize(tmp_archive) <= 0:
        # create temporary placeholder file, to avoid duplicate parallel downloads
        save_file(tmp_archive, "")

        for i in range(retries + 1):
            try:
                download(archive_url, tmp_archive)
                break
            except Exception as e:
                LOG.warning(
                    "Attempt %d. Failed to download archive from %s: %s",
                    i + 1,
                    archive_url,
                    e,
                )
                # only sleep between retries, not after the last one
                if i < retries:
                    time.sleep(sleep)

    # if the temporary file we created above hasn't been replaced, we assume failure
    if os.path.getsize(tmp_archive) <= 0:
        raise Exception("Failed to download archive from %s: . Retries exhausted", archive_url)

    if ext == ".zip":
        unzip(tmp_archive, target_dir)
    elif ext in (
        ".bz2",
        ".gz",
        ".tgz",
        ".xz",
    ):
        untar(tmp_archive, target_dir)
    else:
        raise Exception(f"Unsupported archive format: {ext}")


def download_and_extract_with_retry(archive_url, tmp_archive, target_dir):
    try:
        download_and_extract(archive_url, target_dir, tmp_archive=tmp_archive)
    except Exception as e:
        # try deleting and re-downloading the zip file
        LOG.info("Unable to extract file, re-downloading ZIP archive %s: %s", tmp_archive, e)
        rm_rf(tmp_archive)
        download_and_extract(archive_url, target_dir, tmp_archive=tmp_archive)

1	import io	1✔
2	import tarfile	1✔
3	import zipfile	1✔
4	from subprocess import Popen	1✔
5	from typing import IO, Optional	1✔
6
7	try:	1✔
8	from typing import Literal	1✔
9	except ImportError:	×
10	from typing_extensions import Literal	×
11
12	import glob	1✔
13	import logging	1✔
14	import os	1✔
15	import re	1✔
16	import tempfile	1✔
17	import time	1✔
18	from typing import Union	1✔
19
20	from localstack.constants import MAVEN_REPO_URL	1✔
21	from localstack.utils.files import load_file, mkdir, new_tmp_file, rm_rf, save_file	1✔
22	from localstack.utils.http import download	1✔
23	from localstack.utils.run import run	1✔
24
25	from .run import is_command_available	1✔
26	from .strings import truncate	1✔
27
28	LOG = logging.getLogger(__name__)	1✔
29
30
31	StrPath = Union[str, os.PathLike]	1✔
32
33
34	def is_zip_file(content):	1✔
35	stream = io.BytesIO(content)	1✔
36	return zipfile.is_zipfile(stream)	1✔
37
38
39	def get_unzipped_size(zip_file: Union[str, IO[bytes]]):	1✔
40	"""Returns the size of the unzipped file."""
41	with zipfile.ZipFile(zip_file, "r") as zip_ref:	1✔
42	return sum(f.file_size for f in zip_ref.infolist())	1✔
43
44
45	def unzip(path: str, target_dir: str, overwrite: bool = True) -> Optional[Union[str, Popen]]:	1✔
46	from localstack.utils.platform import is_debian	1✔
47
48	use_native_cmd = is_debian() or is_command_available("unzip")	1✔
49	if use_native_cmd:	1✔
50	# Running the native command can be an order of magnitude faster in the container. Also, `unzip`
51	# is capable of extracting zip files with incorrect CRC codes (sometimes happens, e.g., with some
52	# Node.js/Serverless versions), which can fail with Python's `zipfile` (extracting empty files).
53	flags = ["-o"] if overwrite else []	1✔
54	flags += ["-q"]	1✔
55	try:	1✔
56	cmd = ["unzip"] + flags + [path]	1✔
57	return run(cmd, cwd=target_dir, print_error=False)	1✔
58	except Exception as e:	1✔
59	error_str = truncate(str(e), max_length=200)	1✔
60	LOG.info(	1✔
61	'Unable to use native "unzip" command (using fallback mechanism): %s', error_str
62	)
63
64	try:	1✔
65	zip_ref = zipfile.ZipFile(path, "r")	1✔
66	except Exception as e:	×
67	LOG.warning("Unable to open zip file: %s: %s", path, e)	×
68	raise e	×
69
70	def _unzip_file_entry(zip_ref, file_entry, target_dir):	1✔
71	"""Extracts a Zipfile entry and preserves permissions"""
72	out_path = os.path.join(target_dir, file_entry.filename)	1✔
73	if use_native_cmd and os.path.exists(out_path) and os.path.getsize(out_path) > 0:	1✔
74	# this can happen under certain circumstances if the native "unzip" command
75	# fails with a non-zero exit code, yet manages to extract parts of the zip file
76	return	1✔
UNCOV 77	zip_ref.extract(file_entry.filename, path=target_dir)	×
UNCOV 78	perm = file_entry.external_attr >> 16	×
79	# Make sure to preserve file permissions in the zip file
80	# https://www.burgundywall.com/post/preserving-file-perms-with-python-zipfile-module
UNCOV 81	os.chmod(out_path, perm or 0o777)	×
82
83	try:	1✔
84	for file_entry in zip_ref.infolist():	1✔
85	_unzip_file_entry(zip_ref, file_entry, target_dir)	1✔
86	finally:
87	zip_ref.close()	1✔
88
89
90	def untar(path: str, target_dir: str):	1✔
91	mode = "r:gz" if path.endswith("gz") else "r"	1✔
92	with tarfile.open(path, mode) as tar:	1✔
93	tar.extractall(path=target_dir)	1✔
94
95
96	def create_zip_file_cli(source_path: StrPath, base_dir: StrPath, zip_file: StrPath):	1✔
97	"""
98	Creates a zip archive by using the native zip command. The native command can be an order of magnitude faster in CI
99	"""
100	source = "." if source_path == base_dir else os.path.basename(source_path)	1✔
101	run(["zip", "-r", zip_file, source], cwd=base_dir)	1✔
102
103
104	def create_zip_file_python(	1✔
105	base_dir: StrPath,
106	zip_file: StrPath,
107	mode: Literal["r", "w", "x", "a"] = "w",
108	content_root: Optional[str] = None,
109	):
110	with zipfile.ZipFile(zip_file, mode) as zip_file:	1✔
111	for root, dirs, files in os.walk(base_dir):	1✔
112	for name in files:	1✔
113	full_name = os.path.join(root, name)	1✔
114	relative = os.path.relpath(root, start=base_dir)	1✔
115	if content_root:	1✔
116	dest = os.path.join(content_root, relative, name)	×
117	else:
118	dest = os.path.join(relative, name)	1✔
119	zip_file.write(full_name, dest)	1✔
120
121
122	def add_file_to_jar(class_file, class_url, target_jar, base_dir=None):	1✔
123	base_dir = base_dir or os.path.dirname(target_jar)	×
124	patch_class_file = os.path.join(base_dir, class_file)	×
125	if not os.path.exists(patch_class_file):	×
126	download(class_url, patch_class_file)	×
127	run(["zip", target_jar, class_file], cwd=base_dir)	×
128
129
130	def update_jar_manifest(	1✔
131	jar_file_name: str, parent_dir: str, search: Union[str, re.Pattern], replace: str
132	):
133	manifest_file_path = "META-INF/MANIFEST.MF"	1✔
134	jar_path = os.path.join(parent_dir, jar_file_name)	1✔
135	with tempfile.TemporaryDirectory() as tmp_dir:	1✔
136	tmp_manifest_file = os.path.join(tmp_dir, manifest_file_path)	1✔
137	run(["unzip", "-o", jar_path, manifest_file_path], cwd=tmp_dir)	1✔
138	manifest = load_file(tmp_manifest_file)	1✔
139
140	# return if the search pattern does not match (for idempotence, to avoid file permission issues further below)
141	if isinstance(search, re.Pattern):	1✔
142	if not search.search(manifest):	×
143	return	×
144	manifest = search.sub(replace, manifest, 1)	×
145	else:
146	if search not in manifest:	1✔
147	return	×
148	manifest = manifest.replace(search, replace, 1)	1✔
149
150	manifest_file = os.path.join(parent_dir, manifest_file_path)	1✔
151	save_file(manifest_file, manifest)	1✔
152	run(["zip", jar_file_name, manifest_file_path], cwd=parent_dir)	1✔
153
154
155	def upgrade_jar_file(base_dir: str, file_glob: str, maven_asset: str):	1✔
156	"""
157	Upgrade the matching Java JAR file in a local directory with the given Maven asset
158	:param base_dir: base directory to search the JAR file to replace in
159	:param file_glob: glob pattern for the JAR file to replace
160	:param maven_asset: name of Maven asset to download, in the form "<qualified_name>:<version>"
161	"""
162
163	local_path = os.path.join(base_dir, file_glob)	1✔
164	parent_dir = os.path.dirname(local_path)	1✔
165	maven_asset = maven_asset.replace(":", "/")	1✔
166	parts = maven_asset.split("/")	1✔
167	maven_asset_url = f"{MAVEN_REPO_URL}/{maven_asset}/{parts[-2]}-{parts[-1]}.jar"	1✔
168	target_file = os.path.join(parent_dir, os.path.basename(maven_asset_url))	1✔
169	if os.path.exists(target_file):	1✔
170	# avoid re-downloading the newer JAR version if it already exists locally
171	return	×
172	matches = glob.glob(local_path)	1✔
173	if not matches:	1✔
174	return	×
175	for match in matches:	1✔
176	os.remove(match)	1✔
177	download(maven_asset_url, target_file)	1✔
178
179
180	def download_and_extract(archive_url, target_dir, retries=0, sleep=3, tmp_archive=None):	1✔
181	mkdir(target_dir)	1✔
182
183	_, ext = os.path.splitext(tmp_archive or archive_url)	1✔
184	tmp_archive = tmp_archive or new_tmp_file()	1✔
185	if not os.path.exists(tmp_archive) or os.path.getsize(tmp_archive) <= 0:	1✔
186	# create temporary placeholder file, to avoid duplicate parallel downloads
187	save_file(tmp_archive, "")	1✔
188
189	for i in range(retries + 1):	1✔
190	try:	1✔
191	download(archive_url, tmp_archive)	1✔
192	break	1✔
193	except Exception as e:	×
194	LOG.warning(	×
195	"Attempt %d. Failed to download archive from %s: %s",
196	i + 1,
197	archive_url,
198	e,
199	)
200	# only sleep between retries, not after the last one
201	if i < retries:	×
202	time.sleep(sleep)	×
203
204	# if the temporary file we created above hasn't been replaced, we assume failure
205	if os.path.getsize(tmp_archive) <= 0:	1✔
206	raise Exception("Failed to download archive from %s: . Retries exhausted", archive_url)	×
207
208	if ext == ".zip":	1✔
209	unzip(tmp_archive, target_dir)	1✔
210	elif ext in (	1✔
211	".bz2",
212	".gz",
213	".tgz",
214	".xz",
215	):
216	untar(tmp_archive, target_dir)	1✔
217	else:
218	raise Exception(f"Unsupported archive format: {ext}")	×
219
220
221	def download_and_extract_with_retry(archive_url, tmp_archive, target_dir):	1✔
222	try:	1✔
223	download_and_extract(archive_url, target_dir, tmp_archive=tmp_archive)	1✔
224	except Exception as e:	×
225	# try deleting and re-downloading the zip file
226	LOG.info("Unable to extract file, re-downloading ZIP archive %s: %s", tmp_archive, e)	×
227	rm_rf(tmp_archive)	×
228	download_and_extract(archive_url, target_dir, tmp_archive=tmp_archive)	×

localstack / localstack / 4e27dc30-df7d-47cf-9ddb-0b539d612501

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