17265699519

Committed 27 Aug 2025 11:28AM UTC coverage: 86.827% (-0.01%) from 86.837%

Build # 17265699519

Build Type

push

github

Committed by

web-flow

Commit Message

Fix SQS tests failing due to missing snapshot update after #12957 (#13062)

Coverage Stats

67057 of 77231 relevant lines covered (86.83%)

0.87 hits per line

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

71.57

/localstack-core/localstack/utils/files.py

import configparser
import inspect
import logging
import os
import shutil
import stat
import tempfile
from pathlib import Path

LOG = logging.getLogger(__name__)
TMP_FILES = []


def parse_config_file(file_or_str: str, single_section: bool = True) -> dict:
    """Parse the given properties config file/string and return a dict of section->key->value.
    If the config contains a single section, and 'single_section' is True, returns"""

    config = configparser.RawConfigParser()

    if os.path.exists(file_or_str):
        file_or_str = load_file(file_or_str)

    try:
        config.read_string(file_or_str)
    except configparser.MissingSectionHeaderError:
        file_or_str = f"[default]\n{file_or_str}"
        config.read_string(file_or_str)

    sections = list(config.sections())

    result = {sec: dict(config.items(sec)) for sec in sections}
    if len(sections) == 1 and single_section:
        result = result[sections[0]]

    return result


def get_user_cache_dir() -> Path:
    """
    Returns the path of the user's cache dir (e.g., ~/.cache on Linux, or ~/Library/Caches on Mac).

    :return: a Path pointing to the platform-specific cache dir of the user
    """
    from localstack.utils.platform import is_linux, is_mac_os, is_windows

    if is_windows():
        return Path(os.path.expandvars(r"%LOCALAPPDATA%\cache"))
    if is_mac_os():
        return Path.home() / "Library" / "Caches"
    if is_linux():
        string_path = os.environ.get("XDG_CACHE_HOME")
        if string_path and os.path.isabs(string_path):
            return Path(string_path)
    # Use the common place to store caches in Linux as a default
    return Path.home() / ".cache"


def cache_dir() -> Path:
    """
    Returns the cache dir for localstack (e.g., ~/.cache/localstack)

    :return: a Path pointing to the localstack cache dir
    """
    return get_user_cache_dir() / "localstack"


def save_file(file, content, append=False, permissions=None):
    mode = "a" if append else "w+"
    if not isinstance(content, str):
        mode = mode + "b"

    def _opener(path, flags):
        return os.open(path, flags, permissions)

    # make sure that the parent dir exists
    mkdir(os.path.dirname(file))
    # store file contents
    with open(file, mode, opener=_opener if permissions else None) as f:
        f.write(content)
        f.flush()


def load_file(file_path: str, default=None, mode=None):
    if not os.path.isfile(file_path):
        return default
    if not mode:
        mode = "r"
    with open(file_path, mode) as f:
        result = f.read()
    return result


def get_or_create_file(file_path, content=None, permissions=None):
    if os.path.exists(file_path):
        return load_file(file_path)
    content = "{}" if content is None else content
    try:
        save_file(file_path, content, permissions=permissions)
        return content
    except Exception:
        pass


def replace_in_file(search, replace, file_path):
    """Replace all occurrences of `search` with `replace` in the given file (overwrites in place!)"""
    content = load_file(file_path) or ""
    content_new = content.replace(search, replace)
    if content != content_new:
        save_file(file_path, content_new)


def mkdir(folder: str):
    if not os.path.exists(folder):
        os.makedirs(folder, exist_ok=True)


def is_empty_dir(directory: str, ignore_hidden: bool = False) -> bool:
    """Return whether the given directory contains any entries (files/folders), including hidden
    entries whose name starts with a dot (.), unless ignore_hidden=True is passed."""
    if not os.path.isdir(directory):
        raise Exception(f"Path is not a directory: {directory}")
    entries = os.listdir(directory)
    if ignore_hidden:
        entries = [e for e in entries if not e.startswith(".")]
    return not bool(entries)


def ensure_readable(file_path: str, default_perms: int = None):
    if default_perms is None:
        default_perms = 0o644
    try:
        with open(file_path, "rb"):
            pass
    except Exception:
        LOG.info("Updating permissions as file is currently not readable: %s", file_path)
        os.chmod(file_path, default_perms)


