cephadm: add unit tests for write_new

Signed-off-by: John Mulligan <jmulligan@redhat.com>

diff --git a/src/cephadm/tests/test_util_funcs.py b/src/cephadm/tests/test_util_funcs.py
index 2c20dfb916b32655ec239e25789bf49b184eadd9..f29c0819fb570d0793174925a35d563feb3df3d6 100644 (file)
--- a/src/cephadm/tests/test_util_funcs.py
+++ b/src/cephadm/tests/test_util_funcs.py
@@ -663,3 +663,66 @@ def test_call(caplog, monkeypatch, pyline, expected, call_kwargs, log_check):
         assert result == expected
     if callable(log_check):
         log_check(caplog)
+
+
+class TestWriteNew:
+    def test_success(self, tmp_path):
+        "Test the simple basic feature of writing a file."
+        dest = tmp_path / "foo.txt"
+        with _cephadm.write_new(dest) as fh:
+            fh.write("something\n")
+            fh.write("something else\n")
+
+        with open(dest, "r") as fh:
+            assert fh.read() == "something\nsomething else\n"
+
+    def test_write_ower_mode(self, tmp_path):
+        "Test that the owner and perms options function."
+        dest = tmp_path / "foo.txt"
+
+        # if this is test run as non-root, we can't really change ownership
+        uid = os.getuid()
+        gid = os.getgid()
+
+        with _cephadm.write_new(dest, owner=(uid, gid), perms=0o600) as fh:
+            fh.write("xomething\n")
+            fh.write("xomething else\n")
+
+        with open(dest, "r") as fh:
+            assert fh.read() == "xomething\nxomething else\n"
+            sr = os.fstat(fh.fileno())
+            assert sr.st_uid == uid
+            assert sr.st_gid == gid
+            assert (sr.st_mode & 0o777) == 0o600
+
+    def test_encoding(self, tmp_path):
+        "Test that the encoding option functions."
+        dest = tmp_path / "foo.txt"
+        msg = "\u2603\u26C5\n"
+        with _cephadm.write_new(dest, encoding='utf-8') as fh:
+            fh.write(msg)
+        with open(dest, "rb") as fh:
+            b1 = fh.read()
+            assert b1.decode('utf-8') == msg
+
+        dest = tmp_path / "foo2.txt"
+        with _cephadm.write_new(dest, encoding='utf-16le') as fh:
+            fh.write(msg)
+        with open(dest, "rb") as fh:
+            b2 = fh.read()
+            assert b2.decode('utf-16le') == msg
+
+        # the binary data should differ due to the different encodings
+        assert b1 != b2
+
+    def test_cleanup(self, tmp_path):
+        "Test that an exception during write leaves no file behind."
+        dest = tmp_path / "foo.txt"
+        with pytest.raises(ValueError):
+            with _cephadm.write_new(dest) as fh:
+                fh.write("hello\n")
+                raise ValueError("foo")
+                fh.write("world\n")
+        assert not dest.exists()
+        assert not dest.with_name(dest.name+".new").exists()
+        assert list(dest.parent.iterdir()) == []