def test_job_wrapper_fills_pipe_with_exception_info_bubble_up(self):
module_name = self.test_jobs_module.constants.JOB_FAILED_EXCEPTION_JOB_NAME
config = {}
to_job, to_self = Pipe()
job_wrapper(to_self, module_name, config, MockStatusUpdater(), entry_point_group_name="hoplite.test_jobs")
exec_info = to_job.recv()
exec_info = exec_info.get('previous_exception', None)
try:
self.assertEqual(exec_info['address'], "10.2.1.1")
self.assertEqual(exec_info['uuid'], "5")
self.assertIsInstance(exec_info['traceback'], types.TracebackType)
# Get to the very bottom level of the exception information
exec_info = exec_info.get('previous_exception', None)
self.assertEqual(exec_info['message'], "Test Message")
self.assertEqual(exec_info['type'], "Test Type String")
self.assertEqual(exec_info['exception_object'], "pickled_string")
except Exception, e:
raise e
finally:
to_job.close()
to_self.close()
评论列表
文章目录