test_job_command.py 文件源码

def test_job_command(bucket):
    c = client_from_commandline_args([
        "--kubeface-poll-seconds", "1.1",
        "--kubeface-backend", "local-process",
        "--kubeface-storage", bucket,
    ])

    mapper = c.map(math.exp, range(10), cache_key='FOOBARBAZ')
    testing.assert_equal(next(mapper), 1)
    assert 'FOOBARBAZ' in run_job_command(bucket, [])
    assert 'active' in (
        find_line_with(
            "FOOBARBAZ",
            run_job_command(bucket, ["--include-done"]),
            nth=1))
    list(mapper)
    assert 'FOOBARBAZ' not in run_job_command(bucket, [])