test_job_wrapper.py 文件源码

 复制代码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()