test_service.py 文件源码

def test_log_follow():
    package_install('chronos', deploy=True)

    args = ['dcos', 'service', 'log', 'chronos', '--follow']

    if sys.platform == 'win32':
        proc = subprocess.Popen(
            args,
            creationflags=subprocess.CREATE_NEW_PROCESS_GROUP,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE)
    else:
        # disable stdout/stderr buffering:
        # https://docs.python.org/3/using/cmdline.html#cmdoption-u
        my_env = os.environ.copy()
        my_env['PYTHONUNBUFFERED'] = 'x'

        # os.setsid is only available for Unix:
        # https://docs.python.org/2/library/os.html#os.setsid
        proc = subprocess.Popen(
            args,
            preexec_fn=os.setsid,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            env=my_env)

    time.sleep(10)

    proc.poll()
    assert proc.returncode is None

    if sys.platform == 'win32':
        os.kill(proc.pid, signal.CTRL_BREAK_EVENT)
    else:
        # using Unix-only commands os.killpg + os.getgid
        # https://docs.python.org/2/library/os.html#os.killpg
        # https://docs.python.org/2/library/os.html#os.getpgid
        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

    assert_lines(['dcos', 'service', 'log', 'chronos', '--lines=4'], 4)

    exec_command(['dcos', 'package', 'uninstall', 'chronos', '--yes'])