List packages from repository providers in artifacts.yaml

tests/unit/artifact/conftest.py

@@ -6,6 +6,7 @@
 from tmt.steps.prepare.artifact import get_artifact_provider
 from tmt.steps.prepare.artifact.providers import koji as koji_module
 from tmt.steps.prepare.artifact.providers.brew import BrewArtifactProvider
+from tmt.steps.prepare.artifact.providers.copr_build import CoprBuildArtifactProvider
 from tmt.steps.prepare.artifact.providers.koji import KojiArtifactProvider
 
 
@@ -50,6 +51,13 @@ def mock_initialize_session(self, api_url=None, top_url=None):
         yield mock_koji
 
 
+@pytest.fixture
+def mock_copr():
+    mock_session = MagicMock()
+    with patch.object(CoprBuildArtifactProvider, '_initialize_session', return_value=mock_session):
+        yield mock_session
+
+
 @pytest.fixture
 def artifact_provider(root_logger, _load_plugins):
     def get_provider(provider_id: str, repository_priority: int = 50):

tests/unit/artifact/test_copr_repository.py

@@ -1,4 +1,4 @@
-from unittest.mock import MagicMock
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -28,13 +28,51 @@ def test_invalid_repository_id(raw_id, error, artifact_provider):
         artifact_provider(raw_id)
 
 
-def test_fetch_contents_enables_repository(mock_package_manager, artifact_provider, tmppath):
+def test_fetch_contents_enables_repository(
+    mock_copr, mock_package_manager, artifact_provider, tmppath
+):
+    mock_repo = MagicMock()
     mock_guest = MagicMock()
     mock_guest.package_manager = mock_package_manager
 
     provider = artifact_provider("copr.repository:@teemtee/stable")
 
-    result = provider.fetch_contents(mock_guest, tmppath / "artifacts")
+    with patch(
+        'tmt.steps.prepare.artifact.providers.copr_repository.Repository.from_file_path',
+        return_value=mock_repo,
+    ) as mock_from_file:
+        result = provider.fetch_contents(mock_guest, tmppath / "artifacts")
 
     mock_package_manager.enable_copr.assert_called_once_with('@teemtee/stable')
+    mock_guest.pull.assert_called_once()
+    mock_from_file.assert_called_once()
     assert result == []
+    assert provider.repository is mock_repo
+
+
+def test_enumerate_artifacts(mock_copr, mock_package_manager, artifact_provider, tmppath):
+    mock_repo = MagicMock()
+    mock_repo.repo_ids = ['group_teemtee-stable-fedora-42-x86_64']
+    mock_guest = MagicMock()
+    mock_guest.package_manager = mock_package_manager
+
+    mock_package_manager.list_packages.return_value = [
+        'tmt-1.69.0-1.fc42.noarch',
+        'tmt-all-0:1.69.0-1.fc42.noarch',
+    ]
+
+    provider = artifact_provider("copr.repository:@teemtee/stable")
+
+    with patch(
+        'tmt.steps.prepare.artifact.providers.copr_repository.Repository.from_file_path',
+        return_value=mock_repo,
+    ):
+        provider.fetch_contents(mock_guest, tmppath / "artifacts")
+
+    provider.enumerate_artifacts(mock_guest)
+
+    assert len(provider.artifacts) == 2
+    assert provider.artifacts[0].version.name == 'tmt'
+    assert provider.artifacts[0].version.version == '1.69.0'
+    assert provider.artifacts[1].version.name == 'tmt-all'
+    assert provider.artifacts[1].version.epoch == 0

tmt/steps/prepare/artifact/__init__.py

