Add Copier and GitCopier.

src/antsibull_fileutils/copier.py

@@ -0,0 +1,97 @@
+# Author: Felix Fontein <felix@fontein.de>
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or
+# https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+# SPDX-FileCopyrightText: 2024, Ansible Project
+
+"""
+Directory and collection copying helpers.
+"""
+
+from __future__ import annotations
+
+import os
+import shutil
+import typing as t
+
+from antsibull_fileutils.vcs import list_git_files
+
+if t.TYPE_CHECKING:
+    from _typeshed import StrPath
+
+
+class CopierError(Exception):
+    pass
+
+
+class Copier:
+    """
+    Allows to copy directories.
+    """
+
+    def __init__(self, *, log_debug: t.Callable[[str], None] | None = None):
+        self._log_debug = log_debug
+
+    def _do_log_debug(self, msg: str, *args: t.Any) -> None:
+        if self._log_debug:
+            self._log_debug(msg, *args)
+
+    def copy(self, from_path: StrPath, to_path: StrPath) -> None:
+        """
+        Copy a directory ``from_path`` to a destination ``to_path``.
+
+        ``to_path`` must not exist, but its parent directory must exist.
+        """
+        self._do_log_debug(
+            "Copying complete directory from {!r} to {!r}", from_path, to_path
+        )
+        shutil.copytree(from_path, to_path, symlinks=True)
+
+
+class GitCopier(Copier):
+    """
+    Allows to copy directories that are part of a Git repository.
+    """
+
+    def __init__(
+        self,
+        *,
+        git_bin_path: StrPath = "git",
+        log_debug: t.Callable[[str], None] | None = None,
+    ):
+        super().__init__(log_debug=log_debug)
+        self.git_bin_path = git_bin_path
+
+    def copy(self, from_path: StrPath, to_path: StrPath) -> None:
+        self._do_log_debug("Identifying files not ignored by Git in {!r}", from_path)
+        try:
+            files = list_git_files(
+                from_path, git_bin_path=self.git_bin_path, log_debug=self._log_debug
+            )
+        except ValueError as exc:
+            raise CopierError(
+                f"Error while listing files not ignored by Git in {from_path}: {exc}"
+            ) from exc
+
+        self._do_log_debug(
+            "Copying {} file(s) from {!r} to {!r}", len(files), from_path, to_path
+        )
+        os.mkdir(to_path, mode=0o700)
+        created_directories = set()
+        for file in files:
+            # Decode filename and check whether the file still exists
+            # (deleted files are part of the output)
+            file_decoded = file.decode("utf-8")
+            src_path = os.path.join(from_path, file_decoded)
+            if not os.path.exists(src_path):
+                continue
+
+            # Check whether the directory for this file exists
+            directory, _ = os.path.split(file_decoded)
+            if directory not in created_directories:
+                os.makedirs(os.path.join(to_path, directory), mode=0o700, exist_ok=True)
+                created_directories.add(directory)
+
+            # Copy the file
+            dst_path = os.path.join(to_path, file_decoded)
+            shutil.copyfile(src_path, dst_path)

tests/units/test_copier.py

