From aa8b51a48a30e3a3c863ca0ddb8bc4667026d57a Mon Sep 17 00:00:00 2001 From: harlowja Date: Sat, 27 Oct 2012 19:25:48 -0700 Subject: Helpful cleanups. 1. Remove the usage of the path.join function now that all code should be going through the util file methods (and they can be mocked out as needed). 2. Adjust all occurences of the above join function to either not use it or replace it with the standard os.path.join (which can also be mocked out as needed) 3. Fix pylint from complaining about the tests folder 'helpers.py' not being found 4. Add a pylintrc file that is used instead of the options hidden in the 'run_pylint' tool. --- tests/unittests/test_handler/__init__.py | 0 tests/unittests/test_handler/test_handler_ca_certs.py | 18 +++++++++--------- 2 files changed, 9 insertions(+), 9 deletions(-) create mode 100644 tests/unittests/test_handler/__init__.py (limited to 'tests/unittests/test_handler') diff --git a/tests/unittests/test_handler/__init__.py b/tests/unittests/test_handler/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/tests/unittests/test_handler/test_handler_ca_certs.py b/tests/unittests/test_handler/test_handler_ca_certs.py index d3df5c50..d73c9fa9 100644 --- a/tests/unittests/test_handler/test_handler_ca_certs.py +++ b/tests/unittests/test_handler/test_handler_ca_certs.py @@ -77,7 +77,7 @@ class TestConfig(MockerTestCase): """Test that a single cert gets passed to add_ca_certs.""" config = {"ca-certs": {"trusted": ["CERT1"]}} - self.mock_add(self.paths, ["CERT1"]) + self.mock_add(["CERT1"]) self.mock_update() self.mocker.replay() @@ -87,7 +87,7 @@ class TestConfig(MockerTestCase): """Test that multiple certs get passed to add_ca_certs.""" config = {"ca-certs": {"trusted": ["CERT1", "CERT2"]}} - self.mock_add(self.paths, ["CERT1", "CERT2"]) + self.mock_add(["CERT1", "CERT2"]) self.mock_update() self.mocker.replay() @@ -97,7 +97,7 @@ class TestConfig(MockerTestCase): """Test remove_defaults works as expected.""" config = {"ca-certs": {"remove-defaults": True}} - self.mock_remove(self.paths) + self.mock_remove() self.mock_update() self.mocker.replay() @@ -116,8 +116,8 @@ class TestConfig(MockerTestCase): """Test remove_defaults is not called when config value is False.""" config = {"ca-certs": {"remove-defaults": True, "trusted": ["CERT1"]}} - self.mock_remove(self.paths) - self.mock_add(self.paths, ["CERT1"]) + self.mock_remove() + self.mock_add(["CERT1"]) self.mock_update() self.mocker.replay() @@ -136,7 +136,7 @@ class TestAddCaCerts(MockerTestCase): """Test that no certificate are written if not provided.""" self.mocker.replace(util.write_file, passthrough=False) self.mocker.replay() - cc_ca_certs.add_ca_certs(self.paths, []) + cc_ca_certs.add_ca_certs([]) def test_single_cert(self): """Test adding a single certificate to the trusted CAs.""" @@ -149,7 +149,7 @@ class TestAddCaCerts(MockerTestCase): "\ncloud-init-ca-certs.crt", omode="ab") self.mocker.replay() - cc_ca_certs.add_ca_certs(self.paths, [cert]) + cc_ca_certs.add_ca_certs([cert]) def test_multiple_certs(self): """Test adding multiple certificates to the trusted CAs.""" @@ -163,7 +163,7 @@ class TestAddCaCerts(MockerTestCase): "\ncloud-init-ca-certs.crt", omode="ab") self.mocker.replay() - cc_ca_certs.add_ca_certs(self.paths, certs) + cc_ca_certs.add_ca_certs(certs) class TestUpdateCaCerts(MockerTestCase): @@ -198,4 +198,4 @@ class TestRemoveDefaultCaCerts(MockerTestCase): "ca-certificates ca-certificates/trust_new_crts select no") self.mocker.replay() - cc_ca_certs.remove_default_ca_certs(self.paths) + cc_ca_certs.remove_default_ca_certs() -- cgit v1.2.3 From bd01f3466e10ca515a8e8aec42d00201f40cbd53 Mon Sep 17 00:00:00 2001 From: Joshua Harlow Date: Thu, 8 Nov 2012 17:41:36 -0800 Subject: Forgot the test! --- .../test_handler/test_handler_set_hostname.py | 58 ++++++++++++++++++++++ 1 file changed, 58 insertions(+) create mode 100644 tests/unittests/test_handler/test_handler_set_hostname.py (limited to 'tests/unittests/test_handler') diff --git a/tests/unittests/test_handler/test_handler_set_hostname.py b/tests/unittests/test_handler/test_handler_set_hostname.py new file mode 100644 index 00000000..a1aba62f --- /dev/null +++ b/tests/unittests/test_handler/test_handler_set_hostname.py @@ -0,0 +1,58 @@ +from cloudinit.config import cc_set_hostname + +from cloudinit import cloud +from cloudinit import distros +from cloudinit import helpers +from cloudinit import util + +from tests.unittests import helpers as t_help + +import logging + +from StringIO import StringIO + +from configobj import ConfigObj + +LOG = logging.getLogger(__name__) + + +class TestHostname(t_help.FilesystemMockingTestCase): + def setUp(self): + super(TestHostname, self).setUp() + self.tmp = self.makeDir(prefix="unittest_") + + def _fetch_distro(self, kind): + cls = distros.fetch(kind) + paths = helpers.Paths({}) + return cls(kind, {}, paths) + + def test_write_hostname_rhel(self): + cfg = { + 'hostname': 'blah.blah.blah.yahoo.com', + } + distro = self._fetch_distro('rhel') + paths = helpers.Paths({}) + ds = None + cc = cloud.Cloud(ds, paths, {}, distro, None) + self.patchUtils(self.tmp) + self.patchOS(self.tmp) + cc_set_hostname.handle('cc_set_hostname', + cfg, cc, LOG, []) + contents = util.load_file("/etc/sysconfig/network") + n_cfg = ConfigObj(StringIO(contents)) + self.assertEquals({'HOSTNAME': 'blah.blah.blah.yahoo.com'}, + dict(n_cfg)) + + def test_write_hostname_debian(self): + cfg = { + 'hostname': 'blah.blah.blah.yahoo.com', + } + distro = self._fetch_distro('debian') + paths = helpers.Paths({}) + ds = None + cc = cloud.Cloud(ds, paths, {}, distro, None) + self.patchUtils(self.tmp) + cc_set_hostname.handle('cc_set_hostname', + cfg, cc, LOG, []) + contents = util.load_file("/etc/hostname") + self.assertEquals('blah', contents.strip()) -- cgit v1.2.3 From 4944bc286970da1ca59b7d57509e51b6817c60c2 Mon Sep 17 00:00:00 2001 From: Scott Moser Date: Mon, 12 Nov 2012 14:29:14 -0500 Subject: test_handler_yum_add_repo: fix broken test --- tests/unittests/test_handler/test_handler_yum_add_repo.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) (limited to 'tests/unittests/test_handler') diff --git a/tests/unittests/test_handler/test_handler_yum_add_repo.py b/tests/unittests/test_handler/test_handler_yum_add_repo.py index c7fcddc4..8df592f9 100644 --- a/tests/unittests/test_handler/test_handler_yum_add_repo.py +++ b/tests/unittests/test_handler/test_handler_yum_add_repo.py @@ -35,8 +35,8 @@ class TestConfig(helpers.FilesystemMockingTestCase): } self.patchUtils(self.tmp) cc_yum_add_repo.handle('yum_add_repo', cfg, None, LOG, []) - with self.assertRaises(IOError): - util.load_file("/etc/yum.repos.d/epel_testing.repo") + self.assertRaises(IOError, util.load_file, + "/etc/yum.repos.d/epel_testing.repo") def test_write_config(self): cfg = { -- cgit v1.2.3 From 2113e89b6816d2c9d442103698414cd189ca3412 Mon Sep 17 00:00:00 2001 From: Scott Moser Date: Tue, 13 Nov 2012 11:18:22 -0500 Subject: implement power_state with tests. --- ChangeLog | 4 +- cloudinit/config/cc_power_state_change.py | 122 +++++++++++---------- doc/examples/cloud-config.txt | 39 ++++--- .../test_handler/test_handler_power_state.py | 88 +++++++++++++++ 4 files changed, 175 insertions(+), 78 deletions(-) create mode 100644 tests/unittests/test_handler/test_handler_power_state.py (limited to 'tests/unittests/test_handler') diff --git a/ChangeLog b/ChangeLog index 57bfc1a4..23612228 100644 --- a/ChangeLog +++ b/ChangeLog @@ -43,8 +43,8 @@ - Added dependency on distribute's python-pkg-resources - use a set of helper/parsing classes to perform system configuration for easier test. (/etc/sysconfig, /etc/hostname, resolv.conf, /etc/hosts) - - add 'finalcmd' config module to execute 'finalcmd' entries like - 'runcmd' but detached from cloud-init (LP: #1064665) + - add power_state_change config module for shutting down stystem after + cloud-init finishes. (LP: #1064665) 0.7.0: - add a 'exception_cb' argument to 'wait_for_url'. If provided, this method will be called back with the exception received and the message. diff --git a/cloudinit/config/cc_power_state_change.py b/cloudinit/config/cc_power_state_change.py index 67e0316b..07de548c 100644 --- a/cloudinit/config/cc_power_state_change.py +++ b/cloudinit/config/cc_power_state_change.py @@ -21,6 +21,7 @@ from cloudinit import util import errno import os +import re import subprocess import sys import time @@ -28,83 +29,91 @@ import time frequency = PER_INSTANCE -def handle(_name, cfg, cloud, log, _args): +def handle(_name, cfg, _cloud, log, _args): - finalcmds = cfg.get("finalcmd") - - if not finalcmds: - log.debug("No final commands") + try: + (args, timeout) = load_power_state(cfg) + if args is None: + log.debug("no power_state provided. doing nothing") + return + except Exception as e: + log.warn("%s Not performing power state change!" % str(e)) return mypid = os.getpid() cmdline = util.load_file("/proc/%s/cmdline") if not cmdline: - log.warn("Failed to get cmdline of current process") - return - - try: - timeout = float(cfg.get("finalcmd_timeout", 30.0)) - except ValueError: - log.warn("failed to convert finalcmd_timeout '%s' to float" % - cfg.get("finalcmd_timeout", 30.0)) + log.warn("power_state: failed to get cmdline of current process") return devnull_fp = open("/dev/null", "w") - shellcode = util.shellify(finalcmds) + log.debug("After pid %s ends, will execute: %s" % (mypid, ' '.join(args))) + + util.fork_cb(run_after_pid_gone, mypid, cmdline, timeout, log, execmd, + [args, devnull_fp]) + + +def load_power_state(cfg): + # returns a tuple of shutdown_command, timeout + # shutdown_command is None if no config found + pstate = cfg.get('power_state') + + if pstate is None: + return (None, None) + + if not isinstance(pstate, dict): + raise TypeError("power_state is not a dict.") - # note, after the fork, we do not use any of cloud-init's functions - # that would attempt to log. The primary reason for that is - # to allow the 'finalcmd' the ability to do just about anything - # and not depend on syslog services. - # Basically, it should "just work" to have finalcmd of: - # - sleep 30 - # - /sbin/poweroff - finalcmd_d = os.path.join(cloud.get_ipath_cur(), "finalcmds") + opt_map = {'halt': '-H', 'poweroff': '-P', 'reboot': '-r'} - util.fork_cb(run_after_pid_gone, mypid, cmdline, timeout, - runfinal, (shellcode, finalcmd_d, devnull_fp)) + mode = pstate.get("mode") + if mode not in opt_map: + raise TypeError("power_state[mode] required, must be one of: %s." % + ','.join(opt_map.keys())) + delay = pstate.get("delay", "now") + if delay != "now" and not re.match("\+[0-9]+", delay): + raise TypeError("power_state[delay] must be 'now' or '+m' (minutes).") -def execmd(exe_args, data_in=None, output=None): + args = ["shutdown", opt_map[mode], delay] + if pstate.get("message"): + args.append(pstate.get("message")) + + try: + timeout = float(pstate.get('timeout', 30.0)) + except ValueError: + raise ValueError("failed to convert timeout '%s' to float." % + pstate['timeout']) + + return (args, timeout) + + +def execmd(exe_args, output=None, data_in=None): try: proc = subprocess.Popen(exe_args, stdin=subprocess.PIPE, stdout=output, stderr=subprocess.STDOUT) proc.communicate(data_in) except Exception: - return 254 - return proc.returncode() + sys.exit(254) + sys.exit(proc.returncode()) -def runfinal(shellcode, finalcmd_d, output=None): - ret = execmd(["/bin/sh"], data_in=shellcode, output=output) - if not (finalcmd_d and os.path.isdir(finalcmd_d)): - sys.exit(ret) - - fails = 0 - if ret != 0: - fails = 1 - - # now runparts the final command dir - for exe_name in sorted(os.listdir(finalcmd_d)): - exe_path = os.path.join(finalcmd_d, exe_name) - if os.path.isfile(exe_path) and os.access(exe_path, os.X_OK): - ret = execmd([exe_path], data_in=None, output=output) - if ret != 0: - fails += 1 - sys.exit(fails) - - -def run_after_pid_gone(pid, pidcmdline, timeout, func, args): +def run_after_pid_gone(pid, pidcmdline, timeout, log, func, args): # wait until pid, with /proc/pid/cmdline contents of pidcmdline # is no longer alive. After it is gone, or timeout has passed # execute func(args) - msg = "ERROR: Uncaught error" + msg = None end_time = time.time() + timeout cmdline_f = "/proc/%s/cmdline" % pid + def fatal(msg): + if log: + log.warn(msg) + sys.exit(254) + while True: if time.time() > end_time: msg = "timeout reached before %s ended" % pid @@ -122,18 +131,15 @@ def run_after_pid_gone(pid, pidcmdline, timeout, func, args): if ioerr.errno == errno.ENOENT: msg = "pidfile '%s' gone" % cmdline_f else: - msg = "ERROR: IOError: %s" % ioerr - raise + fatal("IOError during wait: %s" % ioerr) break except Exception as e: - msg = "ERROR: Exception: %s" % e - raise + fatal("Unexpected Exception: %s" % e) - if msg.startswith("ERROR:"): - sys.stderr.write(msg) - sys.stderr.write("Not executing finalcmd") - sys.exit(1) + if not msg: + fatal("Unexpected error in run_after_pid_gone") - sys.stderr.write("calling %s with %s\n" % (func, args)) - sys.exit(func(*args)) + if log: + log.debug(msg) + func(*args) diff --git a/doc/examples/cloud-config.txt b/doc/examples/cloud-config.txt index 4fc5f351..09298655 100644 --- a/doc/examples/cloud-config.txt +++ b/doc/examples/cloud-config.txt @@ -256,24 +256,6 @@ bootcmd: - echo 192.168.1.130 us.archive.ubuntu.com > /etc/hosts - [ cloud-init-per, once, mymkfs, mkfs, /dev/vdb ] -# final commands -# default: none -# This can be used to execute commands after and fully detached from -# a cloud-init stage. The initial purpose of it was to allow 'poweroff' -# detached from cloud-init. If poweroff was run from 'runcmd' or userdata -# then messages may be spewed from cloud-init about logging failing or other -# issues as a result of the system being turned off. -# -# You probably are better off using 'runcmd' for this. -# -# The output of finalcmd will redirected redirected to /dev/null -# If you want output to be seen, take care to do so in your commands -# themselves. See example. -finalcmd: - - sleep 30 - - "echo $(date -R): powering off > /dev/console" - - /sbin/poweroff - # cloud_config_modules: # default: # cloud_config_modules: @@ -596,3 +578,24 @@ manual_cache_clean: False # A list of key types (first token of a /etc/ssh/ssh_key_*.pub file) # that should be skipped when outputting key fingerprints and keys # to the console respectively. + +## poweroff or reboot system after finished +# default: none +# +# power_state can be used to make the system shutdown, reboot or +# halt after boot is finished. This same thing can be acheived by +# user-data scripts or by runcmd by simply invoking 'shutdown'. +# +# Doing it this way ensures that cloud-init is entirely finished with +# modules that would be executed, and avoids any error/log messages +# that may go to the console as a result of system services like +# syslog being taken down while cloud-init is running. +# +# delay: form accepted by shutdown. default is 'now'. other format +# accepted is +m (m in minutes) +# mode: required. must be one of 'poweroff', 'halt', 'reboot' +# message: provided as the message argument to 'shutdown'. default is none. +power_state: + delay: 30 + mode: poweroff + message: Bye Bye diff --git a/tests/unittests/test_handler/test_handler_power_state.py b/tests/unittests/test_handler/test_handler_power_state.py new file mode 100644 index 00000000..1149fedc --- /dev/null +++ b/tests/unittests/test_handler/test_handler_power_state.py @@ -0,0 +1,88 @@ +from unittest import TestCase + +from cloudinit.config import cc_power_state_change as psc + + +class TestLoadPowerState(TestCase): + def setUp(self): + super(self.__class__, self).setUp() + + def test_no_config(self): + # completely empty config should mean do nothing + (cmd, _timeout) = psc.load_power_state({}) + self.assertEqual(cmd, None) + + def test_irrelevant_config(self): + # no power_state field in config should return None for cmd + (cmd, _timeout) = psc.load_power_state({'foo': 'bar'}) + self.assertEqual(cmd, None) + + def test_invalid_mode(self): + cfg = {'power_state': {'mode': 'gibberish'}} + self.assertRaises(TypeError, psc.load_power_state, cfg) + + cfg = {'power_state': {'mode': ''}} + self.assertRaises(TypeError, psc.load_power_state, cfg) + + def test_empty_mode(self): + cfg = {'power_state': {'message': 'goodbye'}} + self.assertRaises(TypeError, psc.load_power_state, cfg) + + 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), mode=mode) + + def test_invalid_delay(self): + cfg = {'power_state': {'mode': 'poweroff', 'delay': 'goodbye'}} + self.assertRaises(TypeError, psc.load_power_state, cfg) + + 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)) + + def test_message_present(self): + cfg = {'power_state': {'mode': 'poweroff', 'message': 'GOODBYE'}} + ret = psc.load_power_state(cfg) + check_lps_ret(psc.load_power_state(cfg)) + 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) = psc.load_power_state(cfg) + self.assertNotIn("", cmd) + check_lps_ret(psc.load_power_state(cfg)) + self.assertTrue(len(cmd) == 3) + + +def check_lps_ret(psc_return, mode=None): + if len(psc_return) != 2: + raise TypeError("length returned = %d" % len(psc_return)) + + errs = [] + cmd = psc_return[0] + timeout = psc_return[1] + + if not 'shutdown' in psc_return[0][0]: + errs.append("string 'shutdown' not in cmd") + + 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: + errs.append("timeout failed convert to float") + + if len(errs): + lines = ["Errors in result: %s" % str(psc_return)] + errs + raise Exception('\n'.join(lines)) -- cgit v1.2.3 From 5c63171d2ac45a1931130d14e33579beb6902fdb Mon Sep 17 00:00:00 2001 From: Joshua Harlow Date: Tue, 13 Nov 2012 15:49:46 -0800 Subject: Create a utility testcase class that fixes some of the 2.6 missing pieces - Add a helper testcase class that can add additional features into the unit test class as we need for features that are useful to have which starts with features that are missing including assertIn and assertNotIn LP: #1078473 --- tests/unittests/helpers.py | 23 ++++++++++++++++++++++ .../test_handler/test_handler_power_state.py | 6 +++--- 2 files changed, 26 insertions(+), 3 deletions(-) (limited to 'tests/unittests/test_handler') diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index e8080668..92540b0c 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -1,4 +1,6 @@ import os +import sys +import unittest from mocker import MockerTestCase @@ -7,6 +9,27 @@ from cloudinit import util import shutil +# Handle how 2.6 doesn't have the assertIn or assertNotIn +_PY_VER = sys.version_info +_PY_MAJOR, _PY_MINOR = _PY_VER[0:2] +if (_PY_MAJOR, _PY_MINOR) <= (2, 6): + # For now add these on, taken from python 2.7 + slightly adjusted + class TestCase(unittest.TestCase): + def assertIn(self, member, container, msg=None): + if member not in container: + standardMsg = '%r not found in %r' % (member, container) + self.fail(self._formatMessage(msg, standardMsg)) + + def assertNotIn(self, member, container, msg=None): + if member in container: + standardMsg = '%r unexpectedly found in %r' + standardMsg = standardMsg % (member, container) + self.fail(self._formatMessage(msg, standardMsg)) + +else: + class TestCase(unittest.TestCase): + pass + # Makes the old path start # with new base instead of whatever diff --git a/tests/unittests/test_handler/test_handler_power_state.py b/tests/unittests/test_handler/test_handler_power_state.py index 1149fedc..f6e37fa5 100644 --- a/tests/unittests/test_handler/test_handler_power_state.py +++ b/tests/unittests/test_handler/test_handler_power_state.py @@ -1,9 +1,9 @@ -from unittest import TestCase - from cloudinit.config import cc_power_state_change as psc +from tests.unittests import helpers as t_help + -class TestLoadPowerState(TestCase): +class TestLoadPowerState(t_help.TestCase): def setUp(self): super(self.__class__, self).setUp() -- cgit v1.2.3 From 974e76eab2e43718802c8ef845e6696637e46930 Mon Sep 17 00:00:00 2001 From: Scott Moser Date: Sat, 1 Dec 2012 21:46:27 -0500 Subject: make sure no blank lines before cloud-init entry in ca-certificates.conf when /etc/ca-certificates.conf is read by update-ca-certificates lines after a blank line get ignored. Here, ensure that there are no blank lines, and no duplicate entries for cloud-init are added. LP: #1077020 --- ChangeLog | 2 + cloudinit/config/cc_ca_certs.py | 9 +++- .../test_handler/test_handler_ca_certs.py | 50 +++++++++++++++++++--- 3 files changed, 55 insertions(+), 6 deletions(-) (limited to 'tests/unittests/test_handler') diff --git a/ChangeLog b/ChangeLog index bd52f182..13afb2c2 100644 --- a/ChangeLog +++ b/ChangeLog @@ -2,6 +2,8 @@ - add a debian watch file - add 'sudo' entry to ubuntu's default user (LP: #1080717) - fix resizefs module when 'noblock' was provided (LP: #1080985) + - make sure there is no blank line before cloud-init entry in + there are no blank lines in /etc/ca-certificates.conf (LP: #1077020) 0.7.1: - sysvinit: fix missing dependency in cloud-init job for RHEL 5.6 - config-drive: map hostname to local-hostname (LP: #1061964) diff --git a/cloudinit/config/cc_ca_certs.py b/cloudinit/config/cc_ca_certs.py index 20f24357..4f2a46a1 100644 --- a/cloudinit/config/cc_ca_certs.py +++ b/cloudinit/config/cc_ca_certs.py @@ -45,8 +45,15 @@ def add_ca_certs(certs): # First ensure they are strings... cert_file_contents = "\n".join([str(c) for c in certs]) util.write_file(CA_CERT_FULL_PATH, cert_file_contents, mode=0644) + # Append cert filename to CA_CERT_CONFIG file. - util.write_file(CA_CERT_CONFIG, "\n%s" % CA_CERT_FILENAME, omode="ab") + # We have to strip the content because blank lines in the file + # causes subsequent entries to be ignored. (LP: #1077020) + orig = util.load_file(CA_CERT_CONFIG) + cur_cont = '\n'.join([l for l in orig.splitlines() + if l != CA_CERT_FILENAME]) + out = "%s\n%s\n" % (cur_cont.rstrip(), CA_CERT_FILENAME) + util.write_file(CA_CERT_CONFIG, out, omode="wb") def remove_default_ca_certs(): diff --git a/tests/unittests/test_handler/test_handler_ca_certs.py b/tests/unittests/test_handler/test_handler_ca_certs.py index d73c9fa9..0558023a 100644 --- a/tests/unittests/test_handler/test_handler_ca_certs.py +++ b/tests/unittests/test_handler/test_handler_ca_certs.py @@ -138,15 +138,47 @@ class TestAddCaCerts(MockerTestCase): self.mocker.replay() cc_ca_certs.add_ca_certs([]) - def test_single_cert(self): - """Test adding a single certificate to the trusted CAs.""" + def test_single_cert_trailing_cr(self): + """Test adding a single certificate to the trusted CAs + when existing ca-certificates has trailing newline""" cert = "CERT1\nLINE2\nLINE3" + ca_certs_content = "line1\nline2\ncloud-init-ca-certs.crt\nline3\n" + expected = "line1\nline2\nline3\ncloud-init-ca-certs.crt\n" + + mock_write = self.mocker.replace(util.write_file, passthrough=False) + mock_load = self.mocker.replace(util.load_file, passthrough=False) + + mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt", + cert, mode=0644) + + mock_load("/etc/ca-certificates.conf") + self.mocker.result(ca_certs_content) + + mock_write("/etc/ca-certificates.conf", expected, omode="wb") + self.mocker.replay() + + cc_ca_certs.add_ca_certs([cert]) + + def test_single_cert_no_trailing_cr(self): + """Test adding a single certificate to the trusted CAs + when existing ca-certificates has no trailing newline""" + cert = "CERT1\nLINE2\nLINE3" + + ca_certs_content = "line1\nline2\nline3" + mock_write = self.mocker.replace(util.write_file, passthrough=False) + mock_load = self.mocker.replace(util.load_file, passthrough=False) + mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt", cert, mode=0644) + + mock_load("/etc/ca-certificates.conf") + self.mocker.result(ca_certs_content) + mock_write("/etc/ca-certificates.conf", - "\ncloud-init-ca-certs.crt", omode="ab") + "%s\n%s\n" % (ca_certs_content, "cloud-init-ca-certs.crt"), + omode="wb") self.mocker.replay() cc_ca_certs.add_ca_certs([cert]) @@ -157,10 +189,18 @@ class TestAddCaCerts(MockerTestCase): expected_cert_file = "\n".join(certs) mock_write = self.mocker.replace(util.write_file, passthrough=False) + mock_load = self.mocker.replace(util.load_file, passthrough=False) + mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt", expected_cert_file, mode=0644) - mock_write("/etc/ca-certificates.conf", - "\ncloud-init-ca-certs.crt", omode="ab") + + ca_certs_content = "line1\nline2\nline3" + mock_load("/etc/ca-certificates.conf") + self.mocker.result(ca_certs_content) + + out = "%s\n%s\n" % (ca_certs_content, "cloud-init-ca-certs.crt") + mock_write("/etc/ca-certificates.conf", out, omode="wb") + self.mocker.replay() cc_ca_certs.add_ca_certs(certs) -- cgit v1.2.3