test_service.py 文件源码

def test_log_follow():
    wait_for_service('chronos')
    proc = subprocess.Popen(['dcos', 'service', 'log', 'chronos', '--follow'],
                            preexec_fn=os.setsid,
                            stdout=subprocess.PIPE,
                            stderr=subprocess.PIPE)
    time.sleep(10)

    proc.poll()
    assert proc.returncode is None

    os.killpg(os.getpgid(proc.pid), 15)

    stdout = proc.stdout.read()
    stderr = proc.stderr.read()

    print('STDOUT: {}'.format(stdout))
    print('STDERR: {}'.format(stderr))
    assert len(stdout.decode('utf-8').split('\n')) > 3