sources: add an rpm source handler

.github/workflows/tests.yaml

@@ -95,9 +95,11 @@ jobs:
           # Remove newer go and install regular version for 20.04
           sudo snap remove go
           sudo apt install -y golang
+          # Install RPM dependencies for RPM tests
+          sudo apt install rpm
           # Ensure we don't have dotnet installed, to properly test dotnet-deps
           # Based on https://github.com/actions/runner-images/blob/main/images/linux/scripts/installers/dotnetcore-sdk.sh
-          sudo apt remove dotnet-* || true
+          sudo apt remove -y dotnet-* || true
           # Remove manually-installed dotnet from tarballs
           sudo rm -rf /usr/share/dotnet
           # Remove dotnet tools

craft_parts/packages/deb.py

@@ -370,7 +370,7 @@ def get_package_libraries(cls, package_name: str) -> Set[str]:
     @classmethod
     @_apt_cache_wrapper
     def get_packages_for_source_type(cls, source_type: str) -> Set[str]:
-        """Return a list of packages required to to work with source_type."""
+        """Return the packages required to work with source_type."""
         if source_type == "bzr":
             packages = {"bzr"}
         elif source_type == "git":
@@ -383,6 +383,8 @@ def get_packages_for_source_type(cls, source_type: str) -> Set[str]:
             packages = {"subversion"}
         elif source_type == "rpm2cpio":
             packages = {"rpm2cpio"}
+        elif source_type == "rpm":
+            packages = {"rpm"}
         elif source_type == "7zip":
             packages = {"p7zip-full"}
         else:

craft_parts/packages/yum.py

