sources: add an rpm source handler

.github/workflows/tests.yaml

@@ -95,6 +95,8 @@ 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
       - name: specify node version
         uses: actions/setup-node@v3
         with:

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,115 @@
+# -*- 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)
+        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,110 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2022 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,
+    )
+
+    # The asterisks in this glob allow this to be
+    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"