summaryrefslogtreecommitdiff
path: root/tests/unittests/test__init__.py
diff options
context:
space:
mode:
authorBarry Warsaw <barry@python.org>2015-01-22 11:38:51 -0500
committerBarry Warsaw <barry@python.org>2015-01-22 11:38:51 -0500
commit6363b546c8bacb24b9350d108165a09bb65828a1 (patch)
treeea14c3bb4fc2915c9ad183b9a3ed21c93b11b3dc /tests/unittests/test__init__.py
parentf895cb12141281702b34da18f2384deb64c881e7 (diff)
downloadvyos-cloud-init-6363b546c8bacb24b9350d108165a09bb65828a1.tar.gz
vyos-cloud-init-6363b546c8bacb24b9350d108165a09bb65828a1.zip
Port test__init__.py to unittest.mock.
Diffstat (limited to 'tests/unittests/test__init__.py')
-rw-r--r--tests/unittests/test__init__.py239
1 files changed, 115 insertions, 124 deletions
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