@@ -265,6 +265,10 @@ def go(
             guest.package_manager.install_repository(repo)
             logger.debug(f"Installed repository '{repo.name}'.")
 
+        # Enumerate artifacts from installed repositories.
+        for provider in providers:
+            provider.enumerate_artifacts(guest)
+
         # Persist artifact metadata to YAML
         self._save_artifacts_metadata(providers)
 

tmt/steps/prepare/artifact/providers/__init__.py

@@ -5,7 +5,6 @@
 import configparser
 from abc import ABC, abstractmethod
 from collections.abc import Iterator, Sequence
-from functools import cached_property
 from re import Pattern
 from shlex import quote
 from typing import Any, Optional
@@ -86,6 +85,32 @@ def from_rpm_meta(cls, rpm_meta: dict[str, Any]) -> Self:
             epoch=rpm_meta.get("epoch", 0),
         )
 
+    @classmethod
+    def from_nevra(
+        cls, nevra: str
+    ) -> Self:  # TODO: move this to `tmt.package_managers.PackageManager.list_packages`
+        """
+        Version constructed from a NEVRA string as returned by ``dnf repoquery``.
+
+        Example usage:
+
+        .. code-block:: python
+
+            version_info = RpmVersion.from_nevra("curl-0:8.11.1-7.fc42.x86_64")
+            version_info = RpmVersion.from_nevra("curl-8.11.1-7.fc42.x86_64")
+        """
+        nvr_epoch, sep, arch = nevra.rpartition('.')
+        if not sep:
+            raise ValueError(f"Cannot parse arch from NEVRA '{nevra}'.")
+        parts = nvr_epoch.rsplit('-', 2)
+        if len(parts) != 3:
+            raise ValueError(f"Cannot parse NVR from NEVRA '{nevra}'.")
+        name, ev, release = parts
+        epoch_str, sep, version = ev.partition(':')
+        epoch = int(epoch_str) if sep else 0
+        version = version if sep else epoch_str
+        return cls(name=name, version=version, release=release, arch=arch, epoch=epoch)
+
     @classmethod
     def from_filename(cls, filename: str) -> Self:
         """
@@ -177,6 +202,7 @@ def __init__(self, raw_id: str, repository_priority: int, logger: tmt.log.Logger
         self.sanitized_id = tmt.utils.sanitize_name(raw_id, allow_slash=False)
 
         self.id = self._extract_provider_id(raw_id)
+        self._artifacts: list[ArtifactInfo] = []
 
     @classmethod
     @abstractmethod
@@ -191,16 +217,12 @@ def _extract_provider_id(cls, raw_id: str) -> ArtifactProviderId:
 
         raise NotImplementedError
 
-    @cached_property
+    @property
     @abstractmethod
     def artifacts(self) -> Sequence[ArtifactInfo]:
         """
         Collect all artifacts available from this provider.
 
-        The method is left for derived classes to implement with respect
-        to the actual artifact provider they implement. The list of
-        artifacts will be cached, and is treated as read-only.
-
         :returns: a list of provided artifacts.
         """
 
@@ -300,6 +322,39 @@ def get_repositories(self) -> list['Repository']:
         """
         return []
 
+    def enumerate_artifacts(
+        self, guest: Guest
+    ) -> None:  # TODO: refactor this once the NEVRA parsing is centralized.
+        """
+        Enumerate artifacts available from this provider after its repositories
+        have been installed on the guest.
+        """
+        for repository in self.get_repositories():
+            try:
+                nevras = guest.package_manager.list_packages(repository)
+            except Exception:
+                self.logger.warning(
+                    f"Failed to enumerate packages from repository '{repository.name}'."
+                )
+                continue
+            for nevra in nevras:
+                nevra = nevra.strip()
+                if not nevra:
+                    continue
+                try:
+                    self._artifacts.append(
+                        ArtifactInfo(
+                            version=RpmVersion.from_nevra(nevra),
+                            provider=self,
+                            location=repository.name,
+                        )
+                    )
+                except ValueError:
+                    self.logger.warning(f"Could not parse NEVRA '{nevra}'.")
+            self.logger.debug(
+                f"Enumerated {len(self._artifacts)} packages from repository '{repository.name}'."
+            )
+
     # B027: "... is an empty method in an abstract base class, but has
     # no abstract decorator" - expected, it's a default implementation
     # provided for subclasses. It is acceptable to do nothing.

