a53c0273-9548-479e-ab3c-f3af40c0e980

Committed 13 May 2025 05:31PM UTC coverage: 86.624% (-0.03%) from 86.658%

Build # a53c0273-9548-479e-ab3c-f3af40c0e980

Build Type

push

circleci

Committed by

web-flow

Commit Message

ASF: Mark optional params as such (X | None) (#12614)

Coverage Stats

5 of 7 new or added lines in 2 files covered. (71.43%)

34 existing lines in 16 files now uncovered.

64347 of 74283 relevant lines covered (86.62%)

0.87 hits per line

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

79.41

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

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

localstack / localstack / a53c0273-9548-479e-ab3c-f3af40c0e980

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