Gecko [ dataharbor.africa ]

Name	Size	Permission
?;fixtures	[ DIR ]	drwxr-xr-x
__init__.py	20 B	-rw-r--r--
__init__.pyc	148 B	-rw-r--r--
__init__.pyo	148 B	-rw-r--r--
base.py	33.88 KB	-rw-r--r--
base.pyc	52.56 KB	-rw-r--r--
base.pyo	52.56 KB	-rw-r--r--
test_childutils.py	5.27 KB	-rw-r--r--
test_childutils.pyc	7.63 KB	-rw-r--r--
test_childutils.pyo	7.63 KB	-rw-r--r--
test_confecho.py	540 B	-rw-r--r--
test_confecho.pyc	1.25 KB	-rw-r--r--
test_confecho.pyo	1.25 KB	-rw-r--r--
test_datatypes.py	26.65 KB	-rw-r--r--
test_datatypes.pyc	47.5 KB	-rw-r--r--
test_datatypes.pyo	47.5 KB	-rw-r--r--
test_dispatchers.py	47.68 KB	-rw-r--r--
test_dispatchers.pyc	47.49 KB	-rw-r--r--
test_dispatchers.pyo	47.49 KB	-rw-r--r--
test_events.py	20.35 KB	-rw-r--r--
test_events.pyc	24.96 KB	-rw-r--r--
test_events.pyo	24.96 KB	-rw-r--r--
test_http.py	24.48 KB	-rw-r--r--
test_http.pyc	34.38 KB	-rw-r--r--
test_http.pyo	34.38 KB	-rw-r--r--
test_loggers.py	12.65 KB	-rw-r--r--
test_loggers.pyc	18.18 KB	-rw-r--r--
test_loggers.pyo	18.18 KB	-rw-r--r--
test_options.py	132.88 KB	-rw-r--r--
test_options.pyc	136.01 KB	-rw-r--r--
test_options.pyo	136.01 KB	-rw-r--r--
test_poller.py	16.29 KB	-rw-r--r--
test_poller.pyc	20.74 KB	-rw-r--r--
test_poller.pyo	20.5 KB	-rw-r--r--
test_process.py	91.59 KB	-rw-r--r--
test_process.pyc	93.63 KB	-rw-r--r--
test_process.pyo	93.63 KB	-rw-r--r--
test_rpcinterfaces.py	94.86 KB	-rw-r--r--
test_rpcinterfaces.pyc	80.93 KB	-rw-r--r--
test_rpcinterfaces.pyo	80.93 KB	-rw-r--r--
test_socket_manager.py	7.84 KB	-rw-r--r--
test_socket_manager.pyc	11.29 KB	-rw-r--r--
test_socket_manager.pyo	11.29 KB	-rw-r--r--
test_states.py	2.22 KB	-rw-r--r--
test_states.pyc	4.4 KB	-rw-r--r--
test_states.pyo	4.4 KB	-rw-r--r--
test_supervisorctl.py	66.26 KB	-rw-r--r--
test_supervisorctl.pyc	87.29 KB	-rw-r--r--
test_supervisorctl.pyo	87.29 KB	-rw-r--r--
test_supervisord.py	29.39 KB	-rw-r--r--
test_supervisord.pyc	27.4 KB	-rw-r--r--
test_supervisord.pyo	27.4 KB	-rw-r--r--
test_web.py	6.67 KB	-rw-r--r--
test_web.pyc	9.52 KB	-rw-r--r--
test_web.pyo	9.52 KB	-rw-r--r--
test_xmlrpc.py	34.05 KB	-rw-r--r--
test_xmlrpc.pyc	47.64 KB	-rw-r--r--
test_xmlrpc.pyo	47.64 KB	-rw-r--r--

Code Editor : test_supervisorctl.py

import sys
import unittest
import xmlrpclib
from StringIO import StringIO

from supervisor.tests.base import DummyRPCServer

class ControllerTests(unittest.TestCase):
    def _getTargetClass(self):
        from supervisor.supervisorctl import Controller
        return Controller

def _makeOne(self, options):
        return self._getTargetClass()(options)

