test_job_wrapper.py 文件源码

def test_job_wrapper_fills_pipe_with_exception_info(self):
        module_name = self.test_jobs_module.constants.THROW_AN_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()
        # Get to the bottom level of the exception information
        while ('type' not in exec_info) and (exec_info is not None):
            exec_info = exec_info.get('previous_exception', None)
        try:
            self.assertEqual(exec_info['type'], str(TypeError))
            self.assertEqual(exec_info['message'], "THE SKY IS FALLING!!")
        except Exception, e:
            raise e
        finally:
            to_job.close()
            to_self.close()