tmt/steps/prepare/artifact/providers/_copr.py

@@ -0,0 +1,120 @@
+"""
+Shared COPR utilities.
+"""
+
+import re
+import types
+from abc import abstractmethod
+from functools import cached_property
+from typing import Any, Optional
+
+import tmt.log
+import tmt.utils
+import tmt.utils.hints
+from tmt.steps.prepare.artifact.providers import ArtifactProvider
+
+copr: Optional[types.ModuleType] = None
+
+# To silence mypy
+Client: Any
+
+tmt.utils.hints.register_hint(
+    'artifact-provider/copr',
+    """
+The ``copr`` Python package is required by tmt for Copr integration.
+
+To quickly test Copr presence, you can try running:
+
+    python -c 'import copr'
+
+* Users who installed tmt from PyPI should install the ``copr`` package
+  via ``pip install copr``.
+""",
+)
+
+
+COPR_URL = 'https://copr.fedorainfracloud.org/coprs'
+COPR_REPO_PATTERN = re.compile(r'^(@)?([^/]+)/([^/]+)$')
+
+
+def parse_copr_repo(copr_repo: str) -> tuple[bool, str, str]:
+    """
+    Parse a COPR repository identifier into its components.
+    """
+    matched = COPR_REPO_PATTERN.match(copr_repo)
+    if not matched:
+        raise tmt.utils.PrepareError(f"Invalid copr repository '{copr_repo}'.")
+    is_group, name, project = matched.groups()
+    return bool(is_group), name, project
+
+
+def build_copr_repo_url(copr_repo: str, chroot: str) -> str:
+    """
+    Construct the URL for a COPR ``.repo`` file.
+    """
+    is_group, name, project = parse_copr_repo(copr_repo)
+    group = 'group_' if is_group else ''
+    parts = [COPR_URL] + (['g'] if is_group else [])
+    parts += [name, project, 'repo', chroot]
+    parts += [f"{group}{name}-{project}-{chroot}.repo"]
+    return '/'.join(parts)
+
+
+def import_copr(logger: tmt.log.Logger) -> None:
+    """Import copr module with error handling."""
+    global copr, Client
+    try:
+        import copr
+        from copr.v3 import Client
+    except ImportError as error:
+        from tmt.utils.hints import print_hints
+
+        print_hints('artifact-provider/copr', logger=logger)
+
+        raise tmt.utils.GeneralError("Could not import copr package.") from error
+
+
+class CoprArtifactProvider(ArtifactProvider):
+    """
+    Base class for COPR-based artifact providers.
+    """
+
+    def __init__(self, raw_id: str, repository_priority: int, logger: tmt.log.Logger) -> None:
+        super().__init__(raw_id, repository_priority, logger)
+        self._session = self._initialize_session()
+
+    def _initialize_session(self) -> 'Client':
+        """
+        Initialize copr client session.
+        """
+        import_copr(self.logger)
+
+        try:
+            config = {"copr_url": "https://copr.fedorainfracloud.org"}
+            return Client(config)
+        except Exception as error:
+            raise tmt.utils.GeneralError("Failed to initialize Copr client session.") from error
+
+    @property
+    @abstractmethod
+    def _copr_owner(self) -> str:
+        raise NotImplementedError
+
+    @property
+    @abstractmethod
+    def _copr_project(self) -> str:
+        raise NotImplementedError
+
+    @cached_property
+    def project_info(self) -> Any:
+        """
+        Fetch and return the COPR project metadata.
+        """
+        try:
+            return self._session.project_proxy.get(
+                ownername=self._copr_owner, projectname=self._copr_project
+            )
+        except Exception as error:
+            raise tmt.utils.GeneralError(
+                f"Failed to fetch COPR project info for '{self._copr_owner}/{self._copr_project}'."
+            ) from error