# Path: pgctl/daemontools.py
# class SvStat(
#         namedtuple('SvStat', ['state', 'pid', 'exitcode', 'seconds', 'process'])
# ):
#     __slots__ = ()
#     UNSUPERVISED = 'could not get status, supervisor is down'
#     INVALID = 'no such service'
# 
#     def __repr__(self):
#         format = '{0.state}'
#         if self.pid is not None:
#             format += ' (pid {0.pid})'
#         if self.exitcode is not None:
#             format += ' (exitcode {0.exitcode})'
#         if self.seconds is not None:
#             format += ' {0.seconds} seconds'
#         if self.process is not None:
#             format += ', {0.process}'
# 
#         return format.format(self)
# 
# Path: pgctl/daemontools.py
# def svstat(path):
#     return svstat_parse(svstat_string(path))
# 
# Path: pgctl/errors.py
# class LockHeld(PgctlUserMessage):
#     """The pgctl supervision lock is held. This generally indicates subprocesses escaping supervision."""
# 
# Path: pgctl/functions.py
# def show_runaway_processes(path):
#     from .fuser import fuser
#     processes = ps(fuser(path))
#     if processes:
#         raise LockHeld(
#             '''\
# these runaway processes did not stop:
# {}
# This usually means these processes are buggy.
# Normally pgctl would kill these automatically for you, but you specified the --no-force option.
# Learn more: https://pgctl.readthedocs.org/en/latest/user/quickstart.html#writing-playground-services
# '''.format(processes)
#         )
# 
# Path: pgctl/fuser.py
# def fuser(path, allow_deleted=False):
#     """Return the list of pids that have 'path' open, for the current user"""
#     search = stat(path)
#     if search is None and not allow_deleted:
#         return
# 
#     from glob import glob
#     for fddir in glob('/proc/*/fd/'):
#         try:
#             pid = int(fddir.split('/', 3)[2])
#         except ValueError:
#             continue
# 
#         fds = listdir(fddir)
#         for fd in fds:
#             from os.path import join
#             fd = join(fddir, fd)
#             found = stat(fd)
#             if found is None:
#                 # fd disappeared since we listed
#                 continue
# 
#             if found == search:
#                 yield pid
#                 break
# 
#             if allow_deleted and found.st_nlink == 0:
#                 from os import readlink
#                 if readlink(fd) == path + ' (deleted)':
#                     yield pid
#                     break
# 
# Path: pgctl/subprocess.py
# def set_defaults(func, **defaults):
#     def wrapped(cmd, **kwargs):
# PIPE = subprocess.PIPE
# STDOUT = subprocess.STDOUT


import os
import signal
import subprocess
import time
import pytest
from testing import norm
from testing.assertions import assert_svstat
from testing.assertions import wait_for
from testing.service_context import set_slow_shutdown_sleeptime
from testing.subprocess import assert_command
from pgctl.daemontools import SvStat
from pgctl.daemontools import svstat
from pgctl.errors import LockHeld
from pgctl.functions import show_runaway_processes
from pgctl.fuser import fuser
from pgctl.subprocess import check_call
    def service_name(self):
        yield 'orphan-subprocess'

    def it_starts_up_fine(self):
        assert_command(
            ('pgctl', 'start'),
            '',
            '''\
[pgctl] Starting: slow-startup, sweet
[pgctl] Started: sweet
[pgctl] Started: slow-startup
''',
            0,
        )
        assert_command(
            ('pgctl', 'log'),
            '''\
==> playground/slow-startup/logs/current <==
{TIMESTAMP} pgctl-poll-ready: service's ready check succeeded

==> playground/sweet/logs/current <==
{TIMESTAMP} sweet
{TIMESTAMP} sweet_error
''',
            '',
            0,
            norm=norm.pgctl,
        )

    def it_shows_error_but_still_succeeds_on_stop_for_sweet(self):

