7ae8c08b-27e7-4df6-bfb9-4892ae974ff7

Committed 17 Mar 2025 11:44PM UTC coverage: 86.954% (+0.02%) from 86.93%

Build # 7ae8c08b-27e7-4df6-bfb9-4892ae974ff7

Build Type

push

circleci

Committed by

web-flow

Commit Message

SNS: fix Filter Policy engine to not evaluate full complex payload (#12395)

Run Details

8 of 8 new or added lines in 1 file covered. (100.0%)

27 existing lines in 8 files now uncovered.

62326 of 71677 relevant lines covered (86.95%)

0.87 hits per line

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

89.42

/localstack-core/localstack/packages/core.py

import logging
import os
import re
from abc import ABC
from functools import lru_cache
from sys import version_info
from typing import Optional, Tuple

import requests

from localstack import config

from ..constants import LOCALSTACK_VENV_FOLDER, MAVEN_REPO_URL
from ..utils.archives import download_and_extract
from ..utils.files import chmod_r, chown_r, mkdir, rm_rf
from ..utils.http import download
from ..utils.run import is_root, run
from ..utils.venv import VirtualEnvironment
from .api import InstallTarget, PackageException, PackageInstaller

LOG = logging.getLogger(__name__)


class SystemNotSupportedException(PackageException):
    """Exception indicating that the current system is not allowed."""

    pass


class ExecutableInstaller(PackageInstaller, ABC):
    """
    This installer simply adds a clean interface for accessing a downloaded executable directly
    """

    def get_executable_path(self) -> str | None:
        """
        :return: the path to the downloaded binary or None if it's not yet downloaded / installed.
        """
        install_dir = self.get_installed_dir()
        if install_dir:
            return self._get_install_marker_path(install_dir)


class DownloadInstaller(ExecutableInstaller):
    def __init__(self, name: str, version: str):
        super().__init__(name, version)

    def _get_download_url(self) -> str:
        raise NotImplementedError()

    def _get_install_marker_path(self, install_dir: str) -> str:
        url = self._get_download_url()
        binary_name = os.path.basename(url)
        return os.path.join(install_dir, binary_name)

    def _install(self, target: InstallTarget) -> None:
        target_directory = self._get_install_dir(target)
        mkdir(target_directory)
        download_url = self._get_download_url()
        target_path = self._get_install_marker_path(target_directory)
        download(download_url, target_path)


class ArchiveDownloadAndExtractInstaller(ExecutableInstaller):
    def __init__(self, name: str, version: str, extract_single_directory: bool = False):
        """
        :param name: technical package name, f.e. "opensearch"
        :param version: version of the package to install
        :param extract_single_directory: whether to extract files from single root folder in the archive
        """
        super().__init__(name, version)
        self.extract_single_directory = extract_single_directory

    def _get_install_marker_path(self, install_dir: str) -> str:
        raise NotImplementedError()

    def _get_download_url(self) -> str:
        raise NotImplementedError()

    def get_installed_dir(self) -> str | None:
        installed_dir = super().get_installed_dir()
        subdir = self._get_archive_subdir()

        # If the specific installer defines a subdirectory, we return the subdirectory.
        # f.e. /var/lib/localstack/lib/amazon-mq/5.16.5/apache-activemq-5.16.5/
        if installed_dir and subdir:
            return os.path.join(installed_dir, subdir)

        return installed_dir

    def _get_archive_subdir(self) -> str | None:
        """
        :return: name of the subdirectory contained in the archive or none if the package content is at the root level
                of the archive
        """
        return None

    def get_executable_path(self) -> str | None:
        subdir = self._get_archive_subdir()
        if subdir is None:
            return super().get_executable_path()
        else:
            install_dir = self.get_installed_dir()
            if install_dir:
                install_dir = install_dir[: -len(subdir)]
                return self._get_install_marker_path(install_dir)

    def _install(self, target: InstallTarget) -> None:
        target_directory = self._get_install_dir(target)
        mkdir(target_directory)
        download_url = self._get_download_url()
        archive_name = os.path.basename(download_url)
        archive_path = os.path.join(config.dirs.tmp, archive_name)
        download_and_extract(
            download_url,
            retries=3,
            tmp_archive=archive_path,
            target_dir=target_directory,
        )
        rm_rf(archive_path)
        if self.extract_single_directory:
            dir_contents = os.listdir(target_directory)
            if len(dir_contents) != 1:
                return
            target_subdir = os.path.join(target_directory, dir_contents[0])
            if not os.path.isdir(target_subdir):
                return
            os.rename(target_subdir, f"{target_directory}.backup")
            rm_rf(target_directory)
            os.rename(f"{target_directory}.backup", target_directory)


class PermissionDownloadInstaller(DownloadInstaller, ABC):
    def _install(self, target: InstallTarget) -> None:
        super()._install(target)
        chmod_r(self.get_executable_path(), 0o777)


class GitHubReleaseInstaller(PermissionDownloadInstaller):
    """
    Installer which downloads an asset from a GitHub project's tag.
    """

    def __init__(self, name: str, tag: str, github_slug: str):
        super().__init__(name, tag)
        self.github_tag_url = (
            f"https://api.github.com/repos/{github_slug}/releases/tags/{self.version}"
        )

    @lru_cache()
    def _get_download_url(self) -> str:
        asset_name = self._get_github_asset_name()
        # try to use a token when calling the GH API for increased API rate limits
        headers = None
        gh_token = os.environ.get("GITHUB_API_TOKEN")
        if gh_token:
            headers = {"authorization": f"Bearer {gh_token}"}
        response = requests.get(self.github_tag_url, headers=headers)
        if not response.ok:
            raise PackageException(
                f"Could not get list of releases from {self.github_tag_url}: {response.text}"
            )
        github_release = response.json()
        download_url = None
        for asset in github_release.get("assets", []):
            # find the correct binary in the release
            if asset["name"] == asset_name:
                download_url = asset["browser_download_url"]
                break
        if download_url is None:
            raise PackageException(
                f"Could not find required binary {asset_name} in release {self.github_tag_url}"
            )
        return download_url

    def _get_install_marker_path(self, install_dir: str) -> str:
        # Use the GitHub asset name instead of the download URL (since the download URL needs to be fetched online).
        return os.path.join(install_dir, self._get_github_asset_name())

    def _get_github_asset_name(self) -> str:
        """
        Determines the name of the asset to download.
        The asset name must be determinable without having any online data (because it is used in offline scenarios to
        determine if the package is already installed).

        :return: name of the asset to download from the GitHub project's tag / version
        """
        raise NotImplementedError()


class NodePackageInstaller(ExecutableInstaller):
    """Package installer for Node / NPM packages."""

    def __init__(
        self,
        package_name: str,
        version: str,
        package_spec: Optional[str] = None,
        main_module: str = "main.js",
    ):
        """
        Initializes the Node / NPM package installer.
        :param package_name: npm package name
        :param version: version of the package which should be installed
        :param package_spec: optional package spec for the installation.
                If not set, the package name and version will be used for the installation.
        :param main_module: main module file of the package
        """
        super().__init__(package_name, version)
        self.package_name = package_name
        # If the package spec is not explicitly set (f.e. to a repo), we build it and pin the version
        self.package_spec = package_spec or f"{self.package_name}@{version}"
        self.main_module = main_module

    def _get_install_marker_path(self, install_dir: str) -> str:
        return os.path.join(install_dir, "node_modules", self.package_name, self.main_module)

    def _install(self, target: InstallTarget) -> None:
        target_dir = self._get_install_dir(target)

        run(
            [
                "npm",
                "install",
                "--prefix",
                target_dir,
                self.package_spec,
            ]
        )
        # npm 9+ does _not_ set the ownership of files anymore if run as root
        # - https://github.blog/changelog/2022-10-24-npm-v9-0-0-released/
        # - https://github.com/npm/cli/pull/5704
        # - https://github.com/localstack/localstack/issues/7620
        if is_root():
            # if the package was installed as root, set the ownership manually
            LOG.debug("Setting ownership root:root on %s", target_dir)
            chown_r(target_dir, "root")


LOCALSTACK_VENV = VirtualEnvironment(LOCALSTACK_VENV_FOLDER)


class PythonPackageInstaller(PackageInstaller):
    """
    Package installer which allows the runtime-installation of additional python packages used by certain services.
    f.e. vosk as offline speech recognition toolkit (which is ~7MB in size compressed and ~26MB uncompressed).
    """

    normalized_name: str
    """Normalized package name according to PEP440."""

    def __init__(self, name: str, version: str, *args, **kwargs):
        super().__init__(name, version, *args, **kwargs)
        self.normalized_name = self._normalize_package_name(name)

    def _normalize_package_name(self, name: str):
        """
        Normalized the Python package name according to PEP440.
        https://packaging.python.org/en/latest/specifications/name-normalization/#name-normalization
        """
        return re.sub(r"[-_.]+", "-", name).lower()

    def _get_install_dir(self, target: InstallTarget) -> str:
        # all python installers share a venv
        return os.path.join(target.value, "python-packages")

    def _get_install_marker_path(self, install_dir: str) -> str:
        python_subdir = f"python{version_info[0]}.{version_info[1]}"
        dist_info_dir = f"{self.normalized_name}-{self.version}.dist-info"
        # the METADATA file is mandatory, use it as install marker
        return os.path.join(
            install_dir, "lib", python_subdir, "site-packages", dist_info_dir, "METADATA"
        )

    def _get_venv(self, target: InstallTarget) -> VirtualEnvironment:
        venv_dir = self._get_install_dir(target)
        return VirtualEnvironment(venv_dir)

    def _prepare_installation(self, target: InstallTarget) -> None:
        # make sure the venv is properly set up before installing the package
        venv = self._get_venv(target)
        if not venv.exists:
            LOG.info("creating virtual environment at %s", venv.venv_dir)
            venv.create()
            LOG.info("adding localstack venv path %s", venv.venv_dir)
            venv.add_pth("localstack-venv", LOCALSTACK_VENV)
        LOG.debug("injecting venv into path %s", venv.venv_dir)
        venv.inject_to_sys_path()

    def _install(self, target: InstallTarget) -> None:
        venv = self._get_venv(target)
        python_bin = os.path.join(venv.venv_dir, "bin/python")

        # run pip via the python binary of the venv
        run([python_bin, "-m", "pip", "install", f"{self.name}=={self.version}"], print_error=False)

    def _setup_existing_installation(self, target: InstallTarget) -> None:
        """If the venv is already present, it just needs to be initialized once."""
        self._prepare_installation(target)


class MavenDownloadInstaller(DownloadInstaller):
    """The packageURL is easy copy/pastable from the Maven central repository and the first package URL
    defines the package name and version.
    Example package_url: pkg:maven/software.amazon.event.ruler/event-ruler@1.7.3
    => name: event-ruler
    => version: 1.7.3
    """

    # Example: software.amazon.event.ruler
    group_id: str
    # Example: event-ruler
    artifact_id: str

    # Custom installation directory
    install_dir_suffix: str | None

    def __init__(self, package_url: str, install_dir_suffix: str | None = None):
        self.group_id, self.artifact_id, version = parse_maven_package_url(package_url)
        super().__init__(self.artifact_id, version)
        self.install_dir_suffix = install_dir_suffix

    def _get_download_url(self) -> str:
        group_id_path = self.group_id.replace(".", "/")
        return f"{MAVEN_REPO_URL}/{group_id_path}/{self.artifact_id}/{self.version}/{self.artifact_id}-{self.version}.jar"

    def _get_install_dir(self, target: InstallTarget) -> str:
        """Allow to overwrite the default installation directory.
        This enables downloading transitive dependencies into the same directory.
        """
        if self.install_dir_suffix:
            return os.path.join(target.value, self.install_dir_suffix)
        else:
            return super()._get_install_dir(target)


class MavenPackageInstaller(MavenDownloadInstaller):
    """Package installer for downloading Maven JARs, including optional dependencies.
    The first Maven package is used as main LPM package and other dependencies are installed additionally.
    Follows the Maven naming conventions: https://maven.apache.org/guides/mini/guide-naming-conventions.html
    """

    # Installers for Maven dependencies
    dependencies: list[MavenDownloadInstaller]

    def __init__(self, *package_urls: str):
        super().__init__(package_urls[0])
        self.dependencies = []

        # Create installers for dependencies
        for package_url in package_urls[1:]:
            install_dir_suffix = os.path.join(self.name, self.version)
            self.dependencies.append(MavenDownloadInstaller(package_url, install_dir_suffix))

    def _install(self, target: InstallTarget) -> None:
        # Install all dependencies first
        for dependency in self.dependencies:
            dependency._install(target)
        # Install the main Maven package once all dependencies are installed.
        # This main package indicates whether all dependencies are installed.
        super()._install(target)


def parse_maven_package_url(package_url: str) -> Tuple[str, str, str]:
    """Example: parse_maven_package_url("pkg:maven/software.amazon.event.ruler/event-ruler@1.7.3")
    -> software.amazon.event.ruler, event-ruler, 1.7.3
    """
    parts = package_url.split("/")
    group_id = parts[1]
    sub_parts = parts[2].split("@")
    artifact_id = sub_parts[0]
    version = sub_parts[1]
    return group_id, artifact_id, version

1	import logging	1✔
2	import os	1✔
3	import re	1✔
4	from abc import ABC	1✔
5	from functools import lru_cache	1✔
6	from sys import version_info	1✔
7	from typing import Optional, Tuple	1✔
8
9	import requests	1✔
10
11	from localstack import config	1✔
12
13	from ..constants import LOCALSTACK_VENV_FOLDER, MAVEN_REPO_URL	1✔
14	from ..utils.archives import download_and_extract	1✔
15	from ..utils.files import chmod_r, chown_r, mkdir, rm_rf	1✔
16	from ..utils.http import download	1✔
17	from ..utils.run import is_root, run	1✔
18	from ..utils.venv import VirtualEnvironment	1✔
19	from .api import InstallTarget, PackageException, PackageInstaller	1✔
20
21	LOG = logging.getLogger(__name__)	1✔
22
23
24	class SystemNotSupportedException(PackageException):	1✔
25	"""Exception indicating that the current system is not allowed."""
26
27	pass	1✔
28
29
30	class ExecutableInstaller(PackageInstaller, ABC):	1✔
31	"""
32	This installer simply adds a clean interface for accessing a downloaded executable directly
33	"""
34
35	def get_executable_path(self) -> str \| None:	1✔
36	"""
37	:return: the path to the downloaded binary or None if it's not yet downloaded / installed.
38	"""
39	install_dir = self.get_installed_dir()	1✔
40	if install_dir:	1✔
41	return self._get_install_marker_path(install_dir)	1✔
42
43
44	class DownloadInstaller(ExecutableInstaller):	1✔
45	def __init__(self, name: str, version: str):	1✔
46	super().__init__(name, version)	1✔
47
48	def _get_download_url(self) -> str:	1✔
49	raise NotImplementedError()
50
51	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
52	url = self._get_download_url()	1✔
53	binary_name = os.path.basename(url)	1✔
54	return os.path.join(install_dir, binary_name)	1✔
55
56	def _install(self, target: InstallTarget) -> None:	1✔
57	target_directory = self._get_install_dir(target)	1✔
58	mkdir(target_directory)	1✔
59	download_url = self._get_download_url()	1✔
60	target_path = self._get_install_marker_path(target_directory)	1✔
61	download(download_url, target_path)	1✔
62
63
64	class ArchiveDownloadAndExtractInstaller(ExecutableInstaller):	1✔
65	def __init__(self, name: str, version: str, extract_single_directory: bool = False):	1✔
66	"""
67	:param name: technical package name, f.e. "opensearch"
68	:param version: version of the package to install
69	:param extract_single_directory: whether to extract files from single root folder in the archive
70	"""
71	super().__init__(name, version)	1✔
72	self.extract_single_directory = extract_single_directory	1✔
73
74	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
75	raise NotImplementedError()
76
77	def _get_download_url(self) -> str:	1✔
78	raise NotImplementedError()
79
80	def get_installed_dir(self) -> str \| None:	1✔
81	installed_dir = super().get_installed_dir()	1✔
82	subdir = self._get_archive_subdir()	1✔
83
84	# If the specific installer defines a subdirectory, we return the subdirectory.
85	# f.e. /var/lib/localstack/lib/amazon-mq/5.16.5/apache-activemq-5.16.5/
86	if installed_dir and subdir:	1✔
87	return os.path.join(installed_dir, subdir)	1✔
88
89	return installed_dir	1✔
90
91	def _get_archive_subdir(self) -> str \| None:	1✔
92	"""
93	:return: name of the subdirectory contained in the archive or none if the package content is at the root level
94	of the archive
95	"""
96	return None	1✔
97
98	def get_executable_path(self) -> str \| None:	1✔
99	subdir = self._get_archive_subdir()	1✔
100	if subdir is None:	1✔
101	return super().get_executable_path()	1✔
102	else:
103	install_dir = self.get_installed_dir()	×
104	if install_dir:	×
105	install_dir = install_dir[: -len(subdir)]	×
106	return self._get_install_marker_path(install_dir)	×
107
108	def _install(self, target: InstallTarget) -> None:	1✔
109	target_directory = self._get_install_dir(target)	1✔
110	mkdir(target_directory)	1✔
111	download_url = self._get_download_url()	1✔
112	archive_name = os.path.basename(download_url)	1✔
113	archive_path = os.path.join(config.dirs.tmp, archive_name)	1✔
114	download_and_extract(	1✔
115	download_url,
116	retries=3,
117	tmp_archive=archive_path,
118	target_dir=target_directory,
119	)
120	rm_rf(archive_path)	1✔
121	if self.extract_single_directory:	1✔
122	dir_contents = os.listdir(target_directory)	1✔
123	if len(dir_contents) != 1:	1✔
124	return	×
125	target_subdir = os.path.join(target_directory, dir_contents[0])	1✔
126	if not os.path.isdir(target_subdir):	1✔
127	return	×
128	os.rename(target_subdir, f"{target_directory}.backup")	1✔
129	rm_rf(target_directory)	1✔
130	os.rename(f"{target_directory}.backup", target_directory)	1✔
131
132
133	class PermissionDownloadInstaller(DownloadInstaller, ABC):	1✔
134	def _install(self, target: InstallTarget) -> None:	1✔
135	super()._install(target)	1✔
136	chmod_r(self.get_executable_path(), 0o777)	×
137
138
139	class GitHubReleaseInstaller(PermissionDownloadInstaller):	1✔
140	"""
141	Installer which downloads an asset from a GitHub project's tag.
142	"""
143
144	def __init__(self, name: str, tag: str, github_slug: str):	1✔
145	super().__init__(name, tag)	1✔
146	self.github_tag_url = (	1✔
147	f"https://api.github.com/repos/{github_slug}/releases/tags/{self.version}"
148	)
149
150	@lru_cache()	1✔
151	def _get_download_url(self) -> str:	1✔
152	asset_name = self._get_github_asset_name()	1✔
153	# try to use a token when calling the GH API for increased API rate limits
154	headers = None	1✔
155	gh_token = os.environ.get("GITHUB_API_TOKEN")	1✔
156	if gh_token:	1✔
157	headers = {"authorization": f"Bearer {gh_token}"}	1✔
158	response = requests.get(self.github_tag_url, headers=headers)	1✔
159	if not response.ok:	1✔
160	raise PackageException(	1✔
161	f"Could not get list of releases from {self.github_tag_url}: {response.text}"
162	)
163	github_release = response.json()	×
164	download_url = None	×
165	for asset in github_release.get("assets", []):	×
166	# find the correct binary in the release
167	if asset["name"] == asset_name:	×
168	download_url = asset["browser_download_url"]	×
169	break	×
170	if download_url is None:	×
171	raise PackageException(	×
172	f"Could not find required binary {asset_name} in release {self.github_tag_url}"
173	)
174	return download_url	×
175
176	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
177	# Use the GitHub asset name instead of the download URL (since the download URL needs to be fetched online).
178	return os.path.join(install_dir, self._get_github_asset_name())	1✔
179
180	def _get_github_asset_name(self) -> str:	1✔
181	"""
182	Determines the name of the asset to download.
183	The asset name must be determinable without having any online data (because it is used in offline scenarios to
184	determine if the package is already installed).
185
186	:return: name of the asset to download from the GitHub project's tag / version
187	"""
188	raise NotImplementedError()
189
190
191	class NodePackageInstaller(ExecutableInstaller):	1✔
192	"""Package installer for Node / NPM packages."""
193
194	def __init__(	1✔
195	self,
196	package_name: str,
197	version: str,
198	package_spec: Optional[str] = None,
199	main_module: str = "main.js",
200	):
201	"""
202	Initializes the Node / NPM package installer.
203	:param package_name: npm package name
204	:param version: version of the package which should be installed
205	:param package_spec: optional package spec for the installation.
206	If not set, the package name and version will be used for the installation.
207	:param main_module: main module file of the package
208	"""
209	super().__init__(package_name, version)	1✔
210	self.package_name = package_name	1✔
211	# If the package spec is not explicitly set (f.e. to a repo), we build it and pin the version
212	self.package_spec = package_spec or f"{self.package_name}@{version}"	1✔
213	self.main_module = main_module	1✔
214
215	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
216	return os.path.join(install_dir, "node_modules", self.package_name, self.main_module)	1✔
217
218	def _install(self, target: InstallTarget) -> None:	1✔
219	target_dir = self._get_install_dir(target)	1✔
220
221	run(	1✔
222	[
223	"npm",
224	"install",
225	"--prefix",
226	target_dir,
227	self.package_spec,
228	]
229	)
230	# npm 9+ does _not_ set the ownership of files anymore if run as root
231	# - https://github.blog/changelog/2022-10-24-npm-v9-0-0-released/
232	# - https://github.com/npm/cli/pull/5704
233	# - https://github.com/localstack/localstack/issues/7620
234	if is_root():	1✔
235	# if the package was installed as root, set the ownership manually
236	LOG.debug("Setting ownership root:root on %s", target_dir)	1✔
237	chown_r(target_dir, "root")	1✔
238
239
240	LOCALSTACK_VENV = VirtualEnvironment(LOCALSTACK_VENV_FOLDER)	1✔
241
242
243	class PythonPackageInstaller(PackageInstaller):	1✔
244	"""
245	Package installer which allows the runtime-installation of additional python packages used by certain services.
246	f.e. vosk as offline speech recognition toolkit (which is ~7MB in size compressed and ~26MB uncompressed).
247	"""
248
249	normalized_name: str	1✔
250	"""Normalized package name according to PEP440."""	1✔
251
252	def __init__(self, name: str, version: str, args, *kwargs):	1✔
253	super().__init__(name, version, args, *kwargs)	1✔
254	self.normalized_name = self._normalize_package_name(name)	1✔
255
256	def _normalize_package_name(self, name: str):	1✔
257	"""
258	Normalized the Python package name according to PEP440.
259	https://packaging.python.org/en/latest/specifications/name-normalization/#name-normalization
260	"""
261	return re.sub(r"[-_.]+", "-", name).lower()	1✔
262
263	def _get_install_dir(self, target: InstallTarget) -> str:	1✔
264	# all python installers share a venv
265	return os.path.join(target.value, "python-packages")	1✔
266
267	def _get_install_marker_path(self, install_dir: str) -> str:	1✔
268	python_subdir = f"python{version_info[0]}.{version_info[1]}"	1✔
269	dist_info_dir = f"{self.normalized_name}-{self.version}.dist-info"	1✔
270	# the METADATA file is mandatory, use it as install marker
271	return os.path.join(	1✔
272	install_dir, "lib", python_subdir, "site-packages", dist_info_dir, "METADATA"
273	)
274
275	def _get_venv(self, target: InstallTarget) -> VirtualEnvironment:	1✔
276	venv_dir = self._get_install_dir(target)	1✔
277	return VirtualEnvironment(venv_dir)	1✔
278
279	def _prepare_installation(self, target: InstallTarget) -> None:	1✔
280	# make sure the venv is properly set up before installing the package
281	venv = self._get_venv(target)	1✔
282	if not venv.exists:	1✔
283	LOG.info("creating virtual environment at %s", venv.venv_dir)	1✔
284	venv.create()	1✔
285	LOG.info("adding localstack venv path %s", venv.venv_dir)	1✔
286	venv.add_pth("localstack-venv", LOCALSTACK_VENV)	1✔
287	LOG.debug("injecting venv into path %s", venv.venv_dir)	1✔
288	venv.inject_to_sys_path()	1✔
289
290	def _install(self, target: InstallTarget) -> None:	1✔
291	venv = self._get_venv(target)	1✔
292	python_bin = os.path.join(venv.venv_dir, "bin/python")	1✔
293
294	# run pip via the python binary of the venv
295	run([python_bin, "-m", "pip", "install", f"{self.name}=={self.version}"], print_error=False)	1✔
296
297	def _setup_existing_installation(self, target: InstallTarget) -> None:	1✔
298	"""If the venv is already present, it just needs to be initialized once."""
299	self._prepare_installation(target)	1✔
300
301
302	class MavenDownloadInstaller(DownloadInstaller):	1✔
303	"""The packageURL is easy copy/pastable from the Maven central repository and the first package URL
304	defines the package name and version.
305	Example package_url: pkg:maven/software.amazon.event.ruler/event-ruler@1.7.3
306	=> name: event-ruler
307	=> version: 1.7.3
308	"""
309
310	# Example: software.amazon.event.ruler
311	group_id: str	1✔
312	# Example: event-ruler
313	artifact_id: str	1✔
314
315	# Custom installation directory
316	install_dir_suffix: str \| None	1✔
317
318	def __init__(self, package_url: str, install_dir_suffix: str \| None = None):	1✔
319	self.group_id, self.artifact_id, version = parse_maven_package_url(package_url)	1✔
320	super().__init__(self.artifact_id, version)	1✔
321	self.install_dir_suffix = install_dir_suffix	1✔
322
323	def _get_download_url(self) -> str:	1✔
324	group_id_path = self.group_id.replace(".", "/")	1✔
325	return f"{MAVEN_REPO_URL}/{group_id_path}/{self.artifact_id}/{self.version}/{self.artifact_id}-{self.version}.jar"	1✔
326
327	def _get_install_dir(self, target: InstallTarget) -> str:	1✔
328	"""Allow to overwrite the default installation directory.
329	This enables downloading transitive dependencies into the same directory.
330	"""
331	if self.install_dir_suffix:	1✔
UNCOV 332	return os.path.join(target.value, self.install_dir_suffix)	×
333	else:
334	return super()._get_install_dir(target)	1✔
335
336
337	class MavenPackageInstaller(MavenDownloadInstaller):	1✔
338	"""Package installer for downloading Maven JARs, including optional dependencies.
339	The first Maven package is used as main LPM package and other dependencies are installed additionally.
340	Follows the Maven naming conventions: https://maven.apache.org/guides/mini/guide-naming-conventions.html
341	"""
342
343	# Installers for Maven dependencies
344	dependencies: list[MavenDownloadInstaller]	1✔
345
346	def __init__(self, *package_urls: str):	1✔
347	super().__init__(package_urls[0])	1✔
348	self.dependencies = []	1✔
349
350	# Create installers for dependencies
351	for package_url in package_urls[1:]:	1✔
352	install_dir_suffix = os.path.join(self.name, self.version)	1✔
353	self.dependencies.append(MavenDownloadInstaller(package_url, install_dir_suffix))	1✔
354
355	def _install(self, target: InstallTarget) -> None:	1✔
356	# Install all dependencies first
UNCOV 357	for dependency in self.dependencies:	×
UNCOV 358	dependency._install(target)	×
359	# Install the main Maven package once all dependencies are installed.
360	# This main package indicates whether all dependencies are installed.
UNCOV 361	super()._install(target)	×
362
363
364	def parse_maven_package_url(package_url: str) -> Tuple[str, str, str]:	1✔
365	"""Example: parse_maven_package_url("pkg:maven/software.amazon.event.ruler/event-ruler@1.7.3")
366	-> software.amazon.event.ruler, event-ruler, 1.7.3
367	"""
368	parts = package_url.split("/")	1✔
369	group_id = parts[1]	1✔
370	sub_parts = parts[2].split("@")	1✔
371	artifact_id = sub_parts[0]	1✔
372	version = sub_parts[1]	1✔
373	return group_id, artifact_id, version	1✔

localstack / localstack / 7ae8c08b-27e7-4df6-bfb9-4892ae974ff7

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