1 files changed, 165 insertions, 0 deletions
diff --git a/tests/unittests/config/test_cc_bootcmd.py b/tests/unittests/config/test_cc_bootcmd.py
new file mode 100644
index 00000000..34b16b85
--- /dev/null
+++ b/tests/unittests/config/test_cc_bootcmd.py
@@ -0,0 +1,165 @@
+# This file is part of cloud-init. See LICENSE file for license information.
+import logging
+import re
+import tempfile
+
+import pytest
+
+from cloudinit import subp, util
+from cloudinit.config.cc_bootcmd import handle
+from cloudinit.config.schema import (
+    SchemaValidationError,
+    get_schema,
+    validate_cloudconfig_schema,
+)
+from tests.unittests.helpers import CiTestCase, mock, skipUnlessJsonSchema
+from tests.unittests.util import get_cloud
+
+LOG = logging.getLogger(__name__)
+
+
+class FakeExtendedTempFile(object):
+    def __init__(self, suffix):
+        self.suffix = suffix
+        self.handle = tempfile.NamedTemporaryFile(
+            prefix="ci-%s." % self.__class__.__name__, delete=False
+        )
+
+    def __enter__(self):
+        return self.handle
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        self.handle.close()
+        util.del_file(self.handle.name)
+
+
+class TestBootcmd(CiTestCase):
+
+    with_logs = True
+
+    _etmpfile_path = (
+        "cloudinit.config.cc_bootcmd.temp_utils.ExtendedTemporaryFile"
+    )
+
+    def setUp(self):
+        super(TestBootcmd, self).setUp()
+        self.subp = subp.subp
+        self.new_root = self.tmp_dir()
+
+    def test_handler_skip_if_no_bootcmd(self):
+        """When the provided config doesn't contain bootcmd, skip it."""
+        cfg = {}
+        mycloud = get_cloud()
+        handle("notimportant", cfg, mycloud, LOG, None)
+        self.assertIn(
+            "Skipping module named notimportant, no 'bootcmd' key",
+            self.logs.getvalue(),
+        )
+
+    def test_handler_invalid_command_set(self):
+        """Commands which can't be converted to shell will raise errors."""
+        invalid_config = {"bootcmd": 1}
+        cc = get_cloud()
+        with self.assertRaises(TypeError) as context_manager:
+            handle("cc_bootcmd", invalid_config, cc, LOG, [])
+        self.assertIn("Failed to shellify bootcmd", self.logs.getvalue())
+        self.assertEqual(
+            "Input to shellify was type 'int'. Expected list or tuple.",
+            str(context_manager.exception),
+        )
+
+        invalid_config = {
+            "bootcmd": ["ls /", 20, ["wget", "http://stuff/blah"], {"a": "n"}]
+        }
+        cc = get_cloud()
+        with self.assertRaises(TypeError) as context_manager:
+            handle("cc_bootcmd", invalid_config, cc, LOG, [])
+        logs = self.logs.getvalue()
+        self.assertIn("Failed to shellify", logs)
+        self.assertEqual(
+            "Unable to shellify type 'int'. Expected list, string, tuple. "
+            "Got: 20",
+            str(context_manager.exception),
+        )
+
+    def test_handler_creates_and_runs_bootcmd_script_with_instance_id(self):
+        """Valid schema runs a bootcmd script with INSTANCE_ID in the env."""
+        cc = get_cloud()
+        out_file = self.tmp_path("bootcmd.out", self.new_root)
+        my_id = "b6ea0f59-e27d-49c6-9f87-79f19765a425"
+        valid_config = {
+            "bootcmd": ["echo {0} $INSTANCE_ID > {1}".format(my_id, out_file)]
+        }
+
+        with mock.patch(self._etmpfile_path, FakeExtendedTempFile):
+            with self.allow_subp(["/bin/sh"]):
+                handle("cc_bootcmd", valid_config, cc, LOG, [])
+        self.assertEqual(
+            my_id + " iid-datasource-none\n", util.load_file(out_file)
+        )
+
+    def test_handler_runs_bootcmd_script_with_error(self):
+        """When a valid script generates an error, that error is raised."""
+        cc = get_cloud()
+        valid_config = {"bootcmd": ["exit 1"]}  # Script with error
+
+        with mock.patch(self._etmpfile_path, FakeExtendedTempFile):
+            with self.allow_subp(["/bin/sh"]):
+                with self.assertRaises(subp.ProcessExecutionError) as ctxt:
+                    handle("does-not-matter", valid_config, cc, LOG, [])
+        self.assertIn(
+            "Unexpected error while running command.\nCommand: ['/bin/sh',",
+            str(ctxt.exception),
+        )
+        self.assertIn(
+            "Failed to run bootcmd module does-not-matter",
+            self.logs.getvalue(),
+        )
+
+
+@skipUnlessJsonSchema()
+class TestBootCMDSchema:
+    """Directly test schema rather than through handle."""
+
+    @pytest.mark.parametrize(
+        "config, error_msg",
+        (
+            # Valid schemas tested by meta.examples in test_schema
+            # Invalid schemas
+            (
+                {"bootcmd": 1},
+                "Cloud config schema errors: bootcmd: 1 is not of type"
+                " 'array'",
+            ),
+            ({"bootcmd": []}, re.escape("bootcmd: [] is too short")),
+            (
+                {"bootcmd": []},
+                re.escape(
+                    "Cloud config schema errors: bootcmd: [] is too short"
+                ),
+            ),
+            (
+                {
+                    "bootcmd": [
+                        "ls /",
+                        20,
+                        ["wget", "http://stuff/blah"],
+                        {"a": "n"},
+                    ]
+                },
+                "Cloud config schema errors: bootcmd.1: 20 is not valid under"
+                " any of the given schemas, bootcmd.3: {'a': 'n'} is not"
+                " valid under any of the given schemas",
+            ),
+        ),
+    )
+    @skipUnlessJsonSchema()
+    def test_schema_validation(self, config, error_msg):
+        """Assert expected schema validation and error messages."""
+        # New-style schema $defs exist in config/cloud-init-schema*.json
+        schema = get_schema()
+        with pytest.raises(SchemaValidationError, match=error_msg):
+            validate_cloudconfig_schema(config, schema, strict=True)
+
+
+# vi: ts=4 expandtab