def test_ctor(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        self.assertEqual(controller.prompt, options.prompt + '> ')

def test__upcheck(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        result = controller.upcheck()
        self.assertEqual(result, True)

def test__upcheck_wrong_server_version(self):
        options = DummyClientOptions()
        options._server.supervisor.getVersion = lambda *x: '1.0'
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        result = controller.upcheck()
        self.assertEqual(result, False)
        self.assertEqual(controller.stdout.getvalue(),
                         'Sorry, this version of supervisorctl expects'
                         ' to talk to a server with API version 3.0, but'
                         ' the remote version is 1.0.\n')

def test__upcheck_unknown_method(self):
        options = DummyClientOptions()
        from xmlrpclib import Fault
        from supervisor.xmlrpc import Faults
        def getVersion():
            raise Fault(Faults.UNKNOWN_METHOD, 'duh')
        options._server.supervisor.getVersion = getVersion
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        result = controller.upcheck()
        self.assertEqual(result, False)
        self.assertEqual(controller.stdout.getvalue(),
                         'Sorry, supervisord responded but did not recognize'
                         ' the supervisor namespace commands that'
                         ' supervisorctl uses to control it.  Please check'
                         ' that the [rpcinterface:supervisor] section is'
                         ' enabled in the configuration file'
                         ' (see sample.conf).\n')

def test__upcheck_reraises_other_xmlrpc_faults(self):
        options = DummyClientOptions()
        from supervisor.xmlrpc import Faults
        def f(*arg, **kw):
            raise xmlrpclib.Fault(Faults.FAILED, '')
        options._server.supervisor.getVersion = f
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        self.assertRaises(xmlrpclib.Fault, controller.upcheck)

def test__upcheck_catches_socket_error_ECONNREFUSED(self):
        options = DummyClientOptions()
        import socket
        import errno
        def raise_fault(*arg, **kw):
            raise socket.error(errno.ECONNREFUSED, 'nobody home')
        options._server.supervisor.getVersion = raise_fault

controller = self._makeOne(options)
        controller.stdout = StringIO()

result = controller.upcheck()
        self.assertEqual(result, False)

output = controller.stdout.getvalue()
        self.assertTrue('refused connection' in output)

def test__upcheck_catches_socket_error_ENOENT(self):
        options = DummyClientOptions()
        import socket
        import errno
        def raise_fault(*arg, **kw):
            raise socket.error(errno.ENOENT, 'nobody home')
        options._server.supervisor.getVersion = raise_fault

controller = self._makeOne(options)
        controller.stdout = StringIO()

result = controller.upcheck()
        self.assertEqual(result, False)

output = controller.stdout.getvalue()
        self.assertTrue('no such file' in output)

def test__upcheck_reraises_other_socket_faults(self):
        options = DummyClientOptions()
        import socket
        import errno
        def f(*arg, **kw):
            raise socket.error(errno.EBADF, '')
        options._server.supervisor.getVersion = f
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        self.assertRaises(socket.error, controller.upcheck)

def test_onecmd(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        plugin = DummyPlugin()
        controller.options.plugins = (plugin,)
        result = controller.onecmd('help')
        self.assertEqual(result, None)
        self.assertEqual(plugin.helped, True)

def test_onecmd_multi_colonseparated(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        plugin = DummyPlugin()
        controller.options.plugins = (plugin,)
        result = controller.onecmd('help; help')
        self.assertEqual(result, None)
        self.assertEqual(controller.cmdqueue, [' help'])
        self.assertEqual(plugin.helped, True)

def test_onecmd_empty_does_not_repeat_previous_cmd(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        plugin = DummyPlugin()
        controller.options.plugins = (plugin,)
        plugin.helped = False
        controller.onecmd('help')
        self.assertTrue(plugin.helped)
        plugin.helped = False
        controller.onecmd('')
        self.assertFalse(plugin.helped)

def test_onecmd_clears_completion_cache(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        controller._complete_info = {}
        controller.onecmd('help')
        self.assertEqual(controller._complete_info, None)

def test_complete_action_empty(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help']
        result = controller.complete('', 0, line='')
        self.assertEqual(result, 'help ')
        result = controller.complete('', 1, line='')
        self.assertEqual(result, None)

def test_complete_action_partial(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help']
        result = controller.complete('h', 0, line='h')
        self.assertEqual(result, 'help ')
        result = controller.complete('h', 1, line='h')
        self.assertEqual(result, None)

def test_complete_action_whole(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help']
        result = controller.complete('help', 0, line='help')
        self.assertEqual(result, 'help ')

def test_complete_unknown_action_uncompletable(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        result = controller.complete('bad', 0, line='bad')
        self.assertEqual(result, None)

def test_complete_unknown_action_arg_uncompletable(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'add']
        result = controller.complete('', 1, line='bad ')
        self.assertEqual(result, None)

def test_complete_help_empty(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('', 0, line='help ')
        self.assertEqual(result, 'help ')
        result = controller.complete('', 1, line='help ')
        self.assertEqual(result, 'start ')
        result = controller.complete('', 2, line='help ')
        self.assertEqual(result, None)

def test_complete_help_action(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('he', 0, line='help he')
        self.assertEqual(result, 'help ')
        result = controller.complete('he', 1, line='help he')
        self.assertEqual(result, None)

def test_complete_start_empty(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('', 0, line='start ')
        self.assertEqual(result, 'foo ')
        result = controller.complete('', 1, line='start ')
        self.assertEqual(result, 'bar ')
        result = controller.complete('', 2, line='start ')
        self.assertEqual(result, 'baz:baz_01 ')
        result = controller.complete('', 3, line='start ')
        self.assertEqual(result, 'baz:* ')
        result = controller.complete('', 4, line='start ')
        self.assertEqual(result, None)

def test_complete_start_no_colon(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('f', 0, line='start f')
        self.assertEqual(result, 'foo ')
        result = controller.complete('f', 1, line='start f')
        self.assertEqual(result, None)

def test_complete_start_with_colon(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('foo:', 0, line='start foo:')
        self.assertEqual(result, 'foo:foo ')
        result = controller.complete('foo:', 1, line='start foo:')
        self.assertEqual(result, 'foo:* ')
        result = controller.complete('foo:', 2, line='start foo:')
        self.assertEqual(result, None)

def test_complete_start_uncompletable(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('bad', 0, line='start bad')
        self.assertEqual(result, None)

def test_complete_caches_process_info(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('', 0, line='start ')
        self.assertNotEqual(result, None)
        def f(*arg, **kw):
            raise Exception("should not have called getAllProcessInfo")
        controller.options._server.supervisor.getAllProcessInfo = f
        controller.complete('', 1, line='start ')

def test_complete_add_empty(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'add']
        result = controller.complete('', 0, line='add ')
        self.assertEqual(result, 'foo ')
        result = controller.complete('', 1, line='add ')
        self.assertEqual(result, 'bar ')
        result = controller.complete('', 2, line='add ')
        self.assertEqual(result, 'baz ')
        result = controller.complete('', 3, line='add ')
        self.assertEqual(result, None)

def test_complete_add_uncompletable(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'add']
        result = controller.complete('bad', 0, line='add bad')
        self.assertEqual(result, None)

def test_complete_add_group(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'add']
        result = controller.complete('f', 0, line='add f')
        self.assertEqual(result, 'foo ')
        result = controller.complete('f', 1, line='add f')
        self.assertEqual(result, None)

def test_complete_reload_arg_uncompletable(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'reload']
        result = controller.complete('', 1, line='reload ')
        self.assertEqual(result, None)

def test_complete_semicolon_separated_commands(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout=StringIO()
        controller.vocab = ['help', 'start']
        result = controller.complete('f', 0, line='help;start f')
        self.assertEqual(result, 'foo ')
        result = controller.complete('f', 1, line='help;start f')
        self.assertEqual(result, None)

def test_nohelp(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        self.assertEqual(controller.nohelp, '*** No help on %s')

def test_do_help(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        results = controller.do_help('')
        helpval = controller.stdout.getvalue()
        self.assertEqual(results, None)
        self.assertEqual(helpval, 'foo helped')

def test_do_help_for_help(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)
        controller.stdout = StringIO()
        results = controller.do_help("help")
        self.assertEqual(results, None)
        helpval = controller.stdout.getvalue()
        self.assertTrue("help\t\tPrint a list" in helpval)

def test_get_supervisor_returns_serverproxy_supervisor_namespace(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)

proxy = controller.get_supervisor()
        expected = options.getServerProxy().supervisor
        self.assertEqual(proxy, expected)

def test_get_server_proxy_with_no_args_returns_serverproxy(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)

proxy = controller.get_server_proxy()
        expected = options.getServerProxy()
        self.assertEqual(proxy, expected)

def test_get_server_proxy_with_namespace_returns_that_namespace(self):
        options = DummyClientOptions()
        controller = self._makeOne(options)

proxy = controller.get_server_proxy('system')
        expected = options.getServerProxy().system
        self.assertEqual(proxy, expected)

def test_real_controller_initialization(self):
        from supervisor.options import ClientOptions
        args = [] # simulating starting without parameters
        options = ClientOptions()

# No default config file search in case they would exist
        self.assertTrue(len(options.searchpaths) > 0)
        options.searchpaths = []

options.realize(args, doc=__doc__)
        self._makeOne(options) # should not raise

class TestControllerPluginBase(unittest.TestCase):
    def _getTargetClass(self):
        from supervisor.supervisorctl import ControllerPluginBase
        return ControllerPluginBase

def _makeOne(self, *arg, **kw):
        klass = self._getTargetClass()
        options = DummyClientOptions()
        ctl = DummyController(options)
        plugin = klass(ctl, *arg, **kw)
        return plugin

def test_do_help_noarg(self):
        plugin = self._makeOne()
        result = plugin.do_help(None)
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), '\n')
        self.assertEqual(len(plugin.ctl.topics_printed), 1)
        topics = plugin.ctl.topics_printed[0]
        self.assertEqual(topics[0], 'unnamed commands (type help <topic>):')
        self.assertEqual(topics[1], [])
        self.assertEqual(topics[2], 15)
        self.assertEqual(topics[3], 80)

def test_do_help_witharg(self):
        plugin = self._makeOne()
        result = plugin.do_help('foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), 'no help on foo\n')
        self.assertEqual(len(plugin.ctl.topics_printed), 0)

class TestDefaultControllerPlugin(unittest.TestCase):

def _getTargetClass(self):
        from supervisor.supervisorctl import DefaultControllerPlugin
        return DefaultControllerPlugin

def test_tail_toofewargs(self):
        plugin = self._makeOne()
        result = plugin.do_tail('')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(lines[0], 'Error: too few arguments')

def test_tail_toomanyargs(self):
        plugin = self._makeOne()
        result = plugin.do_tail('one two three four')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(lines[0], 'Error: too many arguments')

def test_tail_f_noprocname(self):
        plugin = self._makeOne()
        result = plugin.do_tail('-f')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(lines[0], 'Error: tail requires process name')

def test_tail_bad_modifier(self):
        plugin = self._makeOne()
        result = plugin.do_tail('-z foo')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(lines[0], 'Error: bad argument -z')

def test_tail_defaults(self):
        plugin = self._makeOne()
        result = plugin.do_tail('foo')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 12)
        self.assertEqual(lines[0], 'output line')

def test_tail_no_file(self):
        plugin = self._makeOne()
        result = plugin.do_tail('NO_FILE')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 2)
        self.assertEqual(lines[0], 'NO_FILE: ERROR (no log file)')

def test_tail_failed(self):
        plugin = self._makeOne()
        result = plugin.do_tail('FAILED')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 2)
        self.assertEqual(lines[0], 'FAILED: ERROR (unknown error reading log)')

def test_tail_bad_name(self):
        plugin = self._makeOne()
        result = plugin.do_tail('BAD_NAME')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 2)
        self.assertEqual(lines[0], 'BAD_NAME: ERROR (no such process name)')

def test_tail_bytesmodifier(self):
        plugin = self._makeOne()
        result = plugin.do_tail('-10 foo')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 3)
        self.assertEqual(lines[0], 'tput line')

def test_tail_explicit_channel_stdout_nomodifier(self):
        plugin = self._makeOne()
        result = plugin.do_tail('foo stdout')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 12)
        self.assertEqual(lines[0], 'output line')

def test_tail_explicit_channel_stderr_nomodifier(self):
        plugin = self._makeOne()
        result = plugin.do_tail('foo stderr')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 12)
        self.assertEqual(lines[0], 'output line')

def test_tail_explicit_channel_unrecognized(self):
        plugin = self._makeOne()
        result = plugin.do_tail('foo fudge')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().strip()
        self.assertEqual(value, "Error: bad channel 'fudge'")

def test_status_help(self):
        plugin = self._makeOne()
        plugin.help_status()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("status <name>" in out)

def test_status_table_process_column_min_width(self):
        plugin = self._makeOne()
        result = plugin.do_status('')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split("\n")
        self.assertEqual(lines[0].index("RUNNING"), 33)

def test_status_table_process_column_expands(self):
        plugin = self._makeOne()
        options = plugin.ctl.options
        def f(*arg, **kw):
            from supervisor.states import ProcessStates
            return [{'name': 'foo'*50, # long name
                     'group':'foo',
                     'pid': 11,
                     'state': ProcessStates.RUNNING,
                     'statename': 'RUNNING',
                     'start': 0,
                     'stop': 0,
                     'spawnerr': '',
                     'now': 0,
                     'description':'foo description'},
                    {
                    'name': 'bar', # short name
                    'group': 'bar',
                    'pid': 12,
                    'state': ProcessStates.FATAL,
                    'statename': 'RUNNING',
                    'start': 0,
                    'stop': 0,
                    'spawnerr': '',
                    'now': 0,
                    'description': 'bar description',
                    }]
        options._server.supervisor.getAllProcessInfo = f
        self.assertEqual(plugin.do_status(''), None)
        lines = plugin.ctl.stdout.getvalue().split("\n")
        self.assertEqual(lines[0].index("RUNNING"), 157)
        self.assertEqual(lines[1].index("RUNNING"), 157)

def test_status_all_processes_no_arg(self):
        plugin = self._makeOne()
        result = plugin.do_status('')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0].split(None, 2),
                         ['foo', 'RUNNING', 'foo description'])
        self.assertEqual(value[1].split(None, 2),
                         ['bar', 'FATAL', 'bar description'])
        self.assertEqual(value[2].split(None, 2),
                         ['baz:baz_01', 'STOPPED', 'baz description'])

def test_status_all_processes_all_arg(self):
        plugin = self._makeOne()
        result = plugin.do_status('all')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0].split(None, 2),
                         ['foo', 'RUNNING', 'foo description'])
        self.assertEqual(value[1].split(None, 2),
                         ['bar', 'FATAL', 'bar description'])
        self.assertEqual(value[2].split(None, 2),
                         ['baz:baz_01', 'STOPPED', 'baz description'])

def test_status_process_name(self):
        plugin = self._makeOne()
        result = plugin.do_status('foo')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().strip()
        self.assertEqual(value.split(None, 2),
                         ['foo', 'RUNNING', 'foo description'])

def test_status_group_name(self):
        plugin = self._makeOne()
        result = plugin.do_status('baz:*')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0].split(None, 2),
                         ['baz:baz_01', 'STOPPED', 'baz description'])

def test_status_mixed_names(self):
        plugin = self._makeOne()
        result = plugin.do_status('foo baz:*')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0].split(None, 2),
                         ['foo', 'RUNNING', 'foo description'])
        self.assertEqual(value[1].split(None, 2),
                         ['baz:baz_01', 'STOPPED', 'baz description'])

def test_status_bad_group_name(self):
        plugin = self._makeOne()
        result = plugin.do_status('badgroup:*')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0], "badgroup: ERROR (no such group)")

def test_status_bad_process_name(self):
        plugin = self._makeOne()
        result = plugin.do_status('badprocess')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0], "badprocess: ERROR (no such process)")

