Port test__init__.py to unittest.mock.

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