T2117: Cloud-init updated to 22.1

Merged with 22.1 tag from the upstream Cloud-init repository.

Our modules were slightly modified for compatibility with the new
version.

1 files changed, 198 insertions, 0 deletions

diff --git a/tests/unittests/config/test_cc_rsyslog.py b/tests/unittests/config/test_cc_rsyslog.py
new file mode 100644
index 00000000..e5d06ca2
--- /dev/null
+++ b/tests/unittests/config/test_cc_rsyslog.py
@@ -0,0 +1,198 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+
+import os
+import shutil
+import tempfile
+
+from cloudinit import util
+from cloudinit.config.cc_rsyslog import (
+    DEF_DIR,
+    DEF_FILENAME,
+    DEF_RELOAD,
+    apply_rsyslog_changes,
+    load_config,
+    parse_remotes_line,
+    remotes_to_rsyslog_cfg,
+)
+from tests.unittests import helpers as t_help
+
+
+class TestLoadConfig(t_help.TestCase):
+    def setUp(self):
+        super(TestLoadConfig, self).setUp()
+        self.basecfg = {
+            "config_filename": DEF_FILENAME,
+            "config_dir": DEF_DIR,
+            "service_reload_command": DEF_RELOAD,
+            "configs": [],
+            "remotes": {},
+        }
+
+    def test_legacy_full(self):
+        found = load_config(
+            {
+                "rsyslog": ["*.* @192.168.1.1"],
+                "rsyslog_dir": "mydir",
+                "rsyslog_filename": "myfilename",
+            }
+        )
+        self.basecfg.update(
+            {
+                "configs": ["*.* @192.168.1.1"],
+                "config_dir": "mydir",
+                "config_filename": "myfilename",
+                "service_reload_command": "auto",
+            }
+        )
+
+        self.assertEqual(found, self.basecfg)
+
+    def test_legacy_defaults(self):
+        found = load_config({"rsyslog": ["*.* @192.168.1.1"]})
+        self.basecfg.update({"configs": ["*.* @192.168.1.1"]})
+        self.assertEqual(found, self.basecfg)
+
+    def test_new_defaults(self):
+        self.assertEqual(load_config({}), self.basecfg)
+
+    def test_new_configs(self):
+        cfgs = ["*.* myhost", "*.* my2host"]
+        self.basecfg.update({"configs": cfgs})
+        self.assertEqual(
+            load_config({"rsyslog": {"configs": cfgs}}), self.basecfg
+        )
+
+
+class TestApplyChanges(t_help.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.mkdtemp()
+        self.addCleanup(shutil.rmtree, self.tmp)
+
+    def test_simple(self):
+        cfgline = "*.* foohost"
+        changed = apply_rsyslog_changes(
+            configs=[cfgline], def_fname="foo.cfg", cfg_dir=self.tmp
+        )
+
+        fname = os.path.join(self.tmp, "foo.cfg")
+        self.assertEqual([fname], changed)
+        self.assertEqual(util.load_file(fname), cfgline + "\n")
+
+    def test_multiple_files(self):
+        configs = [
+            "*.* foohost",
+            {"content": "abc", "filename": "my.cfg"},
+            {
+                "content": "filefoo-content",
+                "filename": os.path.join(self.tmp, "mydir/mycfg"),
+            },
+        ]
+
+        changed = apply_rsyslog_changes(
+            configs=configs, def_fname="default.cfg", cfg_dir=self.tmp
+        )
+
+        expected = [
+            (os.path.join(self.tmp, "default.cfg"), "*.* foohost\n"),
+            (os.path.join(self.tmp, "my.cfg"), "abc\n"),
+            (os.path.join(self.tmp, "mydir/mycfg"), "filefoo-content\n"),
+        ]
+        self.assertEqual([f[0] for f in expected], changed)
+        actual = []
+        for fname, _content in expected:
+            util.load_file(fname)
+            actual.append(
+                (
+                    fname,
+                    util.load_file(fname),
+                )
+            )
+        self.assertEqual(expected, actual)
+
+    def test_repeat_def(self):
+        configs = ["*.* foohost", "*.warn otherhost"]
+
+        changed = apply_rsyslog_changes(
+            configs=configs, def_fname="default.cfg", cfg_dir=self.tmp
+        )
+
+        fname = os.path.join(self.tmp, "default.cfg")
+        self.assertEqual([fname], changed)
+
+        expected_content = "\n".join([c for c in configs]) + "\n"
+        found_content = util.load_file(fname)
+        self.assertEqual(expected_content, found_content)
+
+    def test_multiline_content(self):
+        configs = ["line1", "line2\nline3\n"]
+
+        apply_rsyslog_changes(
+            configs=configs, def_fname="default.cfg", cfg_dir=self.tmp
+        )
+
+        fname = os.path.join(self.tmp, "default.cfg")
+        expected_content = "\n".join([c for c in configs])
+        found_content = util.load_file(fname)
+        self.assertEqual(expected_content, found_content)
+
+
+class TestParseRemotesLine(t_help.TestCase):
+    def test_valid_port(self):
+        r = parse_remotes_line("foo:9")
+        self.assertEqual(9, r.port)
+
+    def test_invalid_port(self):
+        with self.assertRaises(ValueError):
+            parse_remotes_line("*.* foo:abc")
+
+    def test_valid_ipv6(self):
+        r = parse_remotes_line("*.* [::1]")
+        self.assertEqual("*.* @[::1]", str(r))
+
+    def test_valid_ipv6_with_port(self):
+        r = parse_remotes_line("*.* [::1]:100")
+        self.assertEqual(r.port, 100)
+        self.assertEqual(r.addr, "::1")
+        self.assertEqual("*.* @[::1]:100", str(r))
+
+    def test_invalid_multiple_colon(self):
+        with self.assertRaises(ValueError):
+            parse_remotes_line("*.* ::1:100")
+
+    def test_name_in_string(self):
+        r = parse_remotes_line("syslog.host", name="foobar")
+        self.assertEqual("*.* @syslog.host # foobar", str(r))
+
+
+class TestRemotesToSyslog(t_help.TestCase):
+    def test_simple(self):
+        # str rendered line must appear in remotes_to_ryslog_cfg return
+        mycfg = "*.* myhost"
+        myline = str(parse_remotes_line(mycfg, name="myname"))
+        r = remotes_to_rsyslog_cfg({"myname": mycfg})
+        lines = r.splitlines()
+        self.assertEqual(1, len(lines))
+        self.assertTrue(myline in r.splitlines())
+
+    def test_header_footer(self):
+        header = "#foo head"
+        footer = "#foo foot"
+        r = remotes_to_rsyslog_cfg(
+            {"myname": "*.* myhost"}, header=header, footer=footer
+        )
+        lines = r.splitlines()
+        self.assertTrue(header, lines[0])
+        self.assertTrue(footer, lines[-1])
+
+    def test_with_empty_or_null(self):
+        mycfg = "*.* myhost"
+        myline = str(parse_remotes_line(mycfg, name="myname"))
+        r = remotes_to_rsyslog_cfg(
+            {"myname": mycfg, "removed": None, "removed2": ""}
+        )
+        lines = r.splitlines()
+        self.assertEqual(1, len(lines))
+        self.assertTrue(myline in r.splitlines())
+
+
+# vi: ts=4 expandtab