def test_status_bad_process_name_with_group(self):
        plugin = self._makeOne()
        result = plugin.do_status('badgroup:badprocess')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0], "badgroup:badprocess: "
                                   "ERROR (no such process)")

def test_start_help(self):
        plugin = self._makeOne()
        plugin.help_start()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("start <name>" in out)

def test_start_fail(self):
        plugin = self._makeOne()
        result = plugin.do_start('')
        self.assertEqual(result, None)
        expected = "Error: start requires a process name"
        self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], expected)

def test_start_badname(self):
        plugin = self._makeOne()
        result = plugin.do_start('BAD_NAME')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such process)\n')

def test_start_no_file(self):
        plugin = self._makeOne()
        result = plugin.do_start('NO_FILE')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'NO_FILE: ERROR (no such file)\n')

def test_start_not_executable(self):
        plugin = self._makeOne()
        result = plugin.do_start('NOT_EXECUTABLE')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'NOT_EXECUTABLE: ERROR (file is not executable)\n')

def test_start_alreadystarted(self):
        plugin = self._makeOne()
        result = plugin.do_start('ALREADY_STARTED')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ALREADY_STARTED: ERROR (already started)\n')

def test_start_spawnerror(self):
        plugin = self._makeOne()
        result = plugin.do_start('SPAWN_ERROR')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'SPAWN_ERROR: ERROR (spawn error)\n')