def chown_r(path: str, user: str):
    """Recursive chown on the given file/directory path."""
    # keep these imports here for Windows compatibility
    import grp
    import pwd

    uid = pwd.getpwnam(user).pw_uid
    gid = grp.getgrnam(user).gr_gid
    os.chown(path, uid, gid)
    for root, dirs, files in os.walk(path):
        for dirname in dirs:
            os.chown(os.path.join(root, dirname), uid, gid)
        for filename in files:
            os.chown(os.path.join(root, filename), uid, gid)


def chmod_r(path: str, mode: int):
    """
    Recursive chmod
    :param path: path to file or directory
    :param mode: permission mask as octal integer value
    """
    if not os.path.exists(path):
        return
    idempotent_chmod(path, mode)
    for root, dirnames, filenames in os.walk(path):
        for dirname in dirnames:
            idempotent_chmod(os.path.join(root, dirname), mode)
        for filename in filenames:
            idempotent_chmod(os.path.join(root, filename), mode)


def idempotent_chmod(path: str, mode: int):
    """
    Perform idempotent chmod on the given file path (non-recursively). The function attempts to call `os.chmod`, and
    will catch and only re-raise exceptions (e.g., PermissionError) if the file does not have the given mode already.
    :param path: path to file
    :param mode: permission mask as octal integer value
    """
    try:
        os.chmod(path, mode)
    except Exception:
        try:
            existing_mode = os.stat(path)
        except FileNotFoundError:
            # file deleted in the meantime, or otherwise not accessible (socket)
            return
        if mode in (existing_mode.st_mode, stat.S_IMODE(existing_mode.st_mode)):
            # file already has the desired permissions -> return
            return
        raise


def rm_rf(path: str):
    """
    Recursively removes a file or directory
    """
    from localstack.utils.platform import is_debian
    from localstack.utils.run import run

    if not path or not os.path.exists(path):
        return
    # Running the native command can be an order of magnitude faster in Alpine on Travis-CI
    if is_debian():
        try:
            return run(f'rm -rf "{path}"')
        except Exception:
            pass
    # Make sure all files are writeable and dirs executable to remove
    try:
        chmod_r(path, 0o777)
    except PermissionError:
        pass  # todo log
    # check if the file is either a normal file, or, e.g., a fifo
    exists_but_non_dir = os.path.exists(path) and not os.path.isdir(path)
    if os.path.isfile(path) or exists_but_non_dir:
        os.remove(path)
    else:
        shutil.rmtree(path)


def cp_r(src: str, dst: str, rm_dest_on_conflict=False, ignore_copystat_errors=False, **kwargs):
    """Recursively copies file/directory"""
    # attention: this patch is not threadsafe
    copystat_orig = shutil.copystat
    if ignore_copystat_errors:

        def _copystat(*args, **kwargs):
            try:
                return copystat_orig(*args, **kwargs)
            except Exception:
                pass

        shutil.copystat = _copystat
    try:
        if os.path.isfile(src):
            if os.path.isdir(dst):
                dst = os.path.join(dst, os.path.basename(src))
            return shutil.copyfile(src, dst)
        if "dirs_exist_ok" in inspect.getfullargspec(shutil.copytree).args:
            kwargs["dirs_exist_ok"] = True
        try:
            return shutil.copytree(src, dst, **kwargs)
        except FileExistsError:
            if rm_dest_on_conflict:
                rm_rf(dst)
                return shutil.copytree(src, dst, **kwargs)
            raise
    except Exception as e:

        def _info(_path):
            return f"{_path} (file={os.path.isfile(_path)}, symlink={os.path.islink(_path)})"

        LOG.debug("Error copying files from %s to %s: %s", _info(src), _info(dst), e)
        raise
    finally:
        shutil.copystat = copystat_orig


def disk_usage(path: str) -> int:
    """Return the disk usage of the given file or directory."""

    if not os.path.exists(path):
        return 0

    if os.path.isfile(path):
        return os.path.getsize(path)

    total_size = 0
    for dirpath, dirnames, filenames in os.walk(path):
        for f in filenames:
            fp = os.path.join(dirpath, f)
            # skip if it is symbolic link
            if not os.path.islink(fp):
                total_size += os.path.getsize(fp)
    return total_size


