diff options
Diffstat (limited to 'tests/unittests/config/test_cc_power_state_change.py')
| -rw-r--r-- | tests/unittests/config/test_cc_power_state_change.py | 159 |
1 files changed, 159 insertions, 0 deletions
diff --git a/tests/unittests/config/test_cc_power_state_change.py b/tests/unittests/config/test_cc_power_state_change.py new file mode 100644 index 00000000..47eb0d58 --- /dev/null +++ b/tests/unittests/config/test_cc_power_state_change.py @@ -0,0 +1,159 @@ +# This file is part of cloud-init. See LICENSE file for license information. + +import sys + +from cloudinit import distros, helpers +from cloudinit.config import cc_power_state_change as psc +from tests.unittests import helpers as t_help +from tests.unittests.helpers import mock + + +class TestLoadPowerState(t_help.TestCase): + def setUp(self): + super(TestLoadPowerState, self).setUp() + cls = distros.fetch("ubuntu") + paths = helpers.Paths({}) + self.dist = cls("ubuntu", {}, paths) + + def test_no_config(self): + # completely empty config should mean do nothing + (cmd, _timeout, _condition) = psc.load_power_state({}, self.dist) + self.assertIsNone(cmd) + + def test_irrelevant_config(self): + # no power_state field in config should return None for cmd + (cmd, _timeout, _condition) = psc.load_power_state( + {"foo": "bar"}, self.dist + ) + self.assertIsNone(cmd) + + def test_invalid_mode(self): + + cfg = {"power_state": {"mode": "gibberish"}} + self.assertRaises(TypeError, psc.load_power_state, cfg, self.dist) + + cfg = {"power_state": {"mode": ""}} + self.assertRaises(TypeError, psc.load_power_state, cfg, self.dist) + + def test_empty_mode(self): + cfg = {"power_state": {"message": "goodbye"}} + self.assertRaises(TypeError, psc.load_power_state, cfg, self.dist) + + def test_valid_modes(self): + cfg = {"power_state": {}} + for mode in ("halt", "poweroff", "reboot"): + cfg["power_state"]["mode"] = mode + check_lps_ret(psc.load_power_state(cfg, self.dist), mode=mode) + + def test_invalid_delay(self): + cfg = {"power_state": {"mode": "poweroff", "delay": "goodbye"}} + self.assertRaises(TypeError, psc.load_power_state, cfg, self.dist) + + def test_valid_delay(self): + cfg = {"power_state": {"mode": "poweroff", "delay": ""}} + for delay in ("now", "+1", "+30"): + cfg["power_state"]["delay"] = delay + check_lps_ret(psc.load_power_state(cfg, self.dist)) + + def test_message_present(self): + cfg = {"power_state": {"mode": "poweroff", "message": "GOODBYE"}} + ret = psc.load_power_state(cfg, self.dist) + check_lps_ret(psc.load_power_state(cfg, self.dist)) + self.assertIn(cfg["power_state"]["message"], ret[0]) + + def test_no_message(self): + # if message is not present, then no argument should be passed for it + cfg = {"power_state": {"mode": "poweroff"}} + (cmd, _timeout, _condition) = psc.load_power_state(cfg, self.dist) + self.assertNotIn("", cmd) + check_lps_ret(psc.load_power_state(cfg, self.dist)) + self.assertTrue(len(cmd) == 3) + + def test_condition_null_raises(self): + cfg = {"power_state": {"mode": "poweroff", "condition": None}} + self.assertRaises(TypeError, psc.load_power_state, cfg, self.dist) + + def test_condition_default_is_true(self): + cfg = {"power_state": {"mode": "poweroff"}} + _cmd, _timeout, cond = psc.load_power_state(cfg, self.dist) + self.assertEqual(cond, True) + + def test_freebsd_poweroff_uses_lowercase_p(self): + cls = distros.fetch("freebsd") + paths = helpers.Paths({}) + freebsd = cls("freebsd", {}, paths) + cfg = {"power_state": {"mode": "poweroff"}} + ret = psc.load_power_state(cfg, freebsd) + self.assertIn("-p", ret[0]) + + def test_alpine_delay(self): + # alpine takes delay in seconds. + cls = distros.fetch("alpine") + paths = helpers.Paths({}) + alpine = cls("alpine", {}, paths) + cfg = {"power_state": {"mode": "poweroff", "delay": ""}} + for delay, value in (("now", 0), ("+1", 60), ("+30", 1800)): + cfg["power_state"]["delay"] = delay + ret = psc.load_power_state(cfg, alpine) + self.assertEqual("-d", ret[0][1]) + self.assertEqual(str(value), ret[0][2]) + + +class TestCheckCondition(t_help.TestCase): + def cmd_with_exit(self, rc): + return [sys.executable, "-c", "import sys; sys.exit(%s)" % rc] + + def test_true_is_true(self): + self.assertEqual(psc.check_condition(True), True) + + def test_false_is_false(self): + self.assertEqual(psc.check_condition(False), False) + + def test_cmd_exit_zero_true(self): + self.assertEqual(psc.check_condition(self.cmd_with_exit(0)), True) + + def test_cmd_exit_one_false(self): + self.assertEqual(psc.check_condition(self.cmd_with_exit(1)), False) + + def test_cmd_exit_nonzero_warns(self): + mocklog = mock.Mock() + self.assertEqual( + psc.check_condition(self.cmd_with_exit(2), mocklog), False + ) + self.assertEqual(mocklog.warning.call_count, 1) + + +def check_lps_ret(psc_return, mode=None): + if len(psc_return) != 3: + raise TypeError("length returned = %d" % len(psc_return)) + + errs = [] + cmd = psc_return[0] + timeout = psc_return[1] + condition = psc_return[2] + + if "shutdown" not in psc_return[0][0]: + errs.append("string 'shutdown' not in cmd") + + if condition is None: + errs.append("condition was not returned") + + if mode is not None: + opt = {"halt": "-H", "poweroff": "-P", "reboot": "-r"}[mode] + if opt not in psc_return[0]: + errs.append("opt '%s' not in cmd: %s" % (opt, cmd)) + + if len(cmd) != 3 and len(cmd) != 4: + errs.append("Invalid command length: %s" % len(cmd)) + + try: + float(timeout) + except Exception: + errs.append("timeout failed convert to float") + + if len(errs): + lines = ["Errors in result: %s" % str(psc_return)] + errs + raise Exception("\n".join(lines)) + + +# vi: ts=4 expandtab |