Port test__init__.py to unittest.mock.

This commit is contained in:
Barry Warsaw 2015-01-22 11:38:51 -05:00
parent 474b7f259d
commit ab0bff66f6
4 changed files with 115 additions and 142 deletions

View File

@ -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):

View File

@ -1,10 +1,19 @@
import os
import shutil
import tempfile
import unittest
from mocker import MockerTestCase, ARGS, KWARGS
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)
self.assertEqual(0, self.data["handlercount"])
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)
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)
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(MockerTestCase):
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

View File

@ -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_")

View File

@ -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