6eb2aab0-9781-40f0-8cc1-9e7c437858cb

Committed 22 Apr 2025 04:28PM UTC coverage: 86.272% (+0.007%) from 86.265%

Build # 6eb2aab0-9781-40f0-8cc1-9e7c437858cb

Build Type

push

circleci

Committed by

web-flow

Commit Message

APIGW: validate REST API custom id tag (#12539)

Coverage Stats

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

82 existing lines in 10 files now uncovered.

63897 of 74065 relevant lines covered (86.27%)

0.86 hits per line

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

88.48

/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 Any, 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)
        return None


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)
        return None

    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)  # type: ignore[arg-type]


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: Any, **kwargs: Any):
        super().__init__(name, version, *args, **kwargs)
        self.normalized_name = self._normalize_package_name(name)

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

localstack / localstack / 6eb2aab0-9781-40f0-8cc1-9e7c437858cb

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