def test_start_abnormaltermination(self):
        plugin = self._makeOne()
        result = plugin.do_start('ABNORMAL_TERMINATION')
        self.assertEqual(result, None)
        expected = 'ABNORMAL_TERMINATION: ERROR (abnormal termination)\n'
        self.assertEqual(plugin.ctl.stdout.getvalue(), expected)

def test_start_one_success(self):
        plugin = self._makeOne()
        result = plugin.do_start('foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: started\n')

def test_start_one_with_group_name_success(self):
        plugin = self._makeOne()
        result = plugin.do_start('foo:foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: started\n')

def test_start_many(self):
        plugin = self._makeOne()
        result = plugin.do_start('foo bar')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: started\nbar: started\n')

def test_start_group(self):
        plugin = self._makeOne()
        result = plugin.do_start('foo:')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo:foo_00: started\n'
                         'foo:foo_01: started\n')

def test_start_group_bad_name(self):
        plugin = self._makeOne()
        result = plugin.do_start('BAD_NAME:')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such group)\n')

def test_start_all(self):
        plugin = self._makeOne()
        result = plugin.do_start('all')
        self.assertEqual(result, None)

self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: started\n'
                         'foo2: started\n'
                         'failed_group:failed: ERROR (spawn error)\n')

def test_stop_help(self):
        plugin = self._makeOne()
        plugin.help_stop()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("stop <name>" in out)

def test_stop_fail(self):
        plugin = self._makeOne()
        result = plugin.do_stop('')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0],
                         "Error: stop requires a process name")

def test_stop_badname(self):
        plugin = self._makeOne()
        result = plugin.do_stop('BAD_NAME')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such process)\n')

def test_stop_notrunning(self):
        plugin = self._makeOne()
        result = plugin.do_stop('NOT_RUNNING')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'NOT_RUNNING: ERROR (not running)\n')

def test_stop_failed(self):
        plugin = self._makeOne()
        result = plugin.do_stop('FAILED')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), 'FAILED\n')

def test_stop_one_success(self):
        plugin = self._makeOne()
        result = plugin.do_stop('foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: stopped\n')

def test_stop_one_with_group_name_success(self):
        plugin = self._makeOne()
        result = plugin.do_stop('foo:foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: stopped\n')

def test_stop_many(self):
        plugin = self._makeOne()
        result = plugin.do_stop('foo bar')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: stopped\n'
                         'bar: stopped\n')

def test_stop_group(self):
        plugin = self._makeOne()
        result = plugin.do_stop('foo:')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo:foo_00: stopped\n'
                         'foo:foo_01: stopped\n')

def test_stop_group_bad_name(self):
        plugin = self._makeOne()
        result = plugin.do_stop('BAD_NAME:')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such group)\n')

def test_stop_all(self):
        plugin = self._makeOne()
        result = plugin.do_stop('all')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: stopped\n'
                         'foo2: stopped\n'
                         'failed_group:failed: ERROR (no such process)\n')

def test_stop_upcheck_failed(self):
        plugin = self._makeOne()
        plugin.ctl.upcheck = lambda: False
        called = []
        def f(*arg, **kw):
            called.append(True)
        supervisor = plugin.ctl.options._server.supervisor
        supervisor.stopAllProcesses = f
        supervisor.stopProcessGroup = f
        plugin.do_stop('foo')
        self.assertEqual(called, [])

def test_signal_help(self):
        plugin = self._makeOne()
        plugin.help_signal()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("signal <signal name> <name>" in out)

def test_signal_fail_no_arg(self):
        plugin = self._makeOne()
        result = plugin.do_signal('')
        self.assertEqual(result, None)
        msg = 'Error: signal requires a signal name and a process name'
        self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], msg)

def test_signal_fail_one_arg(self):
        plugin = self._makeOne()
        result = plugin.do_signal('hup')
        self.assertEqual(result, None)
        msg = 'Error: signal requires a signal name and a process name'
        self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0], msg)

def test_signal_bad_signal(self):
        plugin = self._makeOne()
        result = plugin.do_signal('BAD_SIGNAL foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: ERROR (bad signal name)\n')

def test_signal_bad_name(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP BAD_NAME')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such process)\n')

