How to use pgctl - 10 common examples

def assert_works_interactively():
    read, write = os.openpty()
    pty.normalize_newlines(read)
    # setsid: this simulates the shell's job-control behavior
    proc = Popen(('setsid', 'pgctl', 'debug', 'greeter'), stdin=PIPE, stdout=write)
    os.close(write)

    try:
        assert read_line(read) == 'What is your name?\n'
        proc.stdin.write(b'Buck\n')
        proc.stdin.flush()
        assert read_line(read) == 'Hello, Buck.\n'
    finally:
        ctrl_c(proc)
        proc.wait()

def it_runs_before_debugging_a_service(self):
        proc = Popen(('setsid', 'pgctl', 'debug', 'sweet'), stdin=PIPE, stdout=PIPE)
        proc.stdin.close()
        try:
            assert proc.stdout.readline().decode('utf-8') == 'hello, i am a pre-start script in stdout\n'
        finally:
            ctrl_c(proc)
            proc.wait()

def it_stays_locked_for_the_lifetime_of_subprocesses(self, tmpfile):
        from pgctl.subprocess import Popen

        with flock(tmpfile):
            p = Popen(('sleep', '99999'))
            assert p.poll() is None

            assert_locked(tmpfile)

        assert_locked(tmpfile)

        p.terminate()
        assert p.wait() == -15

        with flock(tmpfile):
            print('oh hi there!')

def it_releases_lock_on_exit(self, tmpfile):
        with flock(tmpfile):
            print('oh, hi!')

        with flock(tmpfile):
            print('oh, hi!')

def it_does_start(self, in_example_dir):
        test_string = 'oh, hi there.\n'
        with open('input', 'w') as input:
            input.write(test_string)
        assert not os.path.isfile('output')

        check_call(('pgctl', 'start', 'tail'))
        wait_for(lambda: os.path.isfile('output'))
        assert open('output').read() == test_string

def it_fails_by_default(self):
        check_call(('pgctl', 'start'))
        assert_svstat('playground/sweet', state='up')
        assert_command(
            ('pgctl', 'stop'),
            '',
            '''\
[pgctl] Stopping: sweet
[pgctl] ERROR: service 'sweet' failed to stop after {TIME} seconds, its status is ready (pid {PID}) {TIME} seconds
==> playground/sweet/logs/current <==
{TIMESTAMP} sweet
{TIMESTAMP} sweet_error
[pgctl]
[pgctl] There might be useful information further up in the log; you can view it by running:
[pgctl]     less +G playground/sweet/logs/current
[pgctl] ERROR: Some services failed to stop: sweet
''',
            1,

def it_succeeds_on_forceful_stop(self):
        check_call(('pgctl', 'start'))
        assert_svstat('playground/sweet', state='up')
        assert_command(
            ('pgctl', 'stop', '--force'),
            '',
            '''\
[pgctl] Stopping: sweet
[pgctl] WARNING: Killing these runaway processes at user's request (--force):
{PS-HEADER}
{PS-STATS} sleep 2.5

Learn why they did not stop: http://pgctl.readthedocs.org/en/latest/user/quickstart.html#writing-playground-services

[pgctl] Stopped: sweet
''',
            0,
            norm=norm.pgctl,

def it_can_recover_from_a_git_clean(self, in_example_dir, service_name):
        def assert_status():
            assert_command(
                ('pgctl', 'status'),
                '''\
 ● sleep: down
 ● tail: ready
   └─ pid: {PID}, {TIME} seconds
''',
                '',
                0,
                norm=norm.pgctl,
            )

        check_call(('pgctl', 'start', 'tail'))
        assert_status()

        # simulate a git-clean: blow everything away, create a fresh copy
        parent_dir = in_example_dir.join('..')
        with parent_dir.as_cwd():
            in_example_dir.remove(rec=True)
            from testing import copy_example
            copy_example(service_name, parent_dir)

        assert_status()

def it_stops_multiple_services(self, in_example_dir):
        check_call(('pgctl', 'start', 'sleep', 'tail'))

        assert_svstat('playground/sleep', state='up')
        assert_svstat('playground/tail', state='up')

        check_call(('pgctl', 'stop', 'sleep', 'tail'))

        assert_svstat('playground/sleep', state=SvStat.UNSUPERVISED)
        assert_svstat('playground/tail', state=SvStat.UNSUPERVISED)

def it_displays_correctly_when_the_service_is_up_json(self, in_example_dir):
        check_call(('pgctl', 'start', 'sleep'))
        stdout, stderr, returncode = run(
            ('pgctl', '--json', 'status', 'sleep'),
        )
        assert returncode == 0
        assert json.loads(stdout) == {
            'sleep': {
                'exitcode': None,
                'pid': ANY_INTEGER(),
                'process': None,
                'seconds': ANY_INTEGER(),
                'state': 'ready',
            },
        }
        assert stderr == ''