Added delete_dir_contents function to cloudinit.util.

2 files changed, 85 insertions, 1 deletions

diff --git a/cloudinit/util.py b/cloudinit/util.py
index b690d517..3701f42d 100644
--- a/cloudinit/util.py
+++ b/cloudinit/util.py
@@ -18,6 +18,7 @@
 import yaml
 import os
 import os.path
+import shutil
 import errno
 import subprocess
 from Cheetah.Template import Template
@@ -127,6 +128,19 @@ def mergedict(src, cand):
                 src[k] = mergedict(src[k], v)
     return src
 
+def delete_dir_contents(dirname):
+    """
+    Deletes all contents of a directory without deleting the directory itself.
+
+    @param dirname: The directory whose contents should be deleted.
+    """
+    for node in os.listdir(dirname):
+        node_fullpath = os.path.join(dirname, node)
+        if os.path.isdir(node_fullpath):
+            shutil.rmtree(node_fullpath)
+        else:
+            os.unlink(node_fullpath)
+
 def write_file(filepath, content, mode=0644, omode="wb"):
     """
     Writes a file with the given content and sets the file mode as specified.
diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py
index 4c512990..f2b2ee3d 100644
--- a/tests/unittests/test_util.py
+++ b/tests/unittests/test_util.py
@@ -5,7 +5,7 @@ from shutil import rmtree
 import os
 import stat
 
-from cloudinit.util import mergedict, get_cfg_option_list_or_str, write_file
+from cloudinit.util import mergedict, get_cfg_option_list_or_str, write_file, delete_dir_contents
 
 class TestMergeDict(TestCase):
     def test_simple_merge(self):
@@ -173,3 +173,73 @@ class TestWriteFile(MockerTestCase):
         self.mocker.replay()
 
         write_file(path, contents)
+
+class TestDeleteDirContents(TestCase):
+    def setUp(self):
+        super(TestDeleteDirContents, self).setUp()
+        # Make a temp directoy for tests to use.
+        self.tmp = mkdtemp(prefix="unittest_")
+
+    def tearDown(self):
+        super(TestDeleteDirContents, self).tearDown()
+        # Clean up temp directory
+        rmtree(self.tmp)
+
+    def assertDirEmpty(self, dirname):
+        self.assertEqual([], os.listdir(dirname))
+
+    def test_does_not_delete_dir(self):
+        """Ensure directory itself is not deleted."""
+        delete_dir_contents(self.tmp)
+
+        self.assertTrue(os.path.isdir(self.tmp))
+        self.assertDirEmpty(self.tmp)
+
+    def test_deletes_files(self):
+        """Single file should be deleted."""
+        with open(os.path.join(self.tmp, "new_file.txt"), "wb") as f:
+            f.write("DELETE ME")
+
+        delete_dir_contents(self.tmp)
+
+        self.assertDirEmpty(self.tmp)
+
+    def test_deletes_empty_dirs(self):
+        """Empty directories should be deleted."""
+        os.mkdir(os.path.join(self.tmp, "new_dir"))
+
+        delete_dir_contents(self.tmp)
+
+        self.assertDirEmpty(self.tmp)
+
+    def test_deletes_nested_dirs(self):
+        """Nested directories should be deleted."""
+        os.mkdir(os.path.join(self.tmp, "new_dir"))
+        os.mkdir(os.path.join(self.tmp, "new_dir", "new_subdir"))
+
+        delete_dir_contents(self.tmp)
+
+        self.assertDirEmpty(self.tmp)
+
+    def test_deletes_non_empty_dirs(self):
+        """Non-empty directories should be deleted."""
+        os.mkdir(os.path.join(self.tmp, "new_dir"))
+        f_name = os.path.join(self.tmp, "new_dir", "new_file.txt")
+        with open(f_name, "wb") as f:
+            f.write("DELETE ME")
+
+        delete_dir_contents(self.tmp)
+
+        self.assertDirEmpty(self.tmp)
+
+    def test_deletes_symlinks(self):
+        """Symlinks should be deleted."""
+        file_name = os.path.join(self.tmp, "new_file.txt")
+        link_name = os.path.join(self.tmp, "new_file_link.txt")
+        with open(file_name, "wb") as f:
+            f.write("DELETE ME")
+        os.symlink(file_name, link_name)
+
+        delete_dir_contents(self.tmp)
+
+        self.assertDirEmpty(self.tmp)