def test_signal_bad_group(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP BAD_NAME:')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such group)\n')

def test_signal_not_running(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP NOT_RUNNING')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'NOT_RUNNING: ERROR (not running)\n')

def test_signal_failed(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP FAILED')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), 'FAILED\n')

def test_signal_one_success(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), 'foo: signalled\n')

def test_signal_many(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP foo bar')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: signalled\n'
                         'bar: signalled\n')

def test_signal_group(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP foo:')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo:foo_00: signalled\n'
                         'foo:foo_01: signalled\n')

def test_signal_all(self):
        plugin = self._makeOne()
        result = plugin.do_signal('HUP all')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: signalled\n'
                         'foo2: signalled\n'
                         'failed_group:failed: ERROR (no such process)\n')

def test_signal_upcheck_failed(self):
        plugin = self._makeOne()
        plugin.ctl.upcheck = lambda: False
        called = []
        def f(*arg, **kw):
            called.append(True)
        supervisor = plugin.ctl.options._server.supervisor
        supervisor.signalAllProcesses = f
        supervisor.signalProcessGroup = f
        plugin.do_signal('term foo')
        self.assertEqual(called, [])

def test_restart_help(self):
        plugin = self._makeOne()
        plugin.help_restart()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("restart <name>" in out)

def test_restart_fail(self):
        plugin = self._makeOne()
        result = plugin.do_restart('')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0],
                         'Error: restart requires a process name')

def test_restart_one(self):
        plugin = self._makeOne()
        result = plugin.do_restart('foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: stopped\nfoo: started\n')

def test_restart_all(self):
        plugin = self._makeOne()
        result = plugin.do_restart('all')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: stopped\nfoo2: stopped\n'
                         'failed_group:failed: ERROR (no such process)\n'
                         'foo: started\nfoo2: started\n'
                         'failed_group:failed: ERROR (spawn error)\n')

def test_clear_help(self):
        plugin = self._makeOne()
        plugin.help_clear()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("clear <name>" in out)

def test_clear_fail(self):
        plugin = self._makeOne()
        result = plugin.do_clear('')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue().split('\n')[0],
                         "Error: clear requires a process name")

def test_clear_badname(self):
        plugin = self._makeOne()
        result = plugin.do_clear('BAD_NAME')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'BAD_NAME: ERROR (no such process)\n')

def test_clear_one_success(self):
        plugin = self._makeOne()
        result = plugin.do_clear('foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: cleared\n')

def test_clear_one_with_group_success(self):
        plugin = self._makeOne()
        result = plugin.do_clear('foo:foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: cleared\n')

def test_clear_many(self):
        plugin = self._makeOne()
        result = plugin.do_clear('foo bar')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: cleared\nbar: cleared\n')

def test_clear_all(self):
        plugin = self._makeOne()
        result = plugin.do_clear('all')
        self.assertEqual(result, None)

self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'foo: cleared\n'
                         'foo2: cleared\n'
                         'failed_group:failed: ERROR (failed)\n')

def test_open_help(self):
        plugin = self._makeOne()
        plugin.help_open()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("open <url>" in out)

def test_open_fail(self):
        plugin = self._makeOne()
        result = plugin.do_open('badname')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: url must be http:// or unix://\n')

def test_open_succeed(self):
        plugin = self._makeOne()
        result = plugin.do_open('http://localhost:9002')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(value[0].split(None, 2),
                         ['foo', 'RUNNING', 'foo description'])
        self.assertEqual(value[1].split(None, 2),
                         ['bar', 'FATAL', 'bar description'])
        self.assertEqual(value[2].split(None, 2),
                         ['baz:baz_01', 'STOPPED', 'baz description'])

def test_version_help(self):
        plugin = self._makeOne()
        plugin.help_version()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Show the version of the remote supervisord" in out)

def test_version(self):
        plugin = self._makeOne()
        plugin.do_version(None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), '3000\n')

def test_version_arg(self):
        plugin = self._makeOne()
        result = plugin.do_version('bad')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: version accepts no arguments'), val)

def test_version_upcheck_failed(self):
        plugin = self._makeOne()
        plugin.ctl.upcheck = lambda: False
        called = []
        def f(*arg, **kw):
            called.append(True)
        plugin.ctl.options._server.supervisor.getSupervisorVersion = f
        plugin.do_version('')
        self.assertEqual(called, [])

def test_reload_help(self):
        plugin = self._makeOne()
        plugin.help_reload()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Restart the remote supervisord" in out)

def test_reload_fail(self):
        plugin = self._makeOne()
        options = plugin.ctl.options
        options._server.supervisor._restartable = False
        result = plugin.do_reload('')
        self.assertEqual(result, None)
        self.assertEqual(options._server.supervisor._restarted, False)

def test_reload(self):
        plugin = self._makeOne()
        options = plugin.ctl.options
        result = plugin.do_reload('')
        self.assertEqual(result, None)
        self.assertEqual(options._server.supervisor._restarted, True)

def test_reload_arg(self):
        plugin = self._makeOne()
        result = plugin.do_reload('bad')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: reload accepts no arguments'), val)

def test_shutdown_help(self):
        plugin = self._makeOne()
        plugin.help_shutdown()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Shut the remote supervisord down" in out)

def test_shutdown(self):
        plugin = self._makeOne()
        options = plugin.ctl.options
        result = plugin.do_shutdown('')
        self.assertEqual(result, None)
        self.assertEqual(options._server.supervisor._shutdown, True)

def test_shutdown_catches_xmlrpc_fault_shutdown_state(self):
        plugin = self._makeOne()
        from supervisor import xmlrpc
        import xmlrpclib

def raise_fault(*arg, **kw):
            raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, 'bye')
        plugin.ctl.options._server.supervisor.shutdown = raise_fault

result = plugin.do_shutdown('')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: already shutting down\n')

def test_shutdown_reraises_other_xmlrpc_faults(self):
        plugin = self._makeOne()
        from supervisor import xmlrpc
        import xmlrpclib

def raise_fault(*arg, **kw):
            raise xmlrpclib.Fault(xmlrpc.Faults.CANT_REREAD, 'ouch')
        plugin.ctl.options._server.supervisor.shutdown = raise_fault

self.assertRaises(xmlrpclib.Fault,
                          plugin.do_shutdown, '')

def test_shutdown_catches_socket_error_ECONNREFUSED(self):
        plugin = self._makeOne()
        import socket
        import errno

