diff options
| -rw-r--r-- | tests/unittests/helpers.py | 19 | ||||
| -rw-r--r-- | tests/unittests/test__init__.py | 239 | ||||
| -rw-r--r-- | tests/unittests/test_util.py | 3 | ||||
| -rw-r--r-- | tox.ini | 2 |
4 files changed, 118 insertions, 145 deletions
diff --git a/tests/unittests/helpers.py b/tests/unittests/helpers.py index 52305397..4ca460f1 100644 --- a/tests/unittests/helpers.py +++ b/tests/unittests/helpers.py @@ -2,11 +2,6 @@ import os import sys import unittest -from contextlib import contextmanager - -from mocker import Mocker -from mocker import MockerTestCase - from cloudinit import helpers as ch from cloudinit import util @@ -86,17 +81,6 @@ else: pass -@contextmanager -def mocker(verify_calls=True): - m = Mocker() - try: - yield m - finally: - m.restore() - if verify_calls: - m.verify() - - # Makes the old path start # with new base instead of whatever # it previously had @@ -126,9 +110,8 @@ def retarget_many_wrapper(new_base, am, old_func): return wrapper -class ResourceUsingTestCase(MockerTestCase): +class ResourceUsingTestCase(unittest.TestCase): def __init__(self, methodName="runTest"): - MockerTestCase.__init__(self, methodName) self.resource_path = None def resourceLocation(self, subname=None): diff --git a/tests/unittests/test__init__.py b/tests/unittests/test__init__.py index 48db1a5e..ce4704d8 100644 --- a/tests/unittests/test__init__.py +++ b/tests/unittests/test__init__.py @@ -1,10 +1,19 @@ import os - -from mocker import MockerTestCase, ARGS, KWARGS +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 from cloudinit import handlers from cloudinit import helpers -from cloudinit import importer from cloudinit import settings from cloudinit import url_helper from cloudinit import util @@ -22,76 +31,76 @@ class FakeModule(handlers.Handler): pass -class TestWalkerHandleHandler(MockerTestCase): +class TestWalkerHandleHandler(unittest.TestCase): def setUp(self): - - MockerTestCase.setUp(self) + unittest.TestCase.setUp(self) + tmpdir = tempfile.mkdtemp() + self.addCleanup(shutil.rmtree, tmpdir) self.data = { "handlercount": 0, "frequency": "", - "handlerdir": self.makeDir(), + "handlerdir": tmpdir, "handlers": helpers.ContentHandlers(), "data": None} self.expected_module_name = "part-handler-%03d" % ( self.data["handlercount"],) expected_file_name = "%s.py" % self.expected_module_name - expected_file_fullname = os.path.join(self.data["handlerdir"], - expected_file_name) + self.expected_file_fullname = os.path.join( + self.data["handlerdir"], expected_file_name) self.module_fake = FakeModule() self.ctype = None self.filename = None self.payload = "dummy payload" - # Mock the write_file function - write_file_mock = self.mocker.replace(util.write_file, - passthrough=False) - write_file_mock(expected_file_fullname, self.payload, 0o600) + # Mock the write_file() function. We'll assert that it got called as + # expected in each of the individual tests. + self.resources = ExitStack() + self.write_file_mock = self.resources.enter_context( + mock.patch('cloudinit.util.write_file')) + + def tearDown(self): + self.resources.close() + unittest.TestCase.tearDown(self) def test_no_errors(self): """Payload gets written to file and added to C{pdata}.""" - import_mock = self.mocker.replace(importer.import_module, - passthrough=False) - import_mock(self.expected_module_name) - self.mocker.result(self.module_fake) - self.mocker.replay() - - handlers.walker_handle_handler(self.data, self.ctype, self.filename, - self.payload) - - self.assertEqual(1, self.data["handlercount"]) + with mock.patch('cloudinit.importer.import_module', + return_value=self.module_fake) as mockobj: + handlers.walker_handle_handler(self.data, self.ctype, + self.filename, self.payload) + mockobj.assert_called_with_once(self.expected_module_name) + self.write_file_mock.assert_called_with_once( + self.expected_file_fullname, self.payload, 0o600) + self.assertEqual(self.data['handlercount'], 1) def test_import_error(self): """Module import errors are logged. No handler added to C{pdata}.""" - import_mock = self.mocker.replace(importer.import_module, - passthrough=False) - import_mock(self.expected_module_name) - self.mocker.throw(ImportError()) - self.mocker.replay() - - handlers.walker_handle_handler(self.data, self.ctype, self.filename, - self.payload) - - self.assertEqual(0, self.data["handlercount"]) + with mock.patch('cloudinit.importer.import_module', + side_effect=ImportError) as mockobj: + handlers.walker_handle_handler(self.data, self.ctype, + self.filename, self.payload) + mockobj.assert_called_with_once(self.expected_module_name) + self.write_file_mock.assert_called_with_once( + self.expected_file_fullname, self.payload, 0o600) + self.assertEqual(self.data['handlercount'], 0) def test_attribute_error(self): """Attribute errors are logged. No handler added to C{pdata}.""" - import_mock = self.mocker.replace(importer.import_module, - passthrough=False) - import_mock(self.expected_module_name) - self.mocker.result(self.module_fake) - self.mocker.throw(AttributeError()) - self.mocker.replay() - - handlers.walker_handle_handler(self.data, self.ctype, self.filename, - self.payload) + with mock.patch('cloudinit.importer.import_module', + side_effect=AttributeError, + return_value=self.module_fake) as mockobj: + handlers.walker_handle_handler(self.data, self.ctype, + self.filename, self.payload) + mockobj.assert_called_with_once(self.expected_module_name) + self.write_file_mock.assert_called_with_once( + self.expected_file_fullname, self.payload, 0o600) + self.assertEqual(self.data['handlercount'], 0) - self.assertEqual(0, self.data["handlercount"]) - -class TestHandlerHandlePart(MockerTestCase): +class TestHandlerHandlePart(unittest.TestCase): def setUp(self): self.data = "fake data" @@ -108,95 +117,80 @@ class TestHandlerHandlePart(MockerTestCase): C{handle_part} is called without C{frequency} for C{handler_version} == 1. """ - mod_mock = self.mocker.mock() - getattr(mod_mock, "frequency") - self.mocker.result(settings.PER_INSTANCE) - getattr(mod_mock, "handler_version") - self.mocker.result(1) - mod_mock.handle_part(self.data, self.ctype, self.filename, - self.payload) - self.mocker.replay() - - handlers.run_part(mod_mock, self.data, self.filename, - self.payload, self.frequency, self.headers) + mod_mock = mock.Mock(frequency=settings.PER_INSTANCE, + handler_version=1) + handlers.run_part(mod_mock, self.data, self.filename, self.payload, + self.frequency, self.headers) + # Assert that the handle_part() method of the mock object got + # called with the expected arguments. + mod_mock.handle_part.assert_called_with_once( + self.data, self.ctype, self.filename, self.payload) def test_normal_version_2(self): """ C{handle_part} is called with C{frequency} for C{handler_version} == 2. """ - mod_mock = self.mocker.mock() - getattr(mod_mock, "frequency") - self.mocker.result(settings.PER_INSTANCE) - getattr(mod_mock, "handler_version") - self.mocker.result(2) - mod_mock.handle_part(self.data, self.ctype, self.filename, - self.payload, self.frequency) - self.mocker.replay() - - handlers.run_part(mod_mock, self.data, self.filename, - self.payload, self.frequency, self.headers) + mod_mock = mock.Mock(frequency=settings.PER_INSTANCE, + handler_version=2) + handlers.run_part(mod_mock, self.data, self.filename, self.payload, + self.frequency, self.headers) + # Assert that the handle_part() method of the mock object got + # called with the expected arguments. + mod_mock.handle_part.assert_called_with_once( + self.data, self.ctype, self.filename, self.payload) def test_modfreq_per_always(self): """ C{handle_part} is called regardless of frequency if nofreq is always. """ self.frequency = "once" - mod_mock = self.mocker.mock() - getattr(mod_mock, "frequency") - self.mocker.result(settings.PER_ALWAYS) - getattr(mod_mock, "handler_version") - self.mocker.result(1) - mod_mock.handle_part(self.data, self.ctype, self.filename, - self.payload) - self.mocker.replay() - - handlers.run_part(mod_mock, self.data, self.filename, - self.payload, self.frequency, self.headers) + mod_mock = mock.Mock(frequency=settings.PER_ALWAYS, + handler_version=1) + handlers.run_part(mod_mock, self.data, self.filename, self.payload, + self.frequency, self.headers) + # Assert that the handle_part() method of the mock object got + # called with the expected arguments. + mod_mock.handle_part.assert_called_with_once( + self.data, self.ctype, self.filename, self.payload) def test_no_handle_when_modfreq_once(self): """C{handle_part} is not called if frequency is once.""" self.frequency = "once" - mod_mock = self.mocker.mock() - getattr(mod_mock, "frequency") - self.mocker.result(settings.PER_ONCE) - self.mocker.replay() - - handlers.run_part(mod_mock, self.data, self.filename, - self.payload, self.frequency, self.headers) + mod_mock = mock.Mock(frequency=settings.PER_ONCE) + handlers.run_part(mod_mock, self.data, self.filename, self.payload, + self.frequency, self.headers) + # Assert that the handle_part() method of the mock object got + # called with the expected arguments. + mod_mock.handle_part.assert_called_with_once( + self.data, self.ctype, self.filename, self.payload) def test_exception_is_caught(self): """Exceptions within C{handle_part} are caught and logged.""" - mod_mock = self.mocker.mock() - getattr(mod_mock, "frequency") - self.mocker.result(settings.PER_INSTANCE) - getattr(mod_mock, "handler_version") - self.mocker.result(1) - mod_mock.handle_part(self.data, self.ctype, self.filename, - self.payload) - self.mocker.throw(Exception()) - self.mocker.replay() - - handlers.run_part(mod_mock, self.data, self.filename, - self.payload, self.frequency, self.headers) - - -class TestCmdlineUrl(MockerTestCase): + mod_mock = mock.Mock(frequency=settings.PER_INSTANCE, + handler_version=1) + handlers.run_part(mod_mock, self.data, self.filename, self.payload, + self.frequency, self.headers) + mod_mock.handle_part.side_effect = Exception + handlers.run_part(mod_mock, self.data, self.filename, self.payload, + self.frequency, self.headers) + mod_mock.handle_part.assert_called_with_once( + self.data, self.ctype, self.filename, self.payload) + + +class TestCmdlineUrl(unittest.TestCase): def test_invalid_content(self): url = "http://example.com/foo" key = "mykey" payload = "0" cmdline = "ro %s=%s bar=1" % (key, url) - mock_readurl = self.mocker.replace(url_helper.readurl, - passthrough=False) - mock_readurl(url, ARGS, KWARGS) - self.mocker.result(url_helper.StringResponse(payload)) - self.mocker.replay() - - self.assertEqual((key, url, None), - util.get_cmdline_url(names=[key], starts="xxxxxx", - cmdline=cmdline)) + with mock.patch('cloudinit.url_helper.readurl', + return_value=url_helper.StringResponse(payload)): + self.assertEqual( + util.get_cmdline_url(names=[key], starts="xxxxxx", + cmdline=cmdline), + (key, url, None)) def test_valid_content(self): url = "http://example.com/foo" @@ -204,27 +198,24 @@ class TestCmdlineUrl(MockerTestCase): payload = "xcloud-config\nmydata: foo\nbar: wark\n" cmdline = "ro %s=%s bar=1" % (key, url) - mock_readurl = self.mocker.replace(url_helper.readurl, - passthrough=False) - mock_readurl(url, ARGS, KWARGS) - self.mocker.result(url_helper.StringResponse(payload)) - self.mocker.replay() - - self.assertEqual((key, url, payload), - util.get_cmdline_url(names=[key], starts="xcloud-config", - cmdline=cmdline)) + with mock.patch('cloudinit.url_helper.readurl', + return_value=url_helper.StringResponse(payload)): + self.assertEqual( + util.get_cmdline_url(names=[key], starts="xcloud-config", + cmdline=cmdline), + (key, url, payload)) def test_no_key_found(self): url = "http://example.com/foo" key = "mykey" cmdline = "ro %s=%s bar=1" % (key, url) - self.mocker.replace(url_helper.readurl, passthrough=False) - self.mocker.result(url_helper.StringResponse("")) - self.mocker.replay() + with mock.patch('cloudinit.url_helper.readurl', + return_value=url_helper.StringResponse('')): + self.assertEqual( + util.get_cmdline_url(names=["does-not-appear"], + starts="#cloud-config", cmdline=cmdline), + (None, None, None)) - self.assertEqual((None, None, None), - util.get_cmdline_url(names=["does-not-appear"], - starts="#cloud-config", cmdline=cmdline)) # vi: ts=4 expandtab diff --git a/tests/unittests/test_util.py b/tests/unittests/test_util.py index 203445b7..ee938969 100644 --- a/tests/unittests/test_util.py +++ b/tests/unittests/test_util.py @@ -2,7 +2,6 @@ import os import stat import yaml -from mocker import MockerTestCase from . import helpers import unittest @@ -61,7 +60,7 @@ class TestGetCfgOptionListOrStr(unittest.TestCase): self.assertEqual([], result) -class TestWriteFile(MockerTestCase): +class TestWriteFile(unittest.TestCase): def setUp(self): super(TestWriteFile, self).setUp() self.tmp = self.makeDir(prefix="unittest_") @@ -5,9 +5,9 @@ recreate = True [testenv] commands = python -m nose tests deps = + contextlib2 httpretty>=0.7.1 mock - mocker nose pep8==1.5.7 pyflakes |