test_main.py 文件源码

def test_pipeline():
    '''Tests a few pipelines.'''
    global progresses

    server = HTTPServer(('', 9000), SaveHooks)
    thread = threading.Thread(target = server.serve_forever, daemon=True)
    thread.start()

    for spec in pipelines():
        if spec.pipeline_id.startswith('./tests/env/dummy/pipeline-test'):
            eid = gen_execution_id()
            status.get(spec.pipeline_id).queue_execution(eid, 'manual')
            success, _, _ = execute_pipeline(spec, eid, use_cache=False)
            assert success

    assert len(called_hooks) == 3
    assert called_hooks == [
        {"pipeline_id": "./tests/env/dummy/pipeline-test-hooks", "event": "queue"},
        {"pipeline_id": "./tests/env/dummy/pipeline-test-hooks", "event": "start"},
        {"pipeline_id": "./tests/env/dummy/pipeline-test-hooks", "event": "finish", "success": True,
         'stats': {'.dpp': {'out-datapackage-url': 'hooks-outputs/datapackage.json'},
                   'bytes': 258, 'count_of_rows': None,
                   'dataset_name': 'hook-tests', 'hash': 'f3f25f5ecd8e7e2c35d83139178072b8'}}
    ]
    assert progresses >= 1