Merge pull request #1204 from kapicorp/fix_copy_tree

Fix copy tree

kapitan/utils.py

@@ -628,12 +628,18 @@ def safe_copy_tree(src, dst):
     return outputs
 
 
-def copy_tree(src, dst):
+def copy_tree(src: str, dst: str) -> list:
     """Recursively copy a given directory from `src` to `dst`.
 
     Returns a list of the copied files.
     """
-    before = set(glob.iglob("*", recursive=True))
+    if not os.path.isdir(src):
+        raise SafeCopyError(f"Cannot copy tree {src}: not a directory")
+
+    if not os.path.isdir(dst):
+        raise SafeCopyError(f"Cannot copy tree {dst}: not a directory")
+
+    before = set(glob.iglob(f"{dst}/*", recursive=True))
     shutil.copytree(src, dst, dirs_exist_ok=True)
-    after = set(glob.iglob("*", recursive=True))
+    after = set(glob.iglob(f"{dst}/*", recursive=True))
     return list(after - before)

tests/test_compile.py

@@ -16,12 +16,9 @@
 import shutil
 import yaml
 import toml
-import tempfile
 from kapitan.cli import main
-from kapitan.resources import get_inventory
 from kapitan.utils import directory_hash
 from kapitan.cached import reset_cache
-from kapitan.errors import InventoryError
 
 TEST_PWD = os.getcwd()
 TEST_RESOURCES_PATH = os.path.join(os.getcwd(), "tests/test_resources")

tests/test_inventory.py

@@ -23,6 +23,7 @@
 
 class InventoryTargetTestBase(unittest.TestCase):
 
+
     def setUp(self) -> None:
         from kapitan.cached import reset_cache, args
 

tests/test_utils.py

@@ -0,0 +1,48 @@
+#!/usr/bin/env python3
+
+# Copyright 2019 The Kapitan Authors
+# SPDX-FileCopyrightText: 2020 The Kapitan Authors <[email protected]>
+#
+# SPDX-License-Identifier: Apache-2.0
+
+"utils tests"
+
+import unittest
+import os
+import tempfile
+import glob
+import shutil
+
+from kapitan.utils import copy_tree, directory_hash, SafeCopyError
+
+TEST_PWD = os.getcwd()
+TEST_RESOURCES_PATH = os.path.join(os.getcwd(), "tests/test_resources")
+TEST_DOCKER_PATH = os.path.join(os.getcwd(), "examples/docker/")
+TEST_TERRAFORM_PATH = os.path.join(os.getcwd(), "examples/terraform/")
+TEST_KUBERNETES_PATH = os.path.join(os.getcwd(), "examples/kubernetes/")
+
+
+class CopyTreeTest(unittest.TestCase):
+    "Test copy_tree function"
+
+    def setUp(self):
+        self.temp_dir = tempfile.mkdtemp()
+
+
+    def test_copy_dir(self):
+        original = set(glob.iglob(f"{TEST_KUBERNETES_PATH}/*", recursive=True))
+        copied = copy_tree(TEST_KUBERNETES_PATH, self.temp_dir)
+        self.assertEqual(len(copied), len(original))
+
+        original_hash = directory_hash(TEST_KUBERNETES_PATH)
+        copied_hash = directory_hash(self.temp_dir)
+        self.assertEqual(copied_hash, original_hash)
+
+
+    def test_validate_copy_dir(self):
+        with self.assertRaises(SafeCopyError):
+            copy_tree("non_existent_dir", self.temp_dir)
+
+    def tearDown(self):
+        shutil.rmtree(self.temp_dir)
+