x/cloud-init
Code Issues Proposed changes

More conversions from mocker to mock.

Browse Source
This commit is contained in:
Barry Warsaw 2015-01-22 20:30:10 -05:00
parent dd0e2d78bd
commit 17ba3f5394
5 changed files with 153 additions and 120 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
tests/unittests/test_distros/test_resolv.py
View File

@ -1,8 +1,7 @@
from mocker import MockerTestCase
from cloudinit.distros.parsers import resolv_conf
import re
import unittest
BASE_RESOLVE = '''
@ -14,7 +13,7 @@ nameserver 10.15.30.92
BASE_RESOLVE = BASE_RESOLVE.strip()
class TestResolvHelper(MockerTestCase):
class TestResolvHelper(unittest.TestCase):
def test_parse_same(self):
rp = resolv_conf.ResolvConf(BASE_RESOLVE)
rp_r = str(rp).strip()

5
tests/unittests/test_distros/test_sysconfig.py
View File

@ -1,6 +1,5 @@
from mocker import MockerTestCase
import re
import unittest
from cloudinit.distros.parsers.sys_conf import SysConf
@ -8,7 +7,7 @@ from cloudinit.distros.parsers.sys_conf import SysConf
# Lots of good examples @
# http://content.hccfl.edu/pollock/AUnix1/SysconfigFilesDesc.txt
class TestSysConfHelper(MockerTestCase):
class TestSysConfHelper(unittest.TestCase):
# This function was added in 2.7, make it work for 2.6
def assertRegMatches(self, text, regexp):
regexp = re.compile(regexp)

4
tests/unittests/test_distros/test_user_data_normalize.py
View File

@ -1,4 +1,4 @@
from mocker import MockerTestCase
import unittest
from cloudinit import distros
from cloudinit import helpers
@ -15,7 +15,7 @@ bcfg = {
}
class TestUGNormalize(MockerTestCase):
class TestUGNormalize(unittest.TestCase):
def _make_distro(self, dtype, def_user=None):
cfg = dict(settings.CFG_BUILTIN)

10
tests/unittests/test_handler/test_handler_apt_configure.py
View File

@ -1,17 +1,19 @@
from mocker import MockerTestCase
from cloudinit import util
from cloudinit.config import cc_apt_configure
import os
import re
import shutil
import tempfile
import unittest
class TestAptProxyConfig(MockerTestCase):
class TestAptProxyConfig(unittest.TestCase):
def setUp(self):
super(TestAptProxyConfig, self).setUp()
self.tmp = self.makeDir()
self.tmp = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, self.tmp)
self.pfile = os.path.join(self.tmp, "proxy.cfg")
self.cfile = os.path.join(self.tmp, "config.cfg")

249
tests/unittests/test_handler/test_handler_ca_certs.py
View File

@ -1,5 +1,3 @@
from mocker import MockerTestCase
from cloudinit import cloud
from cloudinit import helpers
from cloudinit import util
@ -7,9 +5,21 @@ from cloudinit import util
from cloudinit.config import cc_ca_certs
import logging
import shutil
import tempfile
import unittest
try:
from unittest import mock
except ImportError:
import mock
try:
from contextlib import ExitStack
except ImportError:
from contextlib2 import ExitStack
class TestNoConfig(MockerTestCase):
class TestNoConfig(unittest.TestCase):
def setUp(self):
super(TestNoConfig, self).setUp()
self.name = "ca-certs"
@ -22,15 +32,20 @@ class TestNoConfig(MockerTestCase):
Test that nothing is done if no ca-certs configuration is provided.
"""
config = util.get_builtin_cfg()
self.mocker.replace(util.write_file, passthrough=False)
self.mocker.replace(cc_ca_certs.update_ca_certs, passthrough=False)
self.mocker.replay()
with ExitStack() as mocks:
util_mock = mocks.enter_context(
mock.patch.object(util, 'write_file'))
certs_mock = mocks.enter_context(
mock.patch.object(cc_ca_certs, 'update_ca_certs'))
cc_ca_certs.handle(self.name, config, self.cloud_init, self.log,
self.args)
cc_ca_certs.handle(self.name, config, self.cloud_init, self.log,
self.args)
self.assertEqual(util_mock.call_count, 0)
self.assertEqual(certs_mock.call_count, 0)
class TestConfig(MockerTestCase):
class TestConfig(unittest.TestCase):
def setUp(self):
super(TestConfig, self).setUp()
self.name = "ca-certs"
@ -39,16 +54,16 @@ class TestConfig(MockerTestCase):
self.log = logging.getLogger("TestNoConfig")
self.args = []
# Mock out the functions that actually modify the system
self.mock_add = self.mocker.replace(cc_ca_certs.add_ca_certs,
passthrough=False)
self.mock_update = self.mocker.replace(cc_ca_certs.update_ca_certs,
passthrough=False)
self.mock_remove = self.mocker.replace(
cc_ca_certs.remove_default_ca_certs, passthrough=False)
self.mocks = ExitStack()
self.addCleanup(self.mocks.close)
# Order must be correct
self.mocker.order()
# Mock out the functions that actually modify the system
self.mock_add = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'add_ca_certs'))
self.mock_update = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'update_ca_certs'))
self.mock_remove = self.mocks.enter_context(
mock.patch.object(cc_ca_certs, 'remove_default_ca_certs'))
def test_no_trusted_list(self):
"""
@ -57,86 +72,88 @@ class TestConfig(MockerTestCase):
"""
config = {"ca-certs": {}}
# No functions should be called
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_empty_trusted_list(self):
"""Test that no certificate are written if 'trusted' list is empty."""
config = {"ca-certs": {"trusted": []}}
# No functions should be called
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_single_trusted(self):
"""Test that a single cert gets passed to add_ca_certs."""
config = {"ca-certs": {"trusted": ["CERT1"]}}
self.mock_add(["CERT1"])
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.mock_add.assert_called_once_with(['CERT1'])
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_multiple_trusted(self):
"""Test that multiple certs get passed to add_ca_certs."""
config = {"ca-certs": {"trusted": ["CERT1", "CERT2"]}}
self.mock_add(["CERT1", "CERT2"])
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.mock_add.assert_called_once_with(['CERT1', 'CERT2'])
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_remove_default_ca_certs(self):
"""Test remove_defaults works as expected."""
config = {"ca-certs": {"remove-defaults": True}}
self.mock_remove()
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 1)
def test_no_remove_defaults_if_false(self):
"""Test remove_defaults is not called when config value is False."""
config = {"ca-certs": {"remove-defaults": False}}
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.assertEqual(self.mock_add.call_count, 0)
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 0)
def test_correct_order_for_remove_then_add(self):
"""Test remove_defaults is not called when config value is False."""
config = {"ca-certs": {"remove-defaults": True, "trusted": ["CERT1"]}}
self.mock_remove()
self.mock_add(["CERT1"])
self.mock_update()
self.mocker.replay()
cc_ca_certs.handle(self.name, config, self.cloud, self.log, self.args)
self.mock_add.assert_called_once_with(['CERT1'])
self.assertEqual(self.mock_update.call_count, 1)
self.assertEqual(self.mock_remove.call_count, 1)
class TestAddCaCerts(MockerTestCase):
class TestAddCaCerts(unittest.TestCase):
def setUp(self):
super(TestAddCaCerts, self).setUp()
tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, tmpdir)
self.paths = helpers.Paths({
'cloud_dir': self.makeDir()
'cloud_dir': tmpdir,
})
def test_no_certs_in_list(self):
"""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([])
with mock.patch.object(util, 'write_file') as mockobj:
cc_ca_certs.add_ca_certs([])
self.assertEqual(mockobj.call_count, 0)
def test_single_cert_trailing_cr(self):
"""Test adding a single certificate to the trusted CAs
@ -146,19 +163,21 @@ class TestAddCaCerts(MockerTestCase):
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)
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_load = mocks.enter_context(
mock.patch.object(util, 'load_file',
return_value=ca_certs_content))
mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644)
cc_ca_certs.add_ca_certs([cert])
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])
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644),
mock.call("/etc/ca-certificates.conf", expected, omode="wb"),
])
mock_load.assert_called_once_with("/etc/ca-certificates.conf")
def test_single_cert_no_trailing_cr(self):
"""Test adding a single certificate to the trusted CAs
@ -167,75 +186,89 @@ class TestAddCaCerts(MockerTestCase):
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)
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_load = mocks.enter_context(
mock.patch.object(util, 'load_file',
return_value=ca_certs_content))
mock_write("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644)
cc_ca_certs.add_ca_certs([cert])
mock_load("/etc/ca-certificates.conf")
self.mocker.result(ca_certs_content)
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
cert, mode=0o644),
mock.call("/etc/ca-certificates.conf",
"%s\n%s\n" % (ca_certs_content,
"cloud-init-ca-certs.crt"),
omode="wb"),
])
mock_write("/etc/ca-certificates.conf",
"%s\n%s\n" % (ca_certs_content, "cloud-init-ca-certs.crt"),
omode="wb")
self.mocker.replay()
cc_ca_certs.add_ca_certs([cert])
mock_load.assert_called_once_with("/etc/ca-certificates.conf")
def test_multiple_certs(self):
"""Test adding multiple certificates to the trusted CAs."""
certs = ["CERT1\nLINE2\nLINE3", "CERT2\nLINE2\nLINE3"]
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=0o644)
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")
with ExitStack() as mocks:
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_load = mocks.enter_context(
mock.patch.object(util, 'load_file',
return_value=ca_certs_content))
self.mocker.replay()
cc_ca_certs.add_ca_certs(certs)
cc_ca_certs.add_ca_certs(certs)
mock_write.assert_has_calls([
mock.call("/usr/share/ca-certificates/cloud-init-ca-certs.crt",
expected_cert_file, mode=0o644),
mock.call("/etc/ca-certificates.conf",
"%s\n%s\n" % (ca_certs_content,
"cloud-init-ca-certs.crt"),
omode='wb'),
])
mock_load.assert_called_once_with("/etc/ca-certificates.conf")
class TestUpdateCaCerts(MockerTestCase):
class TestUpdateCaCerts(unittest.TestCase):
def test_commands(self):
mock_check_call = self.mocker.replace(util.subp,
passthrough=False)
mock_check_call(["update-ca-certificates"], capture=False)
self.mocker.replay()
cc_ca_certs.update_ca_certs()
with mock.patch.object(util, 'subp') as mockobj:
cc_ca_certs.update_ca_certs()
mockobj.assert_called_once_with(
["update-ca-certificates"], capture=False)
class TestRemoveDefaultCaCerts(MockerTestCase):
class TestRemoveDefaultCaCerts(unittest.TestCase):
def setUp(self):
super(TestRemoveDefaultCaCerts, self).setUp()
tmpdir = tempfile.mkdtemp()
self.addCleanup(shutil.rmtree, tmpdir)
self.paths = helpers.Paths({
'cloud_dir': self.makeDir()
'cloud_dir': tmpdir,
})
def test_commands(self):
mock_delete_dir_contents = self.mocker.replace(
util.delete_dir_contents, passthrough=False)
mock_write = self.mocker.replace(util.write_file, passthrough=False)
mock_subp = self.mocker.replace(util.subp,
passthrough=False)
with ExitStack() as mocks:
mock_delete = mocks.enter_context(
mock.patch.object(util, 'delete_dir_contents'))
mock_write = mocks.enter_context(
mock.patch.object(util, 'write_file'))
mock_subp = mocks.enter_context(mock.patch.object(util, 'subp'))
mock_delete_dir_contents("/usr/share/ca-certificates/")
mock_delete_dir_contents("/etc/ssl/certs/")
mock_write("/etc/ca-certificates.conf", "", mode=0o644)
mock_subp(('debconf-set-selections', '-'),
"ca-certificates ca-certificates/trust_new_crts select no")
self.mocker.replay()
cc_ca_certs.remove_default_ca_certs()
cc_ca_certs.remove_default_ca_certs()
mock_delete.assert_has_calls([
mock.call("/usr/share/ca-certificates/"),
mock.call("/etc/ssl/certs/"),
])
mock_write.assert_called_once_with(
"/etc/ca-certificates.conf", "", mode=0o644)
mock_subp.assert_called_once_with(
('debconf-set-selections', '-'),
"ca-certificates ca-certificates/trust_new_crts select no")