@@ -0,0 +1,139 @@
+# Author: Felix Fontein <felix@fontein.de>
+# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt)
+# SPDX-License-Identifier: GPL-3.0-or-later
+# SPDX-FileCopyrightText: 2024, Ansible Project
+
+"""
+Test utils module.
+"""
+
+from __future__ import annotations
+
+import pathlib
+import re
+from unittest import mock
+
+import pytest
+
+from antsibull_fileutils.copier import Copier, CopierError, GitCopier
+
+from .utils import collect_log
+
+
+def assert_same(a: pathlib.Path, b: pathlib.Path) -> None:
+    if a.is_file():
+        assert b.is_file()
+        assert a.read_bytes() == b.read_bytes()
+        return
+    if a.is_dir():
+        assert b.is_dir()
+        return
+    if a.is_symlink():
+        assert b.is_symlink()
+        assert a.readlink() == b.readlink()
+        return
+
+
+def test_copier(tmp_path_factory):
+    directory: pathlib.Path = tmp_path_factory.mktemp("changelog-test")
+
+    src_dir = directory / "src"
+    dest_dir = directory / "dest"
+
+    with mock.patch(
+        "antsibull_fileutils.copier.shutil.copytree",
+        return_value=None,
+    ) as m:
+        copier = Copier()
+        copier.copy(str(src_dir), str(dest_dir))
+        m.assert_called_with(str(src_dir), str(dest_dir), symlinks=True)
+
+    with mock.patch(
+        "antsibull_fileutils.copier.shutil.copytree",
+        return_value=None,
+    ) as m:
+        kwargs, debug, info = collect_log(with_info=False)
+        copier = Copier(**kwargs)
+        copier.copy(str(src_dir), str(dest_dir))
+        m.assert_called_with(str(src_dir), str(dest_dir), symlinks=True)
+        assert debug == [
+            (
+                "Copying complete directory from {!r} to {!r}",
+                (str(src_dir), str(dest_dir)),
+            ),
+        ]
+
+    with mock.patch(
+        "antsibull_fileutils.copier.shutil.copytree",
+        return_value=None,
+    ) as m:
+        kwargs, debug, info = collect_log(with_info=False)
+        copier = Copier(**kwargs)
+        copier.copy(src_dir, dest_dir)
+        m.assert_called_with(src_dir, dest_dir, symlinks=True)
+        assert debug == [
+            ("Copying complete directory from {!r} to {!r}", (src_dir, dest_dir)),
+        ]
+
+
+def test_git_copier(tmp_path_factory):
+    directory: pathlib.Path = tmp_path_factory.mktemp("changelog-test")
+
+    src_dir = directory / "src"
+    src_dir.mkdir()
+    (src_dir / "empty").touch()
+    (src_dir / "link").symlink_to("empty")
+    (src_dir / "dir").mkdir()
+    (src_dir / "file").write_text("content", encoding="utf-8")
+    (src_dir / "dir" / "binary_file").write_bytes(b"\x00\x01\x02")
+    (src_dir / "dir" / "another_file").write_text("more", encoding="utf-8")
+
+    dest_dir = directory / "dest1"
+    with mock.patch(
+        "antsibull_fileutils.copier.list_git_files",
+        return_value=[b"file"],
+    ) as m:
+        copier = GitCopier(git_bin_path="/path/to/git")
+        copier.copy(str(src_dir), str(dest_dir))
+        m.assert_called_with(str(src_dir), git_bin_path="/path/to/git", log_debug=None)
+        assert dest_dir.is_dir()
+        assert {p.name for p in dest_dir.iterdir()} == {"file"}
+        assert_same(src_dir / "file", dest_dir / "file")
+
+    dest_dir = directory / "dest2"
+    with mock.patch(
+        "antsibull_fileutils.copier.list_git_files",
+        return_value=[b"link", b"foobar", b"dir/binary_file", b"dir/another_file"],
+    ) as m:
+        kwargs, debug, info = collect_log(with_info=False)
+        copier = GitCopier(git_bin_path="/path/to/git", **kwargs)
+        copier.copy(src_dir, dest_dir)
+        m.assert_called_with(src_dir, git_bin_path="/path/to/git", **kwargs)
+
+        assert debug == [
+            ("Identifying files not ignored by Git in {!r}", (src_dir,)),
+            ("Copying {} file(s) from {!r} to {!r}", (4, src_dir, dest_dir)),
+        ]
+        assert dest_dir.is_dir()
+        assert {p.name for p in dest_dir.iterdir()} == {"link", "dir"}
+        assert_same(src_dir / "link", dest_dir / "link")
+        assert_same(src_dir / "dir", dest_dir / "dir")
+        assert {p.name for p in (dest_dir / "dir").iterdir()} == {
+            "another_file",
+            "binary_file",
+        }
+        assert_same(src_dir / "dir" / "another_file", dest_dir / "dir" / "another_file")
+        assert_same(src_dir / "dir" / "binary_file", dest_dir / "dir" / "binary_file")
+
+    dest_dir = directory / "dest2"
+    with mock.patch(
+        "antsibull_fileutils.copier.list_git_files",
+        side_effect=ValueError("nada"),
+    ) as m:
+        copier = GitCopier(git_bin_path="/path/to/git")
+        with pytest.raises(
+            CopierError,
+            match=f"^Error while listing files not ignored by Git in {re.escape(str(src_dir))}: nada$",
+        ) as exc:
+            copier.copy(str(src_dir), str(dest_dir))
+        m.assert_called_with(str(src_dir), git_bin_path="/path/to/git", log_debug=None)