stages: don't reset permissions of cloud-init.log every boot (#624)

ensure_file needed modification to support doing this, so this commit
also includes the following changes:

test_util: add tests for util.ensure_file
util: add preserve_mode parameter to ensure_file
util: add (partial) type annotations to ensure_file

LP: #1900837

2 files changed, 107 insertions, 0 deletions

diff --git a/cloudinit/tests/test_stages.py b/cloudinit/tests/test_stages.py
index d5c9c0e4..d2d1b37f 100644
--- a/cloudinit/tests/test_stages.py
+++ b/cloudinit/tests/test_stages.py
@@ -3,6 +3,9 @@
 """Tests related to cloudinit.stages module."""
 
 import os
+import stat
+
+import pytest
 
 from cloudinit import stages
 from cloudinit import sources
@@ -341,4 +344,63 @@ class TestInit(CiTestCase):
         self.init.distro.apply_network_config.assert_called_with(
             net_cfg, bring_up=True)
 
+
+class TestInit_InitializeFilesystem:
+    """Tests for cloudinit.stages.Init._initialize_filesystem.
+
+    TODO: Expand these tests to cover all of _initialize_filesystem's behavior.
+    """
+
+    @pytest.yield_fixture
+    def init(self, paths):
+        """A fixture which yields a stages.Init instance with paths and cfg set
+
+        As it is replaced with a mock, consumers of this fixture can set
+        `init.cfg` if the default empty dict configuration is not appropriate.
+        """
+        with mock.patch(
+            "cloudinit.stages.Init.cfg", mock.PropertyMock(return_value={})
+        ):
+            with mock.patch("cloudinit.stages.util.ensure_dirs"):
+                init = stages.Init()
+                init._paths = paths
+                yield init
+
+    @mock.patch("cloudinit.stages.util.ensure_file")
+    def test_ensure_file_not_called_if_no_log_file_configured(
+        self, m_ensure_file, init
+    ):
+        """If no log file is configured, we should not ensure its existence."""
+        init.cfg = {}
+
+        init._initialize_filesystem()
+
+        assert 0 == m_ensure_file.call_count
+
+    def test_log_files_existence_is_ensured_if_configured(self, init, tmpdir):
+        """If a log file is configured, we should ensure its existence."""
+        log_file = tmpdir.join("cloud-init.log")
+        init.cfg = {"def_log_file": str(log_file)}
+
+        init._initialize_filesystem()
+
+        assert log_file.exists
+
+    def test_existing_file_permissions_are_not_modified(self, init, tmpdir):
+        """If the log file already exists, we should not modify its permissions
+
+        See https://bugs.launchpad.net/cloud-init/+bug/1900837.
+        """
+        # Use a mode that will never be made the default so this test will
+        # always be valid
+        mode = 0o606
+        log_file = tmpdir.join("cloud-init.log")
+        log_file.ensure()
+        log_file.chmod(mode)
+        init.cfg = {"def_log_file": str(log_file)}
+
+        init._initialize_filesystem()
+
+        assert mode == stat.S_IMODE(log_file.stat().mode)
+
 # vi: ts=4 expandtab
diff --git a/cloudinit/tests/test_util.py b/cloudinit/tests/test_util.py
index 096a3037..77714928 100644
--- a/cloudinit/tests/test_util.py
+++ b/cloudinit/tests/test_util.py
@@ -771,4 +771,49 @@ class TestMountCb:
         ] == callback.call_args_list
 
 
+@mock.patch("cloudinit.util.write_file")
+class TestEnsureFile:
+    """Tests for ``cloudinit.util.ensure_file``."""
+
+    def test_parameters_passed_through(self, m_write_file):
+        """Test the parameters in the signature are passed to write_file."""
+        util.ensure_file(
+            mock.sentinel.path,
+            mode=mock.sentinel.mode,
+            preserve_mode=mock.sentinel.preserve_mode,
+        )
+
+        assert 1 == m_write_file.call_count
+        args, kwargs = m_write_file.call_args
+        assert (mock.sentinel.path,) == args
+        assert mock.sentinel.mode == kwargs["mode"]
+        assert mock.sentinel.preserve_mode == kwargs["preserve_mode"]
+
+    @pytest.mark.parametrize(
+        "kwarg,expected",
+        [
+            # Files should be world-readable by default
+            ("mode", 0o644),
+            # The previous behaviour of not preserving mode should be retained
+            ("preserve_mode", False),
+        ],
+    )
+    def test_defaults(self, m_write_file, kwarg, expected):
+        """Test that ensure_file defaults appropriately."""
+        util.ensure_file(mock.sentinel.path)
+
+        assert 1 == m_write_file.call_count
+        _args, kwargs = m_write_file.call_args
+        assert expected == kwargs[kwarg]
+
+    def test_static_parameters_are_passed(self, m_write_file):
+        """Test that the static write_files parameters are passed correctly."""
+        util.ensure_file(mock.sentinel.path)
+
+        assert 1 == m_write_file.call_count
+        _args, kwargs = m_write_file.call_args
+        assert "" == kwargs["content"]
+        assert "ab" == kwargs["omode"]
+
+
 # vi: ts=4 expandtab