test_jobq.py 文件源码

def test_exception(self):

        # Add a handler, or python complains "no handler assigned
        # to...."
        jl = logging.getLogger('pykit.jobq')
        jl.addHandler(logging.NullHandler())

        def err_on_even(args):
            if args % 2 == 0:
                raise Exception('even number')
            else:
                return args

        def collect(args):
            rst.append(args)

        rst = []
        jobq.run(range(10), [err_on_even, collect])
        self.assertEqual(list(range(1, 10, 2)), rst)

        # remove NullHandler
        jl.handlers = []