5d1edaff81
might have to rewrite this for Python 2.6. Disable Cepko tests (test_cs_util.py) since they are essentially worthless. Convert test_azure to unittest.mock.
219 lines
8.5 KiB
Python
219 lines
8.5 KiB
Python
import os
|
|
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 settings
|
|
from cloudinit import url_helper
|
|
from cloudinit import util
|
|
|
|
|
|
class FakeModule(handlers.Handler):
|
|
def __init__(self):
|
|
handlers.Handler.__init__(self, settings.PER_ALWAYS)
|
|
self.types = []
|
|
|
|
def list_types(self):
|
|
return self.types
|
|
|
|
def handle_part(self, data, ctype, filename, payload, frequency):
|
|
pass
|
|
|
|
|
|
class TestWalkerHandleHandler(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
unittest.TestCase.setUp(self)
|
|
tmpdir = tempfile.mkdtemp()
|
|
self.addCleanup(shutil.rmtree, tmpdir)
|
|
|
|
self.data = {
|
|
"handlercount": 0,
|
|
"frequency": "",
|
|
"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
|
|
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. We'll assert that it got called as
|
|
# expected in each of the individual tests.
|
|
resources = ExitStack()
|
|
self.addCleanup(resources.close)
|
|
self.write_file_mock = resources.enter_context(
|
|
mock.patch('cloudinit.util.write_file'))
|
|
|
|
def test_no_errors(self):
|
|
"""Payload gets written to file and added to C{pdata}."""
|
|
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}."""
|
|
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}."""
|
|
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(unittest.TestCase):
|
|
|
|
def setUp(self):
|
|
self.data = "fake data"
|
|
self.ctype = "fake ctype"
|
|
self.filename = "fake filename"
|
|
self.payload = "fake payload"
|
|
self.frequency = settings.PER_INSTANCE
|
|
self.headers = {
|
|
'Content-Type': self.ctype,
|
|
}
|
|
|
|
def test_normal_version_1(self):
|
|
"""
|
|
C{handle_part} is called without C{frequency} for
|
|
C{handler_version} == 1.
|
|
"""
|
|
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 = 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 = 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 = 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 = 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)
|
|
|
|
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"
|
|
key = "mykey"
|
|
payload = "xcloud-config\nmydata: foo\nbar: wark\n"
|
|
cmdline = "ro %s=%s bar=1" % (key, url)
|
|
|
|
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)
|
|
|
|
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))
|
|
|
|
|
|
# vi: ts=4 expandtab
|