def raise_fault(*arg, **kw):
            raise socket.error(errno.ECONNREFUSED, 'nobody home')
        plugin.ctl.options._server.supervisor.shutdown = raise_fault

result = plugin.do_shutdown('')
        self.assertEqual(result, None)

output = plugin.ctl.stdout.getvalue()
        self.assertTrue('refused connection (already shut down?)' in output)

def test_shutdown_catches_socket_error_ENOENT(self):
        plugin = self._makeOne()
        import socket
        import errno

def raise_fault(*arg, **kw):
            raise socket.error(errno.ENOENT, 'no file')
        plugin.ctl.options._server.supervisor.shutdown = raise_fault

result = plugin.do_shutdown('')
        self.assertEqual(result, None)

output = plugin.ctl.stdout.getvalue()
        self.assertTrue('no such file (already shut down?)' in output)

def test_shutdown_reraises_other_socket_errors(self):
        plugin = self._makeOne()
        import socket
        import errno

def raise_fault(*arg, **kw):
            raise socket.error(errno.EPERM, 'denied')
        plugin.ctl.options._server.supervisor.shutdown = raise_fault

self.assertRaises(socket.error,
                          plugin.do_shutdown, '')

def test__formatChanges(self):
        plugin = self._makeOne()
        # Don't explode, plz
        plugin._formatChanges([['added'], ['changed'], ['removed']])
        plugin._formatChanges([[], [], []])

def test_reread_help(self):
        plugin = self._makeOne()
        plugin.help_reread()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Reload the daemon's configuration files" in out)

def test_reread(self):
        plugin = self._makeOne()
        calls = []
        plugin._formatChanges = lambda x: calls.append(x)
        result = plugin.do_reread(None)
        self.assertEqual(result, None)
        self.assertEqual(calls[0], [['added'], ['changed'], ['removed']])

def test_reread_arg(self):
        plugin = self._makeOne()
        result = plugin.do_reread('bad')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: reread accepts no arguments'), val)