@@ -54,7 +54,7 @@ def get_package_libraries(
 
     @classmethod
     def get_packages_for_source_type(cls, source_type: str) -> Set[str]:
-        """Return a list of packages required to to work with source_type."""
+        """Return a list of packages required to work with source_type."""
         if source_type == "bzr":
             packages = {"bzr"}
         elif source_type == "git":
@@ -65,11 +65,13 @@ def get_packages_for_source_type(cls, source_type: str) -> Set[str]:
             packages = {"mercurial"}
         elif source_type in ["svn", "subversion"]:
             packages = {"subversion"}
-        elif source_type == "rpm2cpio":
-            # installed by default in CentOS systems
+        elif source_type in ["rpm2cpio", "rpm"]:
+            # installed by default in CentOS systems by the rpm package
             packages = set()
         elif source_type == "7zip":
             packages = {"p7zip"}
+        elif source_type == "deb":
+            raise NotImplementedError("Deb files not yet supported on this base.")
         else:
             packages = set()
 

craft_parts/sources/errors.py

@@ -58,6 +58,27 @@ def __init__(self, *, source_type: str, option: str):
         super().__init__(brief=brief, resolution=resolution)
 
 
+# TODO: Merge this with InvalidSourceOption above
+class InvalidSourceOptions(SourceError):
+    """A source option is not allowed for the given source type.
+
+    :param source_type: The part's source type.
+    :param options: The invalid source options.
+    """
+
+    def __init__(self, *, source_type: str, options: List[str]):
+        self.source_type = source_type
+        self.options = options
+        humanized_options = formatting_utils.humanize_list(options, "and")
+        brief = (
+            f"Failed to pull source: {humanized_options} cannot be used "
+            f"with a {source_type} source."
+        )
+        resolution = "Make sure sources are correctly specified."
+
+        super().__init__(brief=brief, resolution=resolution)
+
+
 class IncompatibleSourceOptions(SourceError):
     """Source specified options that cannot be used at the same time.
 
@@ -148,6 +169,20 @@ def __init__(self, snap_file: str):
         super().__init__(brief=brief, resolution=resolution)
 
 
+class InvalidRpmPackage(SourceError):
+    """An rpm package is invalid.
+
+    :param rpm_file: The filename.
+    """
+
+    def __init__(self, rpm_file: str):
+        self.rpm_file = rpm_file
+        brief = f"RPM file {rpm_file!r} could not be extracted."
+        resolution = "Ensure the source lists a valid rpm file."
+
+        super().__init__(brief=brief, resolution=resolution)
+
+
 class PullError(SourceError):
     """Failed pulling source.
 

craft_parts/sources/rpm_source.py

@@ -0,0 +1,119 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2023 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 3 as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+"""The RPM source handler."""
+
+import logging
+import os
+import subprocess
+import tarfile
+from pathlib import Path
+from typing import List, Optional
+
+from overrides import override
+
+from craft_parts.dirs import ProjectDirs
+
+from . import errors
+from .base import FileSourceHandler
+
+logger = logging.getLogger(__name__)
+
+
+class RpmSource(FileSourceHandler):
+    """The "rpm" file source handler."""
+
+    _invalid_tags = (
+        "source-tag",
+        "source-commit",
+        "source-branch",
+        "source-depth",
+        "source-submodules",
+    )
+    _source_type = "rpm"
+
+    def __init__(
+        self,
+        source: str,
+        part_src_dir: Path,
+        *,
+        cache_dir: Path,
+        source_tag: None = None,
+        source_commit: None = None,
+        source_branch: None = None,
+        source_checksum: Optional[str] = None,
+        source_submodules: None = None,
+        source_depth: None = None,
+        project_dirs: Optional[ProjectDirs] = None,
+        ignore_patterns: Optional[List[str]] = None,
+    ):
+        super().__init__(
+            source,
+            part_src_dir,
+            cache_dir=cache_dir,
+            source_tag=source_tag,
+            source_branch=source_branch,
+            source_commit=source_commit,
+            source_checksum=source_checksum,
+            source_submodules=source_submodules,
+            source_depth=source_depth,
+            project_dirs=project_dirs,
+            ignore_patterns=ignore_patterns,
+        )
+
+        self._validate()
+
+    def _validate(self) -> None:
+        """Validate this source.
+
+        :raises: InvalidSourceOptions if any bad options are used.
+        """
+        bad_options = []
+        for tag in self._invalid_tags:
+            if getattr(self, tag.replace("-", "_")):
+                bad_options.append(tag)
+        if bad_options:
+            raise errors.InvalidSourceOptions(
+                source_type=self._source_type, options=bad_options
+            )
+
+    @override
+    def provision(
+        self,
+        dst: Path,
+        keep: bool = False,
+        src: Optional[Path] = None,
+    ) -> None:
+        """Extract rpm file contents to the part source dir."""
+        rpm_path = src or self.part_src_dir / os.path.basename(self.source)
+        # NOTE: rpm2archive chosen here because while it's slower, it has broader
+        # compatibility than rpm2cpio.
+        # --nocompression parameter excluded until all supported platforms
+        # include rpm >= 4.17
+        command = ["rpm2archive", "-"]
+
+        with rpm_path.open("rb") as rpm:
+            try:
+                with subprocess.Popen(
+                    command, stdin=rpm, stdout=subprocess.PIPE, stderr=subprocess.PIPE
+                ) as archive:
+                    with tarfile.open(mode="r|*", fileobj=archive.stdout) as tar:
+                        tar.extractall(path=dst)
+            except (tarfile.TarError, subprocess.CalledProcessError) as err:
+                raise errors.InvalidRpmPackage(rpm_path.name) from err
+
+        if not keep:
+            rpm_path.unlink()

craft_parts/sources/sources.py

@@ -91,6 +91,7 @@
 from .file_source import FileSource
 from .git_source import GitSource
 from .local_source import LocalSource
+from .rpm_source import RpmSource
 from .snap_source import SnapSource
 from .tar_source import TarSource
 from .zip_source import ZipSource
@@ -109,6 +110,7 @@
     "zip": ZipSource,
     "deb": DebSource,
     "file": FileSource,
+    "rpm": RpmSource,
 }
 
 

tests/integration/sources/test_rpm.py