def file_exists_not_empty(path: str) -> bool:
    """Return whether the given file or directory exists and is non-empty (i.e., >0 bytes content)"""
    return path and disk_usage(path) > 0


def cleanup_tmp_files():
    for tmp in TMP_FILES:
        try:
            rm_rf(tmp)
        except Exception:
            pass  # file likely doesn't exist, or permission denied
    del TMP_FILES[:]


def new_tmp_file(suffix: str = None, dir: str = None) -> str:
    """Return a path to a new temporary file."""
    tmp_file, tmp_path = tempfile.mkstemp(suffix=suffix, dir=dir)
    os.close(tmp_file)
    TMP_FILES.append(tmp_path)
    return tmp_path


def new_tmp_dir(dir: str = None):
    folder = new_tmp_file(dir=dir)
    rm_rf(folder)
    mkdir(folder)
    return folder

1	import configparser	1✔
2	import inspect	1✔
3	import logging	1✔
4	import os	1✔
5	import shutil	1✔
6	import stat	1✔
7	import tempfile	1✔
8	from pathlib import Path	1✔
9
10	LOG = logging.getLogger(__name__)	1✔
11	TMP_FILES = []	1✔
12
13
14	def parse_config_file(file_or_str: str, single_section: bool = True) -> dict:	1✔
15	"""Parse the given properties config file/string and return a dict of section->key->value.
16	If the config contains a single section, and 'single_section' is True, returns"""
17
18	config = configparser.RawConfigParser()	1✔
19
20	if os.path.exists(file_or_str):	1✔
21	file_or_str = load_file(file_or_str)	1✔
22
23	try:	1✔
24	config.read_string(file_or_str)	1✔
25	except configparser.MissingSectionHeaderError:	1✔
26	file_or_str = f"[default]\n{file_or_str}"	1✔
27	config.read_string(file_or_str)	1✔
28
29	sections = list(config.sections())	1✔
30
31	result = {sec: dict(config.items(sec)) for sec in sections}	1✔
32	if len(sections) == 1 and single_section:	1✔
33	result = result[sections[0]]	1✔
34
35	return result	1✔
36
37
38	def get_user_cache_dir() -> Path:	1✔
39	"""
40	Returns the path of the user's cache dir (e.g., ~/.cache on Linux, or ~/Library/Caches on Mac).
41
42	:return: a Path pointing to the platform-specific cache dir of the user
43	"""
44	from localstack.utils.platform import is_linux, is_mac_os, is_windows	1✔
45
46	if is_windows():	1✔
47	return Path(os.path.expandvars(r"%LOCALAPPDATA%\cache"))	×
48	if is_mac_os():	1✔
49	return Path.home() / "Library" / "Caches"	×
50	if is_linux():	1✔
51	string_path = os.environ.get("XDG_CACHE_HOME")	1✔
52	if string_path and os.path.isabs(string_path):	1✔
53	return Path(string_path)	×
54	# Use the common place to store caches in Linux as a default
55	return Path.home() / ".cache"	1✔
56
57
58	def cache_dir() -> Path:	1✔
59	"""
60	Returns the cache dir for localstack (e.g., ~/.cache/localstack)
61
62	:return: a Path pointing to the localstack cache dir
63	"""
64	return get_user_cache_dir() / "localstack"	1✔
65
66
67	def save_file(file, content, append=False, permissions=None):	1✔
68	mode = "a" if append else "w+"	1✔
69	if not isinstance(content, str):	1✔
70	mode = mode + "b"	1✔
71
72	def _opener(path, flags):	1✔
73	return os.open(path, flags, permissions)	1✔
74
75	# make sure that the parent dir exists
76	mkdir(os.path.dirname(file))	1✔
77	# store file contents
78	with open(file, mode, opener=_opener if permissions else None) as f:	1✔
79	f.write(content)	1✔
80	f.flush()	1✔
81
82
83	def load_file(file_path: str, default=None, mode=None):	1✔
84	if not os.path.isfile(file_path):	1✔
85	return default	1✔
86	if not mode:	1✔
87	mode = "r"	1✔
88	with open(file_path, mode) as f:	1✔
89	result = f.read()	1✔
90	return result	1✔
91
92
93	def get_or_create_file(file_path, content=None, permissions=None):	1✔
94	if os.path.exists(file_path):	×
95	return load_file(file_path)	×
96	content = "{}" if content is None else content	×
97	try:	×
98	save_file(file_path, content, permissions=permissions)	×
99	return content	×
100	except Exception:	×
101	pass	×
102
103
104	def replace_in_file(search, replace, file_path):	1✔
105	"""Replace all occurrences of `search` with `replace` in the given file (overwrites in place!)"""
106	content = load_file(file_path) or ""	1✔
107	content_new = content.replace(search, replace)	1✔
108	if content != content_new:	1✔
109	save_file(file_path, content_new)	1✔
110
111
112	def mkdir(folder: str):	1✔
113	if not os.path.exists(folder):	1✔
114	os.makedirs(folder, exist_ok=True)	1✔
115
116
117	def is_empty_dir(directory: str, ignore_hidden: bool = False) -> bool:	1✔
118	"""Return whether the given directory contains any entries (files/folders), including hidden
119	entries whose name starts with a dot (.), unless ignore_hidden=True is passed."""
120	if not os.path.isdir(directory):	1✔
121	raise Exception(f"Path is not a directory: {directory}")	×
122	entries = os.listdir(directory)	1✔
123	if ignore_hidden:	1✔
124	entries = [e for e in entries if not e.startswith(".")]	1✔
125	return not bool(entries)	1✔
126
127
128	def ensure_readable(file_path: str, default_perms: int = None):	1✔
129	if default_perms is None:	×
130	default_perms = 0o644	×
131	try:	×
132	with open(file_path, "rb"):	×
133	pass	×
134	except Exception:	×
135	LOG.info("Updating permissions as file is currently not readable: %s", file_path)	×
136	os.chmod(file_path, default_perms)	×
137
138
139	def chown_r(path: str, user: str):	1✔
140	"""Recursive chown on the given file/directory path."""
141	# keep these imports here for Windows compatibility
142	import grp	1✔
143	import pwd	1✔
144
145	uid = pwd.getpwnam(user).pw_uid	1✔
146	gid = grp.getgrnam(user).gr_gid	1✔
147	os.chown(path, uid, gid)	1✔
148	for root, dirs, files in os.walk(path):	1✔
149	for dirname in dirs:	1✔
150	os.chown(os.path.join(root, dirname), uid, gid)	1✔
151	for filename in files:	1✔
152	os.chown(os.path.join(root, filename), uid, gid)	1✔
153
154
155	def chmod_r(path: str, mode: int):	1✔
156	"""
157	Recursive chmod
158	:param path: path to file or directory
159	:param mode: permission mask as octal integer value
160	"""
161	if not os.path.exists(path):	1✔
162	return	×
163	idempotent_chmod(path, mode)	1✔
164	for root, dirnames, filenames in os.walk(path):	1✔
165	for dirname in dirnames:	1✔
166	idempotent_chmod(os.path.join(root, dirname), mode)	1✔
167	for filename in filenames:	1✔
168	idempotent_chmod(os.path.join(root, filename), mode)	1✔
169
170
171	def idempotent_chmod(path: str, mode: int):	1✔
172	"""
173	Perform idempotent chmod on the given file path (non-recursively). The function attempts to call `os.chmod`, and
174	will catch and only re-raise exceptions (e.g., PermissionError) if the file does not have the given mode already.
175	:param path: path to file
176	:param mode: permission mask as octal integer value
177	"""
178	try:	1✔
179	os.chmod(path, mode)	1✔
180	except Exception:	×
181	try:	×
182	existing_mode = os.stat(path)	×
183	except FileNotFoundError:	×
184	# file deleted in the meantime, or otherwise not accessible (socket)
185	return	×
186	if mode in (existing_mode.st_mode, stat.S_IMODE(existing_mode.st_mode)):	×
187	# file already has the desired permissions -> return
188	return	×
189	raise	×
190
191
192	def rm_rf(path: str):	1✔
193	"""
194	Recursively removes a file or directory
195	"""
196	from localstack.utils.platform import is_debian	1✔
197	from localstack.utils.run import run	1✔
198
199	if not path or not os.path.exists(path):	1✔
200	return	1✔
201	# Running the native command can be an order of magnitude faster in Alpine on Travis-CI
202	if is_debian():	1✔
203	try:	1✔
204	return run(f'rm -rf "{path}"')	1✔
205	except Exception:	×
206	pass	×
207	# Make sure all files are writeable and dirs executable to remove
208	try:	1✔
209	chmod_r(path, 0o777)	1✔
210	except PermissionError:	×
211	pass # todo log	×
212	# check if the file is either a normal file, or, e.g., a fifo
213	exists_but_non_dir = os.path.exists(path) and not os.path.isdir(path)	1✔
214	if os.path.isfile(path) or exists_but_non_dir:	1✔
215	os.remove(path)	1✔
216	else:
217	shutil.rmtree(path)	1✔
218
219
220	def cp_r(src: str, dst: str, rm_dest_on_conflict=False, ignore_copystat_errors=False, **kwargs):	1✔
221	"""Recursively copies file/directory"""
222	# attention: this patch is not threadsafe
223	copystat_orig = shutil.copystat	1✔
224	if ignore_copystat_errors:	1✔
225
226	def _copystat(args, *kwargs):	×
227	try:	×
228	return copystat_orig(args, *kwargs)	×
229	except Exception:	×
230	pass	×
231
232	shutil.copystat = _copystat	×
233	try:	1✔
234	if os.path.isfile(src):	1✔
235	if os.path.isdir(dst):	1✔
236	dst = os.path.join(dst, os.path.basename(src))	×
237	return shutil.copyfile(src, dst)	1✔
238	if "dirs_exist_ok" in inspect.getfullargspec(shutil.copytree).args:	1✔
239	kwargs["dirs_exist_ok"] = True	1✔
240	try:	1✔
241	return shutil.copytree(src, dst, **kwargs)	1✔
242	except FileExistsError:	×
243	if rm_dest_on_conflict:	×
244	rm_rf(dst)	×
245	return shutil.copytree(src, dst, **kwargs)	×
246	raise	×
247	except Exception as e:	×
248
249	def _info(_path):	×
250	return f"{_path} (file={os.path.isfile(_path)}, symlink={os.path.islink(_path)})"	×
251
252	LOG.debug("Error copying files from %s to %s: %s", _info(src), _info(dst), e)	×
253	raise	×
254	finally:
255	shutil.copystat = copystat_orig	1✔
256
257
258	def disk_usage(path: str) -> int:	1✔
259	"""Return the disk usage of the given file or directory."""
260
261	if not os.path.exists(path):	1✔
262	return 0	1✔
263
264	if os.path.isfile(path):	1✔
265	return os.path.getsize(path)	1✔
266
267	total_size = 0	1✔
268	for dirpath, dirnames, filenames in os.walk(path):	1✔
269	for f in filenames:	1✔
270	fp = os.path.join(dirpath, f)	1✔
271	# skip if it is symbolic link
272	if not os.path.islink(fp):	1✔
273	total_size += os.path.getsize(fp)	1✔
274	return total_size	1✔
275
276
277	def file_exists_not_empty(path: str) -> bool:	1✔
278	"""Return whether the given file or directory exists and is non-empty (i.e., >0 bytes content)"""
279	return path and disk_usage(path) > 0	1✔
280
281
282	def cleanup_tmp_files():	1✔
283	for tmp in TMP_FILES:	×
284	try:	×
285	rm_rf(tmp)	×
286	except Exception:	×
287	pass # file likely doesn't exist, or permission denied	×
288	del TMP_FILES[:]	×
289
290
291	def new_tmp_file(suffix: str = None, dir: str = None) -> str:	1✔
292	"""Return a path to a new temporary file."""
293	tmp_file, tmp_path = tempfile.mkstemp(suffix=suffix, dir=dir)	1✔
294	os.close(tmp_file)	1✔
295	TMP_FILES.append(tmp_path)	1✔
296	return tmp_path	1✔
297
298
299	def new_tmp_dir(dir: str = None):	1✔
300	folder = new_tmp_file(dir=dir)	1✔
301	rm_rf(folder)	1✔
302	mkdir(folder)	1✔
303	return folder	1✔

localstack / localstack / 17265699519

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