def test_reread_cant_reread(self):
        plugin = self._makeOne()
        from supervisor import xmlrpc
        def reloadConfig(*arg, **kw):
            raise xmlrpclib.Fault(xmlrpc.Faults.CANT_REREAD, 'cant')
        plugin.ctl.options._server.supervisor.reloadConfig = reloadConfig
        plugin.do_reread(None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: cant\n')

def test_reread_shutdown_state(self):
        plugin = self._makeOne()
        from supervisor import xmlrpc
        def reloadConfig(*arg, **kw):
            raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, '')
        plugin.ctl.options._server.supervisor.reloadConfig = reloadConfig
        plugin.do_reread(None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: supervisor shutting down\n')

def test_reread_reraises_other_faults(self):
        plugin = self._makeOne()
        from supervisor import xmlrpc
        def reloadConfig(*arg, **kw):
            raise xmlrpclib.Fault(xmlrpc.Faults.FAILED, '')
        plugin.ctl.options._server.supervisor.reloadConfig = reloadConfig
        self.assertRaises(xmlrpclib.Fault, plugin.do_reread, '')

def test__formatConfigInfo(self):
        info = { 'group': 'group1',
                 'name': 'process1',
                 'inuse': True,
                 'autostart': True,
                 'process_prio': 999,
                 'group_prio': 999 }
        plugin = self._makeOne()
        result = plugin._formatConfigInfo(info)
        self.assertTrue('in use' in result)
        info = { 'group': 'group1',
                 'name': 'process1',
                 'inuse': False,
                 'autostart': False,
                 'process_prio': 999,
                 'group_prio': 999 }
        result = plugin._formatConfigInfo(info)
        self.assertTrue('avail' in result)

def test_avail_help(self):
        plugin = self._makeOne()
        plugin.help_avail()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Display all configured" in out)

def test_avail(self):
        calls = []
        plugin = self._makeOne()

class FakeSupervisor(object):
            def getAllConfigInfo(self):
                return [{ 'group': 'group1', 'name': 'process1',
                          'inuse': False, 'autostart': False,
                          'process_prio': 999, 'group_prio': 999 }]

plugin.ctl.get_supervisor = lambda : FakeSupervisor()
        plugin.ctl.output = calls.append
        result = plugin.do_avail('')
        self.assertEqual(result, None)

def test_avail_arg(self):
        plugin = self._makeOne()
        result = plugin.do_avail('bad')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: avail accepts no arguments'), val)

def test_avail_shutdown_state(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

def getAllConfigInfo():
            from supervisor import xmlrpc
            raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, '')
        supervisor.getAllConfigInfo = getAllConfigInfo

result = plugin.do_avail('')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: supervisor shutting down\n')

def test_avail_reraises_other_faults(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

def getAllConfigInfo():
            from supervisor import xmlrpc
            raise xmlrpclib.Fault(xmlrpc.Faults.FAILED, '')
        supervisor.getAllConfigInfo = getAllConfigInfo

self.assertRaises(xmlrpclib.Fault, plugin.do_avail, '')

def test_add_help(self):
        plugin = self._makeOne()
        plugin.help_add()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("add <name>" in out)

def test_add(self):
        plugin = self._makeOne()
        result = plugin.do_add('foo')
        self.assertEqual(result, None)
        supervisor = plugin.ctl.options._server.supervisor
        self.assertEqual(supervisor.processes, ['foo'])

def test_add_already_added(self):
        plugin = self._makeOne()
        result = plugin.do_add('ALREADY_ADDED')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: process group already active\n')

def test_add_bad_name(self):
        plugin = self._makeOne()
        result = plugin.do_add('BAD_NAME')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: no such process/group: BAD_NAME\n')

def test_add_shutdown_state(self):
        plugin = self._makeOne()
        result = plugin.do_add('SHUTDOWN_STATE')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: shutting down\n')

def test_add_reraises_other_faults(self):
        plugin = self._makeOne()
        self.assertRaises(xmlrpclib.Fault, plugin.do_add, 'FAILED')

def test_remove_help(self):
        plugin = self._makeOne()
        plugin.help_remove()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("remove <name>" in out)

def test_remove(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor
        supervisor.processes = ['foo']
        result = plugin.do_remove('foo')
        self.assertEqual(result, None)
        self.assertEqual(supervisor.processes, [])

def test_remove_bad_name(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor
        supervisor.processes = ['foo']
        result = plugin.do_remove('BAD_NAME')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: no such process/group: BAD_NAME\n')

def test_remove_still_running(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor
        supervisor.processes = ['foo']
        result = plugin.do_remove('STILL_RUNNING')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'ERROR: process/group still running: STILL_RUNNING\n')

def test_remove_reraises_other_faults(self):
        plugin = self._makeOne()
        self.assertRaises(xmlrpclib.Fault, plugin.do_remove, 'FAILED')

def test_update_help(self):
        plugin = self._makeOne()
        plugin.help_update()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Reload config and add/remove" in out)

def test_update_not_on_shutdown(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor
        def reloadConfig():
            from supervisor import xmlrpc
            import xmlrpclib
            raise xmlrpclib.Fault(xmlrpc.Faults.SHUTDOWN_STATE, 'blah')
        supervisor.reloadConfig = reloadConfig
        supervisor.processes = ['removed']
        plugin.do_update('')
        self.assertEqual(supervisor.processes, ['removed'])

def test_update_added_procs(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

def reloadConfig():
            return [[['new_proc'], [], []]]
        supervisor.reloadConfig = reloadConfig

result = plugin.do_update('')
        self.assertEqual(result, None)
        self.assertEqual(supervisor.processes, ['new_proc'])

def test_update_with_gname(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

def reloadConfig():
            return [[['added1', 'added2'], ['changed'], ['removed']]]
        supervisor.reloadConfig = reloadConfig
        supervisor.processes = ['changed', 'removed']

plugin.do_update('changed')
        self.assertEqual(sorted(supervisor.processes),
                         sorted(['changed', 'removed']))

plugin.do_update('added1 added2')
        self.assertEqual(sorted(supervisor.processes),
                         sorted(['changed', 'removed', 'added1', 'added2']))

plugin.do_update('removed')
        self.assertEqual(sorted(supervisor.processes),
                         sorted(['changed', 'added1', 'added2']))

supervisor.processes = ['changed', 'removed']
        plugin.do_update('removed added1')
        self.assertEqual(sorted(supervisor.processes),
                         sorted(['changed', 'added1']))

supervisor.processes = ['changed', 'removed']
        plugin.do_update('all')
        self.assertEqual(sorted(supervisor.processes),
                         sorted(['changed', 'added1', 'added2']))

def test_update_changed_procs(self):
        from supervisor import xmlrpc

plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

calls = []
        def reloadConfig():
            return [[[], ['changed_group'], []]]
        supervisor.reloadConfig = reloadConfig
        supervisor.startProcess = lambda x: calls.append(('start', x))

supervisor.addProcessGroup('changed_group') # fake existence
        results = [{'name':        'changed_process',
                    'group':       'changed_group',
                    'status':      xmlrpc.Faults.SUCCESS,
                    'description': 'blah'}]
        def stopProcessGroup(name):
            calls.append(('stop', name))
            return results
        supervisor.stopProcessGroup = stopProcessGroup

plugin.do_update('')
        self.assertEqual(calls, [('stop', 'changed_group')])

supervisor.addProcessGroup('changed_group') # fake existence
        calls[:] = []
        results[:] = [{'name':        'changed_process1',
                       'group':       'changed_group',
                       'status':      xmlrpc.Faults.NOT_RUNNING,
                       'description': 'blah'},
                      {'name':        'changed_process2',
                       'group':       'changed_group',
                       'status':      xmlrpc.Faults.FAILED,
                       'description': 'blah'}]

plugin.do_update('')
        self.assertEqual(calls, [('stop', 'changed_group')])

supervisor.addProcessGroup('changed_group') # fake existence
        calls[:] = []
        results[:] = [{'name':        'changed_process1',
                       'group':       'changed_group',
                       'status':      xmlrpc.Faults.FAILED,
                       'description': 'blah'},
                      {'name':        'changed_process2',
                       'group':       'changed_group',
                       'status':      xmlrpc.Faults.SUCCESS,
                       'description': 'blah'}]

plugin.do_update('')
        self.assertEqual(calls, [('stop', 'changed_group')])

def test_update_removed_procs(self):
        from supervisor import xmlrpc

plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

def reloadConfig():
            return [[[], [], ['removed_group']]]
        supervisor.reloadConfig = reloadConfig

results = [{'name':        'removed_process',
                    'group':       'removed_group',
                    'status':      xmlrpc.Faults.SUCCESS,
                    'description': 'blah'}]
        supervisor.processes = ['removed_group']

def stopProcessGroup(name):
            return results
        supervisor.stopProcessGroup = stopProcessGroup

plugin.do_update('')
        self.assertEqual(supervisor.processes, [])

results[:] = [{'name':        'removed_process',
                       'group':       'removed_group',
                       'status':      xmlrpc.Faults.NOT_RUNNING,
                       'description': 'blah'}]
        supervisor.processes = ['removed_group']

plugin.do_update('')
        self.assertEqual(supervisor.processes, [])

results[:] = [{'name':        'removed_process',
                       'group':       'removed_group',
                       'status':      xmlrpc.Faults.FAILED,
                       'description': 'blah'}]
        supervisor.processes = ['removed_group']

plugin.do_update('')
        self.assertEqual(supervisor.processes, ['removed_group'])

def test_update_reraises_other_faults(self):
        plugin = self._makeOne()
        supervisor = plugin.ctl.options._server.supervisor

def reloadConfig():
            from supervisor import xmlrpc
            raise xmlrpclib.Fault(xmlrpc.Faults.FAILED, 'FAILED')
        supervisor.reloadConfig = reloadConfig

self.assertRaises(xmlrpclib.Fault, plugin.do_update, '')

def test_pid_help(self):
        plugin = self._makeOne()
        plugin.help_pid()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("pid <name>" in out)

def test_pid_supervisord(self):
        plugin = self._makeOne()
        result = plugin.do_pid('')
        self.assertEqual(result, None)
        options = plugin.ctl.options
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(len(lines), 2)
        self.assertEqual(lines[0], str(options._server.supervisor.getPID()))

def test_pid_allprocesses(self):
        plugin = self._makeOne()
        result = plugin.do_pid('all')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().strip()
        self.assertEqual(value.split(), ['11', '12', '13'])

def test_pid_badname(self):
        plugin = self._makeOne()
        result = plugin.do_pid('BAD_NAME')
        self.assertEqual(result, None)
        value = plugin.ctl.stdout.getvalue().strip()
        self.assertEqual(value, 'No such process BAD_NAME')

def test_pid_oneprocess(self):
        plugin = self._makeOne()
        result = plugin.do_pid('foo')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue().strip(), '11')

def test_maintail_help(self):
        plugin = self._makeOne()
        plugin.help_maintail()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("tail of supervisor main log file" in out)

def test_maintail_toomanyargs(self):
        plugin = self._makeOne()
        result = plugin.do_maintail('foo bar')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: too many'), val)

def test_maintail_minus_string_fails(self):
        plugin = self._makeOne()
        result = plugin.do_maintail('-wrong')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: bad argument -wrong'), val)

def test_maintail_wrong(self):
        plugin = self._makeOne()
        result = plugin.do_maintail('wrong')
        self.assertEqual(result, None)
        val = plugin.ctl.stdout.getvalue()
        self.assertTrue(val.startswith('Error: bad argument wrong'), val)

def _dont_test_maintail_dashf(self):
        # https://github.com/Supervisor/supervisor/issues/285
        # TODO: Refactor so we can test more of maintail -f than just a
        # connect error, and fix this test so it passes on FreeBSD.
        plugin = self._makeOne()
        plugin.listener = DummyListener()
        result = plugin.do_maintail('-f')
        self.assertEqual(result, None)
        errors = plugin.listener.errors
        self.assertEqual(len(errors), 1)
        error = errors[0]
        self.assertEqual(plugin.listener.closed,
                         'http://localhost:65532/mainlogtail')
        self.assertEqual(error[0],
                         'http://localhost:65532/mainlogtail')
        for msg in ('Cannot connect', 'socket.error'):
            self.assertTrue(msg in error[1])

def test_maintail_bad_modifier(self):
        plugin = self._makeOne()
        result = plugin.do_maintail('-z')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(lines[0], 'Error: bad argument -z')

def test_maintail_nobytes(self):
        plugin = self._makeOne()
        result = plugin.do_maintail('')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), 'mainlogdata\n')

def test_maintail_dashbytes(self):
        plugin = self._makeOne()
        result = plugin.do_maintail('-100')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(), 'mainlogdata\n')

def test_maintail_readlog_error_nofile(self):
        plugin = self._makeOne()
        supervisor_rpc = plugin.ctl.get_supervisor()
        from supervisor import xmlrpc
        supervisor_rpc._readlog_error = xmlrpc.Faults.NO_FILE
        result = plugin.do_maintail('-100')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'supervisord: ERROR (no log file)\n')

def test_maintail_readlog_error_failed(self):
        plugin = self._makeOne()
        supervisor_rpc = plugin.ctl.get_supervisor()
        from supervisor import xmlrpc
        supervisor_rpc._readlog_error = xmlrpc.Faults.FAILED
        result = plugin.do_maintail('-100')
        self.assertEqual(result, None)
        self.assertEqual(plugin.ctl.stdout.getvalue(),
                         'supervisord: ERROR (unknown error reading log)\n')

def test_fg_help(self):
        plugin = self._makeOne()
        plugin.help_fg()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("fg <process>" in out)

def test_fg_too_few_args(self):
        plugin = self._makeOne()
        result = plugin.do_fg('')
        self.assertEqual(result, None)
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(lines[0], 'ERROR: no process name supplied')

def test_fg_too_many_args(self):
        plugin = self._makeOne()
        result = plugin.do_fg('foo bar')
        self.assertEqual(result, None)
        line = plugin.ctl.stdout.getvalue()
        self.assertEqual(line, 'ERROR: too many process names supplied\n')

def test_fg_badprocname(self):
        plugin = self._makeOne()
        result = plugin.do_fg('BAD_NAME')
        self.assertEqual(result, None)
        line = plugin.ctl.stdout.getvalue()
        self.assertEqual(line, 'ERROR: bad process name supplied\n')

def test_fg_procnotrunning(self):
        plugin = self._makeOne()
        result = plugin.do_fg('bar')
        self.assertEqual(result, None)
        line = plugin.ctl.stdout.getvalue()
        self.assertEqual(line, 'ERROR: process not running\n')
        result = plugin.do_fg('baz_01')
        lines = plugin.ctl.stdout.getvalue().split('\n')
        self.assertEqual(result, None)
        self.assertEqual(lines[-2], 'ERROR: process not running')

def test_fg_upcheck_failed(self):
        plugin = self._makeOne()
        plugin.ctl.upcheck = lambda: False
        called = []
        def f(*arg, **kw):
            called.append(True)
        plugin.ctl.options._server.supervisor.getProcessInfo = f
        plugin.do_fg('foo')
        self.assertEqual(called, [])

def test_exit_help(self):
        plugin = self._makeOne()
        plugin.help_exit()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Exit the supervisor shell" in out)

def test_quit_help(self):
        plugin = self._makeOne()
        plugin.help_quit()
        out = plugin.ctl.stdout.getvalue()
        self.assertTrue("Exit the supervisor shell" in out)

class DummyListener:
    def __init__(self):
        self.errors = []
    def error(self, url, msg):
        self.errors.append((url, msg))
    def close(self, url):
        self.closed = url

class DummyPluginFactory:
    def __init__(self, ctl, **kw):
        self.ctl = ctl

def do_help(self, arg):
        self.ctl.stdout.write('foo helped')

class DummyClientOptions:
    def __init__(self):
        self.prompt = 'supervisor'
        self.serverurl = 'http://localhost:65532'
        self.username = 'chrism'
        self.password = '123'
        self.history_file = None
        self.plugins = ()
        self._server = DummyRPCServer()
        self.interactive = False
        self.plugin_factories = [('dummy', DummyPluginFactory, {})]

def getServerProxy(self):
        return self._server

class DummyController:
    nohelp = 'no help on %s'
    def __init__(self, options):
        self.options = options
        self.topics_printed = []
        self.stdout = StringIO()

def upcheck(self):
        return True

def get_supervisor(self):
        return self.get_server_proxy('supervisor')

def get_server_proxy(self, namespace=None):
        proxy = self.options.getServerProxy()
        if namespace is None:
            return proxy
        else:
            return getattr(proxy, namespace)

def output(self, data):
        self.stdout.write(data + '\n')

def print_topics(self, doc_headers, cmds_doc, rows, cols):
        self.topics_printed.append((doc_headers, cmds_doc, rows, cols))

class DummyPlugin:
    def __init__(self, controller=None):
        self.ctl = controller

def do_help(self, arg):
        self.helped = True

def test_suite():
    return unittest.findTestCases(sys.modules[__name__])

if __name__ == '__main__':
    unittest.main(defaultTest='test_suite')

${this.title}

Code Editor : test_supervisorctl.py