"Fossies" - the Fresh Open Source Software Archive  

Source code changes of the file "ironic/tests/unit/conductor/test_utils.py" between
ironic-16.0.2.tar.gz and ironic-16.0.3.tar.gz

About: OpenStack Ironic (Optional Service: Bare-Metal Provisioning) aims to provision bare metal machines instead of virtual machines, forked from the Nova baremetal driver.
The "Victoria" series (latest release).

test_utils.py  (ironic-16.0.2):test_utils.py  (ironic-16.0.3)
skipping to change at line 1061 skipping to change at line 1061
def test_provision_error_handler_other_error(self, log_mock): def test_provision_error_handler_other_error(self, log_mock):
exc = Exception('foo') exc = Exception('foo')
conductor_utils.provisioning_error_handler(exc, self.node, 'state-one', conductor_utils.provisioning_error_handler(exc, self.node, 'state-one',
'state-two') 'state-two')
self.assertFalse(self.node.save.called) self.assertFalse(self.node.save.called)
self.assertFalse(log_mock.warning.called) self.assertFalse(log_mock.warning.called)
@mock.patch.object(conductor_utils, 'cleaning_error_handler', @mock.patch.object(conductor_utils, 'cleaning_error_handler',
autospec=True) autospec=True)
def test_cleanup_cleanwait_timeout_handler_call(self, mock_error_handler): def test_cleanup_cleanwait_timeout_handler_call(self, mock_error_handler):
self.task.node.uuid = '18c95393-b775-4887-a274-c45be47509d5'
self.node.clean_step = {} self.node.clean_step = {}
conductor_utils.cleanup_cleanwait_timeout(self.task) conductor_utils.cleanup_cleanwait_timeout(self.task)
mock_error_handler.assert_called_once_with( mock_error_handler.assert_called_once_with(
self.task, self.task,
msg="Timeout reached while cleaning the node. Please " logmsg="Cleaning for node 18c95393-b775-4887-a274-c45be47509d5 "
"check if the ramdisk responsible for the cleaning is " "failed. Timeout reached while cleaning the node. Please "
"running on the node. Failed on step {}.", "check if the ramdisk responsible for the cleaning is "
"running on the node. Failed on step {}.",
errmsg="Timeout reached while cleaning the node. Please "
"check if the ramdisk responsible for the cleaning is "
"running on the node. Failed on step {}.",
set_fail_state=False) set_fail_state=False)
def test_cleanup_cleanwait_timeout(self): def test_cleanup_cleanwait_timeout(self):
self.node.provision_state = states.CLEANFAIL self.node.provision_state = states.CLEANFAIL
target = 'baz' target = 'baz'
self.node.target_provision_state = target self.node.target_provision_state = target
self.node.driver_internal_info = {} self.node.driver_internal_info = {}
self.node.clean_step = {'key': 'val'} self.node.clean_step = {'key': 'val'}
clean_error = ("Timeout reached while cleaning the node. Please " clean_error = ("Timeout reached while cleaning the node. Please "
"check if the ramdisk responsible for the cleaning is " "check if the ramdisk responsible for the cleaning is "
skipping to change at line 1091 skipping to change at line 1096
'cleaning_reboot': True, 'cleaning_reboot': True,
'clean_step_index': 0} 'clean_step_index': 0}
conductor_utils.cleanup_cleanwait_timeout(self.task) conductor_utils.cleanup_cleanwait_timeout(self.task)
self.assertEqual({}, self.node.clean_step) self.assertEqual({}, self.node.clean_step)
self.assertNotIn('clean_step_index', self.node.driver_internal_info) self.assertNotIn('clean_step_index', self.node.driver_internal_info)
self.assertFalse(self.task.process_event.called) self.assertFalse(self.task.process_event.called)
self.assertTrue(self.node.maintenance) self.assertTrue(self.node.maintenance)
self.assertEqual(clean_error, self.node.maintenance_reason) self.assertEqual(clean_error, self.node.maintenance_reason)
self.assertEqual('clean failure', self.node.fault) self.assertEqual('clean failure', self.node.fault)
def _test_cleaning_error_handler(self, prov_state=states.CLEANING): @mock.patch.object(conductor_utils.LOG, 'error', autospec=True)
def _test_cleaning_error_handler(self, mock_log_error,
prov_state=states.CLEANING):
self.node.provision_state = prov_state self.node.provision_state = prov_state
target = 'baz' target = 'baz'
self.node.target_provision_state = target self.node.target_provision_state = target
self.node.clean_step = {'key': 'val'} self.node.clean_step = {'key': 'val'}
self.node.driver_internal_info = { self.node.driver_internal_info = {
'cleaning_reboot': True, 'cleaning_reboot': True,
'cleaning_polling': True, 'cleaning_polling': True,
'skip_current_clean_step': True, 'skip_current_clean_step': True,
'clean_step_index': 0, 'clean_step_index': 0,
'agent_url': 'url'} 'agent_url': 'url'}
msg = 'error bar' msg = 'error bar'
conductor_utils.cleaning_error_handler(self.task, msg) last_error = "last error"
conductor_utils.cleaning_error_handler(self.task, msg,
errmsg=last_error)
self.node.save.assert_called_once_with() self.node.save.assert_called_once_with()
self.assertEqual({}, self.node.clean_step) self.assertEqual({}, self.node.clean_step)
self.assertNotIn('clean_step_index', self.node.driver_internal_info) self.assertNotIn('clean_step_index', self.node.driver_internal_info)
self.assertNotIn('cleaning_reboot', self.node.driver_internal_info) self.assertNotIn('cleaning_reboot', self.node.driver_internal_info)
self.assertNotIn('cleaning_polling', self.node.driver_internal_info) self.assertNotIn('cleaning_polling', self.node.driver_internal_info)
self.assertNotIn('skip_current_clean_step', self.assertNotIn('skip_current_clean_step',
self.node.driver_internal_info) self.node.driver_internal_info)
self.assertEqual(msg, self.node.last_error) self.assertEqual(last_error, self.node.last_error)
self.assertTrue(self.node.maintenance) self.assertTrue(self.node.maintenance)
self.assertEqual(msg, self.node.maintenance_reason) self.assertEqual(last_error, self.node.maintenance_reason)
self.assertEqual('clean failure', self.node.fault) self.assertEqual('clean failure', self.node.fault)
driver = self.task.driver.deploy driver = self.task.driver.deploy
driver.tear_down_cleaning.assert_called_once_with(self.task) driver.tear_down_cleaning.assert_called_once_with(self.task)
if prov_state == states.CLEANFAIL: if prov_state == states.CLEANFAIL:
self.assertFalse(self.task.process_event.called) self.assertFalse(self.task.process_event.called)
else: else:
self.task.process_event.assert_called_once_with('fail', self.task.process_event.assert_called_once_with('fail',
target_state=None) target_state=None)
self.assertNotIn('agent_url', self.node.driver_internal_info) self.assertNotIn('agent_url', self.node.driver_internal_info)
mock_log_error.assert_called_once_with(msg, exc_info=False)
def test_cleaning_error_handler(self): def test_cleaning_error_handler(self):
self._test_cleaning_error_handler() self._test_cleaning_error_handler()
def test_cleaning_error_handler_cleanwait(self): def test_cleaning_error_handler_cleanwait(self):
self._test_cleaning_error_handler(prov_state=states.CLEANWAIT) self._test_cleaning_error_handler(prov_state=states.CLEANWAIT)
def test_cleaning_error_handler_cleanfail(self): def test_cleaning_error_handler_cleanfail(self):
self._test_cleaning_error_handler(prov_state=states.CLEANFAIL) self._test_cleaning_error_handler(prov_state=states.CLEANFAIL)
skipping to change at line 1167 skipping to change at line 1177
def test_cleaning_error_handler_tear_down_error(self, log_mock): def test_cleaning_error_handler_tear_down_error(self, log_mock):
def _side_effect(task): def _side_effect(task):
# simulate overwriting last error by another operation (e.g. power) # simulate overwriting last error by another operation (e.g. power)
task.node.last_error = None task.node.last_error = None
raise Exception('bar') raise Exception('bar')
driver = self.task.driver.deploy driver = self.task.driver.deploy
msg = 'foo' msg = 'foo'
driver.tear_down_cleaning.side_effect = _side_effect driver.tear_down_cleaning.side_effect = _side_effect
conductor_utils.cleaning_error_handler(self.task, msg) conductor_utils.cleaning_error_handler(self.task, msg)
log_mock.error.assert_called_once_with(msg, exc_info=False)
self.assertTrue(log_mock.exception.called) self.assertTrue(log_mock.exception.called)
self.assertIn(msg, self.node.last_error) self.assertIn(msg, self.node.last_error)
self.assertIn(msg, self.node.maintenance_reason) self.assertIn(msg, self.node.maintenance_reason)
self.assertEqual('clean failure', self.node.fault) self.assertEqual('clean failure', self.node.fault)
def test_abort_on_conductor_take_over_cleaning(self): def test_abort_on_conductor_take_over_cleaning(self):
self.node.provision_state = states.CLEANFAIL self.node.provision_state = states.CLEANFAIL
conductor_utils.abort_on_conductor_take_over(self.task) conductor_utils.abort_on_conductor_take_over(self.task)
self.assertTrue(self.node.maintenance) self.assertTrue(self.node.maintenance)
self.assertIn('take over', self.node.maintenance_reason) self.assertIn('take over', self.node.maintenance_reason)
 End of changes. 8 change blocks. 
7 lines changed or deleted 18 lines changed or added

Home  |  About  |  Features  |  All  |  Newest  |  Dox  |  Diffs  |  RSS Feeds  |  Screenshots  |  Comments  |  Imprint  |  Privacy  |  HTTP(S)