test_client.py 文件源码

def test_worker_exception_delayed(bucket):
    c = client_from_commandline_args([
        "--kubeface-poll-seconds", "1.1",
        "--kubeface-backend", "local-process",
        "--kubeface-storage", bucket,
        "--kubeface-wait-to-raise-task-exception",
    ])
    mapper = c.map(lambda x: 2 / (x - 2), range(10))
    testing.assert_equal(next(mapper), -1)
    testing.assert_equal(next(mapper), -2)
    testing.assert_equal(len(c.job_summary(include_done=False)), 1)
    testing.assert_equal(len(c.job_summary(include_done=True)), 1)
    testing.assert_raises(ZeroDivisionError, next, mapper)
    testing.assert_equal(len(c.job_summary(include_done=False)), 0)
    testing.assert_equal(len(c.job_summary(include_done=True)), 1)
    testing.assert_raises(StopIteration, next, mapper)
    testing.assert_equal(len(c.job_summary(include_done=False)), 0)
    testing.assert_equal(len(c.job_summary(include_done=True)), 1)