test_influxdb.py 文件源码

def workload_fixture_generator(command):
    @pytest.yield_fixture
    def workload(admin_node):

        pids = set()

        def _make_workload():
            pid = admin_node.check_call('{} & echo $!'.format(
                command)).stdout_str
            # check process is running
            cmd = 'ls /proc/{}'.format(pid)
            err_msg = "Background process with pid `{}` is not found".format(
                pid)
            assert admin_node.execute(cmd).exit_code == 0, err_msg
            pids.add(pid)

        yield _make_workload

        for pid in pids:
            admin_node.execute('kill {}'.format(pid))

    return workload