20129546909

Committed 11 Dec 2025 09:09AM UTC coverage: 86.867% (+0.001%) from 86.866%

Build # 20129546909

Build Type

push

github

Committed by

web-flow

Commit Message

Skip failing DNS unit test (#13501)

Run Details

69915 of 80485 relevant lines covered (86.87%)

0.87 hits per line

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

89.5

/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

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, get_proxies
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_checksum_url(self) -> str | None:
        """
        Checksum URL for the archive. This is used to verify the integrity of the downloaded archive.
        This method can be implemented by subclasses to provide the correct URL for the checksum file.
        If not implemented, checksum verification will be skipped.

        :return: URL to the checksum file for the archive, or None if not available.
        """
        return None

    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 _handle_single_directory_extraction(self, target_directory: str) -> None:
        """
        Handle extraction of archives that contain a single root directory.
        Moves the contents up one level if extract_single_directory is True.

        :param target_directory: The target extraction directory
        :return: None
        """
        if not self.extract_single_directory:
            return

        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)

    def _download_archive(
        self,
        target: InstallTarget,
        download_url: str,
    ) -> None:
        target_directory = self._get_install_dir(target)
        mkdir(target_directory)
        download_url = download_url or self._get_download_url()
        archive_name = os.path.basename(download_url)
        archive_path = os.path.join(config.dirs.tmp, archive_name)

        # Get checksum info if available
        checksum_url = self._get_checksum_url()

        try:
            download_and_extract(
                download_url,
                retries=3,
                tmp_archive=archive_path,
                target_dir=target_directory,
                checksum_url=checksum_url,
            )
            self._handle_single_directory_extraction(target_directory)
        finally:
            rm_rf(archive_path)

    def _install(self, target: InstallTarget) -> None:
        self._download_archive(target, self._get_download_url())


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

localstack / localstack / 20129546909

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