sources: add 7z source handler

Allow parts to specify 7z sources. Based on the implementation
from Snapcraft.

Co-authored-by: Tim Süberkrüb <[email protected]>
Co-authored-by: Kyle Fazzari <[email protected]>
Co-authored-by: Sergio Schvezov <[email protected]>
Signed-off-by: Claudio Matsuoka <[email protected]>

.github/workflows/tests.yaml

@@ -79,7 +79,7 @@ jobs:
           pip install -e .
       - name: Install additional test dependencies
         run: |
-          sudo apt install -y golang ninja-build cmake
+          sudo apt install -y golang ninja-build cmake p7zip-full
       - name: Run unit tests
         run: |
           make test-units

craft_parts/sources/base.py

@@ -120,9 +120,9 @@ def _run(cls, command: List[str], **kwargs):
             raise errors.PullError(command=command, exit_code=err.returncode)
 
     @classmethod
-    def _run_output(cls, command: Sequence) -> str:
+    def _run_output(cls, command: Sequence, **kwargs) -> str:
         try:
-            return subprocess.check_output(command, text=True).strip()
+            return subprocess.check_output(command, text=True, **kwargs).strip()
         except subprocess.CalledProcessError as err:
             raise errors.PullError(command=command, exit_code=err.returncode)
 

craft_parts/sources/sevenzip_source.py

@@ -0,0 +1,93 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2017 Tim Süberkrüb
+# Copyright 2018-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/>.
+
+"""Implement the 7zip file source handler."""
+
+import os
+from pathlib import Path
+from typing import List, Optional
+
+from craft_parts.dirs import ProjectDirs
+
+from . import errors
+from .base import FileSourceHandler
+
+
+class SevenzipSource(FileSourceHandler):
+    """The zip file source handler."""
+
+    # pylint: disable=too-many-arguments
+    def __init__(
+        self,
+        source: str,
+        part_src_dir: Path,
+        *,
+        cache_dir: Path,
+        source_tag: Optional[str] = None,
+        source_branch: Optional[str] = None,
+        source_commit: Optional[str] = None,
+        source_checksum: Optional[str] = None,
+        source_depth: Optional[int] = None,
+        source_submodules: Optional[List[str]] = 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_depth=source_depth,
+            source_submodules=source_submodules,
+            project_dirs=project_dirs,
+            ignore_patterns=ignore_patterns,
+            command="7zip",
+        )
+        if source_tag:
+            raise errors.InvalidSourceOption(source_type="7z", option="source-tag")
+
+        if source_commit:
+            raise errors.InvalidSourceOption(source_type="7z", option="source-commit")
+
+        if source_branch:
+            raise errors.InvalidSourceOption(source_type="7z", option="source-branch")
+
+        if source_depth:
+            raise errors.InvalidSourceOption(source_type="7z", option="source-depth")
+
+    # pylint: enable=too-many-arguments
+
+    def provision(
+        self,
+        dst: Path,
+        keep: bool = False,
+        src: Optional[Path] = None,
+    ):
+        """Extract 7z file contents to the part source dir."""
+        if src:
+            sevenzip_file = src
+        else:
+            sevenzip_file = Path(self.part_src_dir, os.path.basename(self.source))
+
+        sevenzip_file = sevenzip_file.expanduser().resolve()
+        self._run_output(["7z", "x", str(sevenzip_file)], cwd=dst)
+
+        if not keep:
+            os.remove(sevenzip_file)

craft_parts/sources/sources.py

@@ -89,6 +89,7 @@
 from .base import SourceHandler
 from .git_source import GitSource
 from .local_source import LocalSource
+from .sevenzip_source import SevenzipSource
 from .snap_source import SnapSource
 from .tar_source import TarSource
 from .zip_source import ZipSource
@@ -105,6 +106,7 @@
     "git": GitSource,
     "snap": SnapSource,
     "zip": ZipSource,
+    "7z": SevenzipSource,
 }
 
 

tests/integration/sources/test_sevenzip.py

