|
1 | 1 | import abc |
| 2 | +import enum |
2 | 3 | import re |
3 | 4 | import shlex |
4 | | -from collections.abc import Iterator |
| 5 | +from collections.abc import Iterable, Iterator |
5 | 6 | from typing import TYPE_CHECKING, Any, Callable, ClassVar, Generic, Optional, TypeVar, Union |
6 | 7 |
|
7 | 8 | import tmt.log |
|
10 | 11 | from tmt.container import container, simple_field |
11 | 12 | from tmt.utils import Command, CommandOutput, GeneralError, Path, PrepareError, ShellScript |
12 | 13 |
|
| 14 | + |
| 15 | +class SpecialPackageOrigin(str, enum.Enum): |
| 16 | + """ |
| 17 | + Sentinel values used in place of an actual repository name to convey |
| 18 | + special package states returned by :py:meth:`PackageManager.get_package_origin`. |
| 19 | + """ |
| 20 | + |
| 21 | + #: Package is not installed on the guest. |
| 22 | + NOT_INSTALLED = '<not-installed>' |
| 23 | + #: Package is installed but its source repository cannot be determined |
| 24 | + #: (e.g. pre-installed in a container image). |
| 25 | + UNKNOWN = '<unknown>' |
| 26 | + |
| 27 | + |
13 | 28 | if TYPE_CHECKING: |
14 | 29 | from tmt._compat.typing import TypeAlias |
15 | 30 |
|
|
21 | 36 |
|
22 | 37 | #: A type of package manager names. |
23 | 38 | GuestPackageManager: TypeAlias = str |
| 39 | + |
| 40 | + #: A package origin: either an actual repository name or a :class:`SpecialPackageOrigin`. |
| 41 | + PackageOrigin: TypeAlias = Union[str, SpecialPackageOrigin] |
24 | 42 | else: |
25 | 43 | Repository: Any = None # type: ignore[assignment] |
26 | 44 |
|
@@ -222,6 +240,27 @@ def list_packages(self, repository: "Repository") -> ShellScript: |
222 | 240 | """ |
223 | 241 | raise NotImplementedError |
224 | 242 |
|
| 243 | + def get_package_origin(self, packages: Iterable[str]) -> ShellScript: |
| 244 | + """ |
| 245 | + List source repositories for each installed package. |
| 246 | +
|
| 247 | + The script must emit one line per package in the format:: |
| 248 | +
|
| 249 | + <name> <origin> |
| 250 | +
|
| 251 | + Empty lines are allowed and will be ignored by the caller. If |
| 252 | + the origin field is omitted the package is treated as having an |
| 253 | + unknown source repository (equivalent to |
| 254 | + :py:attr:`SpecialPackageOrigin.UNKNOWN`). Packages whose name |
| 255 | + does not appear in the output at all are treated as not installed |
| 256 | + (equivalent to :py:attr:`SpecialPackageOrigin.NOT_INSTALLED`). |
| 257 | +
|
| 258 | + :param packages: Package names to query. |
| 259 | + :returns: A shell script to list source repositories for the given packages. |
| 260 | + :raises NotImplementedError: If the package manager does not support this query. |
| 261 | + """ |
| 262 | + raise NotImplementedError |
| 263 | + |
225 | 264 | def create_repository(self, directory: Path) -> ShellScript: |
226 | 265 | """ |
227 | 266 | Create repository metadata for package files in the given directory. |
@@ -338,6 +377,32 @@ def list_packages(self, repository: "Repository") -> list[str]: |
338 | 377 |
|
339 | 378 | return stdout.strip().splitlines() |
340 | 379 |
|
| 380 | + def get_package_origin(self, packages: Iterable[str]) -> 'dict[str, PackageOrigin]': |
| 381 | + """ |
| 382 | + Get the repository each package was installed from. |
| 383 | +
|
| 384 | + :param packages: Package names to query. |
| 385 | + :returns: A mapping of package names to source repository names. |
| 386 | + Packages not installed are mapped to |
| 387 | + :py:attr:`SpecialPackageOrigin.NOT_INSTALLED`. Packages whose |
| 388 | + source repository is unknown are mapped to |
| 389 | + :py:attr:`SpecialPackageOrigin.UNKNOWN`. |
| 390 | + """ |
| 391 | + result: dict[str, PackageOrigin] = dict.fromkeys( |
| 392 | + packages, SpecialPackageOrigin.NOT_INSTALLED |
| 393 | + ) |
| 394 | + script = self.engine.get_package_origin(result.keys()) |
| 395 | + output = self.guest.execute(script) |
| 396 | + for line in (output.stdout or '').strip().splitlines(): |
| 397 | + # Empty lines are allowed by the engine contract. |
| 398 | + if not line.strip(): |
| 399 | + continue |
| 400 | + parts = line.split(maxsplit=1) |
| 401 | + package = parts[0] |
| 402 | + # Omitted origin field → unknown source repository. |
| 403 | + result[package] = parts[1] if len(parts) == 2 else SpecialPackageOrigin.UNKNOWN |
| 404 | + return result |
| 405 | + |
341 | 406 | def create_repository(self, directory: Path) -> CommandOutput: |
342 | 407 | """ |
343 | 408 | Wrapper of :py:meth:`PackageManagerEngine.create_repository`. |
|
0 commit comments