charm-ceilometer-agent/unit_tests/test_ceilometer_utils.py
Alex Kavanagh ed5416c6b8 Enhanced pause and resume for maintenance mode
Adds improved pause and resume unit to the charm such tha the
charm stays paused during maintenance operations.
Sync latest version of charm-helpers for maintenance mode.
Amulet test for pause/resume actions

Change-Id: I3ec090fb9eb85561834b79003606c4d4b38dc84c
2016-03-18 16:31:17 +00:00

103 lines
3.8 KiB
Python

from mock import call, MagicMock, patch
import ceilometer_utils as utils
from test_utils import CharmTestCase
TO_PATCH = [
'get_os_codename_package',
'templating',
'CeilometerServiceContext',
'config',
'get_os_codename_install_source',
'configure_installation_source',
'apt_install',
'apt_update',
'apt_upgrade',
'log'
]
class CeilometerUtilsTest(CharmTestCase):
def setUp(self):
super(CeilometerUtilsTest, self).setUp(utils, TO_PATCH)
def tearDown(self):
super(CeilometerUtilsTest, self).tearDown()
def test_register_configs(self):
configs = utils.register_configs()
calls = []
for conf in utils.CONFIG_FILES:
calls.append(call(conf,
utils.CONFIG_FILES[conf]['hook_contexts']))
configs.register.assert_has_calls(calls, any_order=True)
def test_restart_map(self):
restart_map = utils.restart_map()
self.assertEquals(restart_map,
{'/etc/ceilometer/ceilometer.conf': [
'ceilometer-agent-compute']})
def test_do_openstack_upgrade(self):
self.config.side_effect = self.test_config.get
self.test_config.set('openstack-origin', 'cloud:precise-havana')
self.get_os_codename_install_source.return_value = 'havana'
configs = MagicMock()
utils.do_openstack_upgrade(configs)
configs.set_release.assert_called_with(openstack_release='havana')
self.assertTrue(self.log.called)
self.apt_update.assert_called_with(fatal=True)
dpkg_opts = [
'--option', 'Dpkg::Options::=--force-confnew',
'--option', 'Dpkg::Options::=--force-confdef',
]
self.apt_install.assert_called_with(
packages=utils.CEILOMETER_AGENT_PACKAGES,
options=dpkg_opts, fatal=True
)
self.configure_installation_source.assert_called_with(
'cloud:precise-havana'
)
def test_assess_status(self):
with patch.object(utils, 'assess_status_func') as asf:
callee = MagicMock()
asf.return_value = callee
utils.assess_status('test-config')
asf.assert_called_once_with('test-config')
callee.assert_called_once_with()
@patch.object(utils, 'REQUIRED_INTERFACES')
@patch.object(utils, 'services')
@patch.object(utils, 'make_assess_status_func')
def test_assess_status_func(self,
make_assess_status_func,
services,
REQUIRED_INTERFACES):
services.return_value = 's1'
utils.assess_status_func('test-config')
# ports=None whilst port checks are disabled.
make_assess_status_func.assert_called_once_with(
'test-config', REQUIRED_INTERFACES, services='s1', ports=None)
def test_pause_unit_helper(self):
with patch.object(utils, '_pause_resume_helper') as prh:
utils.pause_unit_helper('random-config')
prh.assert_called_once_with(utils.pause_unit, 'random-config')
with patch.object(utils, '_pause_resume_helper') as prh:
utils.resume_unit_helper('random-config')
prh.assert_called_once_with(utils.resume_unit, 'random-config')
@patch.object(utils, 'services')
def test_pause_resume_helper(self, services):
f = MagicMock()
services.return_value = 's1'
with patch.object(utils, 'assess_status_func') as asf:
asf.return_value = 'assessor'
utils._pause_resume_helper(f, 'some-config')
asf.assert_called_once_with('some-config')
# ports=None whilst port checks are disabled.
f.assert_called_once_with('assessor', services='s1', ports=None)