Add collection copier.

src/antsibull_fileutils/copier.py

@@ -12,6 +12,7 @@
 
 import os
 import shutil
+import tempfile
 import typing as t
 
 from antsibull_fileutils.vcs import list_git_files
@@ -95,3 +96,55 @@ def copy(self, from_path: StrPath, to_path: StrPath) -> None:
             # Copy the file
             dst_path = os.path.join(to_path, file_decoded)
             shutil.copyfile(src_path, dst_path)
+
+
+class CollectionCopier:
+    """
+    Creates a copy of a collection to a place where ``--playbook-dir`` can be used
+    to prefer this copy of the collection over any installed ones.
+    """
+
+    def __init__(
+        self,
+        *,
+        source_directory: str,
+        namespace: str,
+        name: str,
+        copier: Copier,
+        log_debug: t.Callable[[str], None] | None = None,
+    ):
+        self.source_directory = source_directory
+        self.namespace = namespace
+        self.name = name
+        self.copier = copier
+        self._log_debug = log_debug
+
+        self.dir = os.path.realpath(tempfile.mkdtemp(prefix="antsibull-fileutils"))
+
+    def _do_log_debug(self, msg: str, *args: t.Any) -> None:
+        if self._log_debug:
+            self._log_debug(msg, *args)
+
+    def __enter__(self) -> tuple[str, str]:
+        try:
+            collection_container_dir = os.path.join(
+                self.dir, "collections", "ansible_collections", self.namespace
+            )
+            os.makedirs(collection_container_dir)
+
+            collection_dir = os.path.join(collection_container_dir, self.name)
+            self._do_log_debug("Temporary collection directory: {!r}", collection_dir)
+
+            self.copier.copy(self.source_directory, collection_dir)
+
+            self._do_log_debug("Temporary collection directory has been populated")
+            return (
+                self.dir,
+                collection_dir,
+            )
+        except Exception:
+            shutil.rmtree(self.dir, ignore_errors=True)
+            raise
+
+    def __exit__(self, type_, value, traceback_):
+        shutil.rmtree(self.dir, ignore_errors=True)

tests/units/test_copier.py

@@ -9,13 +9,14 @@
 
 from __future__ import annotations
 
+import os
 import pathlib
 import re
 from unittest import mock
 
 import pytest
 
-from antsibull_fileutils.copier import Copier, CopierError, GitCopier
+from antsibull_fileutils.copier import CollectionCopier, Copier, CopierError, GitCopier
 
 from .utils import collect_log
 
@@ -35,7 +36,7 @@ def assert_same(a: pathlib.Path, b: pathlib.Path) -> None:
 
 
 def test_copier(tmp_path_factory):
-    directory: pathlib.Path = tmp_path_factory.mktemp("changelog-test")
+    directory: pathlib.Path = tmp_path_factory.mktemp("copier")
 
     src_dir = directory / "src"
     dest_dir = directory / "dest"
@@ -77,7 +78,7 @@ def test_copier(tmp_path_factory):
 
 
 def test_git_copier(tmp_path_factory):
-    directory: pathlib.Path = tmp_path_factory.mktemp("changelog-test")
+    directory: pathlib.Path = tmp_path_factory.mktemp("git-copier")
 
     src_dir = directory / "src"
     src_dir.mkdir()
@@ -137,3 +138,52 @@ def test_git_copier(tmp_path_factory):
         ) 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)
+
+
+def test_collection_copier(tmp_path_factory):
+    src_dir: pathlib.Path = tmp_path_factory.mktemp("collection-copier")
+
+    copier = mock.MagicMock()
+
+    with CollectionCopier(
+        source_directory=src_dir, namespace="foo", name="bar", copier=copier
+    ) as (root_dir, collection_dir):
+        assert (
+            os.path.join(root_dir, "collections", "ansible_collections", "foo", "bar")
+            == collection_dir
+        )
+        assert os.path.exists(root_dir)
+        assert os.path.exists(
+            os.path.join(root_dir, "collections", "ansible_collections", "foo")
+        )
+        assert not os.path.exists(collection_dir)  # our mock doesn't create it
+
+    assert not os.path.exists(root_dir)
+    copier.copy.assert_called_with(src_dir, collection_dir)
+
+
+def test_collection_copier_fail():
+    copier = mock.MagicMock()
+    copier.copy = mock.MagicMock(side_effect=CopierError("boo"))
+
+    kwargs, debug, info = collect_log(with_info=False)
+
+    cc = CollectionCopier(
+        source_directory="/foo", namespace="foo", name="bar", copier=copier, **kwargs
+    )
+    assert os.path.exists(cc.dir)
+
+    with pytest.raises(CopierError, match="^boo$"):
+        with cc as (root_dir, collection_dir):
+            assert False
+
+    copier.copy.assert_called_with(
+        "/foo", os.path.join(cc.dir, "collections", "ansible_collections", "foo", "bar")
+    )
+    assert not os.path.exists(cc.dir)
+    assert debug == [
+        (
+            "Temporary collection directory: {!r}",
+            (os.path.join(cc.dir, "collections", "ansible_collections", "foo", "bar"),),
+        )
+    ]