Minor cleanups to atomic_helper and add unit tests.

Change atomic_helper.write_file to have same same signature as write_file.
Add some simple unit tests for atomic_helper.

2 files changed, 68 insertions, 0 deletions

diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py
index de2cf638..1cdc05a1 100644
--- a/tests/unittests/helpers.py
+++ b/tests/unittests/helpers.py
@@ -252,6 +252,20 @@ class HttprettyTestCase(TestCase):
         super(HttprettyTestCase, self).tearDown()
 
 
+class TempDirTestCase(TestCase):
+    # provide a tempdir per class, not per test.
+    def setUp(self):
+        super(TempDirTestCase, self).setUp()
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp)
+
+    def tmp_path(self, path):
+        if path.startswith(os.path.sep):
+            path = "." + path
+
+        return os.path.normpath(os.path.join(self.tmp, path))
+
+
 def populate_dir(path, files):
     if not os.path.exists(path):
         os.makedirs(path)
diff --git a/tests/unittests/test_atomic_helper.py b/tests/unittests/test_atomic_helper.py
new file mode 100644
index 00000000..feb81551
--- /dev/null
+++ b/tests/unittests/test_atomic_helper.py
@@ -0,0 +1,54 @@
+import json
+import os
+import stat
+
+from cloudinit import atomic_helper
+
+from . import helpers
+
+
+class TestAtomicHelper(helpers.TempDirTestCase):
+    def test_basic_usage(self):
+        """write_file takes bytes if no omode."""
+        path = self.tmp_path("test_basic_usage")
+        contents = b"Hey there\n"
+        atomic_helper.write_file(path, contents)
+        self.check_file(path, contents)
+
+    def test_string(self):
+        """write_file can take a string with mode w."""
+        path = self.tmp_path("test_string")
+        contents = "Hey there\n"
+        atomic_helper.write_file(path, contents, omode="w")
+        self.check_file(path, contents, omode="r")
+
+    def test_file_permissions(self):
+        """write_file with mode 400 works correctly."""
+        path = self.tmp_path("test_file_permissions")
+        contents = b"test_file_perms"
+        atomic_helper.write_file(path, contents, mode=0o400)
+        self.check_file(path, contents, perms=0o400)
+
+    def test_write_json(self):
+        """write_json output is readable json."""
+        path = self.tmp_path("test_write_json")
+        data = {'key1': 'value1', 'key2': ['i1', 'i2']}
+        atomic_helper.write_json(path, data)
+        with open(path, "r") as fp:
+            found = json.load(fp)
+        self.assertEqual(data, found)
+        self.check_perms(path, 0o644)
+
+    def check_file(self, path, content, omode=None, perms=0o644):
+        if omode is None:
+            omode = "rb"
+        self.assertTrue(os.path.exists(path))
+        self.assertTrue(os.path.isfile(path))
+        with open(path, omode) as fp:
+            found = fp.read()
+            self.assertEqual(content, found)
+        self.check_perms(path, perms)
+
+    def check_perms(self, path, perms):
+        file_stat = os.stat(path)
+        self.assertEqual(perms, stat.S_IMODE(file_stat.st_mode))