@@ -0,0 +1,76 @@
+# -*- 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 subprocess
+import textwrap
+from pathlib import Path
+
+import pytest
+import yaml
+
+import craft_parts
+from craft_parts import Action, Step
+
+
+def test_source_sevenzip(new_dir):
+    _parts_yaml = textwrap.dedent(
+        """\
+        parts:
+          foo:
+            plugin: make
+            source: foobar.7z
+        """
+    )
+
+    Path("foobar.txt").write_text("content")
+    subprocess.run(["7z", "a", "foobar.7z", "foobar.txt"], check=True)
+
+    parts = yaml.safe_load(_parts_yaml)
+    lf = craft_parts.LifecycleManager(
+        parts, application_name="test_7z", cache_dir=new_dir
+    )
+
+    with lf.action_executor() as ctx:
+        ctx.execute(Action("foo", Step.PULL))
+
+    foo_src_dir = Path("parts", "foo", "src")
+    assert list(foo_src_dir.rglob("*")) == [foo_src_dir / "foobar.txt"]
+    assert Path(foo_src_dir, "foobar.txt").read_text() == "content"
+
+
+def test_source_sevenzip_error(new_dir):
+    _parts_yaml = textwrap.dedent(
+        """\
+        parts:
+          foo:
+            plugin: make
+            source: foobar.7z
+        """
+    )
+
+    parts = yaml.safe_load(_parts_yaml)
+    Path("foobar.7z").write_text("not a 7z file")
+    lf = craft_parts.LifecycleManager(
+        parts, application_name="test_7z", cache_dir=new_dir
+    )
+
+    with pytest.raises(craft_parts.PartsError) as raised, lf.action_executor() as ctx:
+        ctx.execute(Action("foo", Step.PULL))
+
+    assert raised.value.brief == (
+        f"Failed to pull source: command ['7z', 'x', "
+        f"'{new_dir}/parts/foo/src/foobar.7z'] exited with code 2."
+    )

tests/unit/sources/test_sevenzip_source.py

@@ -0,0 +1,99 @@
+# -*- Mode:Python; indent-tabs-mode:nil; tab-width:4 -*-
+#
+# Copyright 2017 Tim Süberkrüb
+# Copyright 2017-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 os.path
+import shutil
+import subprocess
+from pathlib import Path
+from unittest.mock import call
+
+import pytest
+
+from craft_parts.sources import sources
+
+
+@pytest.fixture
+def fake_7z_file(new_dir):
+    name = "fake-7z-file.7z"
+    Path(name).touch()
+    return name
+
+
+@pytest.fixture
+def part_src_dir(new_dir):
+    name = "part-src-dir"
+    Path(name).mkdir()
+    return name
+
+
+class TestSevenZip:
+    """Tests for the 7z source handler."""
+
+    def test_pull_7z_file_must_extract(
+        self, fake_7z_file, part_src_dir, new_dir, mocker
+    ):
+        check_output_mock = mocker.patch("subprocess.check_output")
+
+        sevenzip = sources.SevenzipSource(fake_7z_file, part_src_dir, cache_dir=new_dir)
+        sevenzip.pull()
+
+        assert check_output_mock.mock_calls == [
+            call(
+                ["7z", "x", os.path.join(new_dir, part_src_dir, fake_7z_file)],
+                text=True,
+                cwd=part_src_dir,
+            ),
+            call().strip(),
+        ]
+
+    def test_extract_and_keep_7zfile(self, fake_7z_file, part_src_dir, new_dir, mocker):
+        check_output_mock = mocker.patch("subprocess.check_output")
+
+        sevenzip = sources.SevenzipSource(fake_7z_file, part_src_dir, cache_dir=new_dir)
+        # This is the first step done by pull. We don't call pull to call the
+        # second step with a different keep_7z value.
+        shutil.copy2(sevenzip.source, sevenzip.part_src_dir)
+        sevenzip.provision(dst=part_src_dir, keep=True)
+
+        assert check_output_mock.mock_calls == [
+            call(
+                ["7z", "x", os.path.join(new_dir, part_src_dir, fake_7z_file)],
+                text=True,
+                cwd=part_src_dir,
+            ),
+            call().strip(),
+        ]
+        assert Path(fake_7z_file).exists()
+
+    def test_pull_failure(self, fake_7z_file, part_src_dir, new_dir, mocker):
+        check_output_mock = mocker.patch(
+            "subprocess.check_output",
+            side_effect=subprocess.CalledProcessError(1, "error"),
+        )
+        sevenzip = sources.SevenzipSource(fake_7z_file, part_src_dir, cache_dir=new_dir)
+
+        with pytest.raises(sources.errors.PullError) as raised:
+            sevenzip.pull()
+
+        assert check_output_mock.mock_calls == [
+            call(
+                ["7z", "x", os.path.join(new_dir, part_src_dir, fake_7z_file)],
+                text=True,
+                cwd="part-src-dir",
+            )
+        ]
+        assert raised.value.exit_code == 1