# Path: pgctl/daemontools.py
# class SvStat(
#         namedtuple('SvStat', ['state', 'pid', 'exitcode', 'seconds', 'process'])
# ):
#     __slots__ = ()
#     UNSUPERVISED = 'could not get status, supervisor is down'
#     INVALID = 'no such service'
# 
#     def __repr__(self):
#         format = '{0.state}'
#         if self.pid is not None:
#             format += ' (pid {0.pid})'
#         if self.exitcode is not None:
#             format += ' (exitcode {0.exitcode})'
#         if self.seconds is not None:
#             format += ' {0.seconds} seconds'
#         if self.process is not None:
#             format += ', {0.process}'
# 
#         return format.format(self)
# 
# Path: pgctl/daemontools.py
# def svstat(path):
#     return svstat_parse(svstat_string(path))
# 
# Path: pgctl/errors.py
# class LockHeld(PgctlUserMessage):
#     """The pgctl supervision lock is held. This generally indicates subprocesses escaping supervision."""
# 
# Path: pgctl/functions.py
# def show_runaway_processes(path):
#     from .fuser import fuser
#     processes = ps(fuser(path))
#     if processes:
#         raise LockHeld(
#             '''\
# these runaway processes did not stop:
# {}
# This usually means these processes are buggy.
# Normally pgctl would kill these automatically for you, but you specified the --no-force option.
# Learn more: https://pgctl.readthedocs.org/en/latest/user/quickstart.html#writing-playground-services
# '''.format(processes)
#         )
# 
# Path: pgctl/fuser.py
# def fuser(path, allow_deleted=False):
#     """Return the list of pids that have 'path' open, for the current user"""
#     search = stat(path)
#     if search is None and not allow_deleted:
#         return
# 
#     from glob import glob
#     for fddir in glob('/proc/*/fd/'):
#         try:
#             pid = int(fddir.split('/', 3)[2])
#         except ValueError:
#             continue
# 
#         fds = listdir(fddir)
#         for fd in fds:
#             from os.path import join
#             fd = join(fddir, fd)
#             found = stat(fd)
#             if found is None:
#                 # fd disappeared since we listed
#                 continue
# 
#             if found == search:
#                 yield pid
#                 break
# 
#             if allow_deleted and found.st_nlink == 0:
#                 from os import readlink
#                 if readlink(fd) == path + ' (deleted)':
#                     yield pid
#                     break
# 
# Path: pgctl/subprocess.py
# def set_defaults(func, **defaults):
#     def wrapped(cmd, **kwargs):
# PIPE = subprocess.PIPE
# STDOUT = subprocess.STDOUT


# Path: tests/spec/dirty_tests.py
import os
import signal
import subprocess
import time
import pytest
from testing import norm
from testing.assertions import assert_svstat
from testing.assertions import wait_for
from testing.service_context import set_slow_shutdown_sleeptime
from testing.subprocess import assert_command
from pgctl.daemontools import SvStat
from pgctl.daemontools import svstat
from pgctl.errors import LockHeld
from pgctl.functions import show_runaway_processes
from pgctl.fuser import fuser
from pgctl.subprocess import check_call

    def service_name(self):
        yield 'orphan-subprocess'

    def it_starts_up_fine(self):
        assert_command(
            ('pgctl', 'start'),
            '',
            '''\
[pgctl] Starting: slow-startup, sweet
[pgctl] Started: sweet
[pgctl] Started: slow-startup
''',
            0,
        )
        assert_command(
            ('pgctl', 'log'),
            '''\
==> playground/slow-startup/logs/current <==
{TIMESTAMP} pgctl-poll-ready: service's ready check succeeded

==> playground/sweet/logs/current <==
{TIMESTAMP} sweet
{TIMESTAMP} sweet_error
''',
            '',
            0,
            norm=norm.pgctl,
        )

    def it_shows_error_but_still_succeeds_on_stop_for_sweet(self):

        check_call(('pgctl', 'start', 'sweet'))