refactor: replace distutils' copy_tree with shutil copytree

kapitan/initialiser.py

@@ -9,30 +9,34 @@
 
 import logging
 import os
-from distutils.dir_util import copy_tree
+from pathlib import Path
+
+from kapitan.utils import copy_tree
 
 logger = logging.getLogger(__name__)
 
 
 def initialise_skeleton(args):
-    """Initialises a directory with a recommended skeleton structure
+    """Initialises a directory with a recommended skeleton structure and prints list of
+    copied files.
+
     Args:
-        args.directory (string): path which to initialise, directory is assumed to exist
+        args.directory (string): path which to initialise, directory is assumed to exist.
     """
+    templates = Path(__file__).resolve().parent.joinpath("inputs", "templates")
+    destination = Path(args.directory)
+    copied_files = copy_tree(source=templates, destination=destination, dirs_exist_ok=True)
 
-    current_pwd = os.path.dirname(__file__)
-    templates_directory = os.path.join(current_pwd, "inputs", "templates")
-    populated = copy_tree(templates_directory, args.directory)
     logger.info("Populated %s with:", args.directory)
-    for directory, subs, file_list in os.walk(args.directory):
+    for directory, _, file_list in os.walk(args.directory):
         # In order to avoid adding the given directory itself in listing.
         if directory == args.directory:
             continue
-        if any([path.startswith(directory) for path in populated]):
+        if any([path.startswith(directory) for path in copied_files]):
             level = directory.replace(args.directory, "").count(os.sep) - 1
             indent = " " * 4 * (level)
             logger.info("%s%s", indent, os.path.basename(directory))
             for fname in file_list:
-                if os.path.join(directory, fname) in populated:
+                if os.path.join(directory, fname) in copied_files:
                     sub_indent = " " * 4 * (level + 1)
                     logger.info("%s%s", sub_indent, fname)

kapitan/utils.py

@@ -23,6 +23,9 @@
 from distutils.file_util import _copy_file_contents
 from functools import lru_cache, wraps
 from hashlib import sha256
+from pathlib import Path
+from shutil import copytree
+from typing import List
 from zipfile import ZipFile
 
 import jinja2
@@ -631,3 +634,19 @@ def safe_copy_tree(src, dst):
             outputs.append(dst_name)
 
     return outputs
+
+
+def copy_tree(source: Path, destination: Path, dirs_exist_ok: bool = False) -> List[str]:
+    """Recursively copy a given directory from `source` to `destination` using shutil.copytree
+    and return list of copied files. When `dirs_exist_ok` is set, the `FileExistsError` is
+    ignored when destination directory exists.
+    Args:
+        source (str): Path to a source directory
+        destination (str): Path to a destination directory
+    Returns:
+        list[str]: List of copied files
+    """
+    inventory_before = list(destination.rglob("*"))
+    copytree(source, destination, dirs_exist_ok=dirs_exist_ok)
+    inventory_after = list(destination.rglob("*"))
+    return [str(d) for d in inventory_after if d not in inventory_before]

tests/test_utils.py

@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+
+# Copyright 2019 The Kapitan Authors
+# SPDX-FileCopyrightText: 2020 The Kapitan Authors <kapitan-admins@googlegroups.com>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+from pathlib import Path
+import tempfile
+from typing import List
+import unittest
+
+from kapitan.utils import copy_tree
+
+
+class CopytreeTest(unittest.TestCase):
+    def _create_directory_structure(self, root: Path, structure: List[str]):
+        for sub_path in structure:
+            full_path = root / sub_path
+            full_path.parent.mkdir(parents=True, exist_ok=True)
+            full_path.touch(exist_ok=True)
+
+    def setUp(self):
+        self.template_structure = [
+            "components/my_component/my_component.jsonnet",
+            "components/other_component/__init__.py",
+            "inventory/classes/common.yml",
+            "inventory/classes/my_component.yml",
+            "templates/docs/my_readme.md",
+            "templates/scripts/my_script.sh",
+        ]
+        self.template_directory = tempfile.TemporaryDirectory()
+        self.template_directory_path = Path(self.template_directory.name)
+
+        self._create_directory_structure(self.template_directory_path, self.template_structure)
+
+    def tearDown(self):
+        self.template_directory.cleanup()
+
+    def test_empty_dir(self):
+        """Validates that all template files were copied to the correct directory and
+        the right list of copied files was returned."""
+        with tempfile.TemporaryDirectory() as destination_directory:
+            template_files = [destination_directory + "/" + f for f in self.template_structure]
+
+            copied_paths = copy_tree(
+                source=self.template_directory_path,
+                destination=Path(destination_directory),
+                dirs_exist_ok=True,
+            )
+            # We only care about files as they will have full path anyway.
+            copied_files = [p for p in copied_paths if Path(p).is_file()]
+
+            self.assertSequenceEqual(sorted(copied_files), sorted(template_files))
+            self.assertTrue([Path(f).exists() for f in template_files])
+
+    def test_non_empty_dir(self):
+        """Makes sure that existing directories and files are untouched."""
+        with tempfile.TemporaryDirectory() as destination_directory:
+            extra_files = [
+                "file_that_existed_before.txt",
+                "this_existed_before/file.txt",
+            ]
+            self._create_directory_structure(Path(destination_directory), extra_files)
+            template_files = [destination_directory + "/" + f for f in self.template_structure]
+
+            copied_paths = copy_tree(
+                source=self.template_directory_path,
+                destination=Path(destination_directory),
+                dirs_exist_ok=True,
+            )
+            # We only care about files as they will have full path anyway.
+            copied_files = [p for p in copied_paths if Path(p).is_file()]
+
+            self.assertSequenceEqual(sorted(copied_files), sorted(template_files))
+            self.assertTrue([Path(f).exists() for f in template_files])
+            self.assertTrue([Path(destination_directory + "/" + f).exists() for f in extra_files])
+
+
+if __name__ == "__main__":
+    unittest.main()