@@ -0,0 +1,109 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2023 Canonical Ltd.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU Lesser General Public
+# License version 3 as published by the Free Software Foundation.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
+# Lesser General Public License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+import platform
+import shutil
+import subprocess
+import textwrap
+from pathlib import Path
+
+import pytest
+import yaml
+
+import craft_parts
+from craft_parts.actions import Action
+from craft_parts.steps import Step
+
+
+@pytest.fixture
+def sample_rpm(tmp_path: Path) -> Path:
+    """
+    Create a basic .rpm file and return its path.
+    """
+    rpm_dir = tmp_path / "sample"
+    rpm_dir.mkdir()
+
+    # Add spec structure
+    spec_dir = rpm_dir / "SPECS"
+    spec_dir.mkdir()
+    spec_file = spec_dir / "sample.spec"
+    spec_file.write_text(
+        textwrap.dedent(
+            """
+            Name: sample
+            Version: 1.0
+            Release: 0
+            Summary: A sample package
+            License: GPL
+
+            %description
+            A little sample package!
+
+            %install
+            mkdir -p %{buildroot}/etc
+            bash -c "echo Sample contents > %{buildroot}/etc/sample.txt"
+
+            %files
+            /etc/sample.txt
+            """
+        )
+    )
+
+    # Define paths wo we don't litter the system with RPM stuff for this.
+    rpmbuild_params = [
+        f"--define=_topdir {rpm_dir}/build",
+        f"--define=_dbpath {rpm_dir}/rpmdb",
+        f"--define=_var {tmp_path}/var",
+        f"--define=_tmppath {tmp_path}/tmp",
+    ]
+
+    subprocess.run(
+        ["rpmbuild", "-bb", "--verbose"] + rpmbuild_params + [str(spec_file)],
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+
+    arch = platform.machine()
+    rpm_path = rpm_dir / "build/RPMS" / arch / f"sample-1.0-0.{arch}.rpm"
+    if not rpm_path.exists():
+        raise FileNotFoundError("rpmbuild did not create the correct file.")
+    return rpm_path
+
+
+@pytest.mark.skipif(not shutil.which("rpmbuild"), reason="rpmbuild is not installed")
+def test_source_rpm(sample_rpm, tmp_path):
+    parts_yaml = textwrap.dedent(
+        f"""\
+        parts:
+          foo:
+            plugin: nil
+            source: {sample_rpm}
+        """
+    )
+
+    result_dir = tmp_path / "result"
+    result_dir.mkdir()
+
+    parts = yaml.safe_load(parts_yaml)
+    lf = craft_parts.LifecycleManager(
+        parts, application_name="test_rpm", cache_dir=tmp_path, work_dir=result_dir
+    )
+
+    with lf.action_executor() as ctx:
+        ctx.execute(Action("foo", Step.PULL))
+
+    expected_file = result_dir / "parts/foo/src/etc/sample.txt"
+    assert expected_file.read_text() == "Sample contents\n"

tests/unit/packages/test_deb.py

@@ -481,6 +481,27 @@ def test_refresh_packages_list_fails(self, fake_deb_run):
         fake_deb_run.assert_has_calls([call(["apt-get", "update"])])
 
 
+@pytest.mark.parametrize(
+    "source_type, packages",
+    [
+        ("7zip", {"p7zip-full"}),
+        ("bzr", {"bzr"}),
+        ("git", {"git"}),
+        ("hg", {"mercurial"}),
+        ("mercurial", {"mercurial"}),
+        ("rpm2cpio", {"rpm2cpio"}),
+        ("rpm", {"rpm"}),
+        ("subversion", set()),
+        ("svn", set()),
+        ("tar", {"tar"}),
+        ("deb", set()),
+        ("whatever-unknown", set()),
+    ],
+)
+def test_packages_for_source_type(source_type, packages):
+    assert deb.Ubuntu.get_packages_for_source_type(source_type) == packages
+
+
 @pytest.fixture
 def fake_dpkg_query(mocker):
     def dpkg_query(*args, **kwargs):

tests/unit/packages/test_yum.py

@@ -91,6 +91,7 @@ def test_refresh_packages_list(fake_yum_run):
         ("hg", {"mercurial"}),
         ("mercurial", {"mercurial"}),
         ("rpm2cpio", set()),
+        ("rpm", set()),
         ("subversion", {"subversion"}),
         ("svn", {"subversion"}),
         ("tar", {"tar"}),
@@ -101,6 +102,11 @@ def test_packages_for_source_type(source_type, packages):
     assert YUMRepository.get_packages_for_source_type(source_type) == packages
 
 
+def test_deb_source_type_not_implemented():
+    with pytest.raises(NotImplementedError):
+        YUMRepository.get_packages_for_source_type("deb")
+
+
 # -- tests for methods left out of the YUMRepository MVP