[Supervisor-checkins] r805 - in supervisor/trunk/src/supervisor: . tests
Mike Naberezny
mike at maintainable.com
Mon Aug 18 12:57:03 EDT 2008
Author: Mike Naberezny <mike at maintainable.com>
Date: Mon Aug 18 12:57:02 2008
New Revision: 805
Log:
Minor coding standards cleanup.
Modified:
supervisor/trunk/src/supervisor/supervisorctl.py
supervisor/trunk/src/supervisor/tests/test_supervisorctl.py
Modified: supervisor/trunk/src/supervisor/supervisorctl.py
==============================================================================
--- supervisor/trunk/src/supervisor/supervisorctl.py (original)
+++ supervisor/trunk/src/supervisor/supervisorctl.py Mon Aug 18 12:57:02 2008
@@ -49,24 +49,24 @@
from supervisor import xmlrpc
class fgthread(threading.Thread):
-
- # A subclass of threading.Thread, with a kill() method.
- # To be used for foreground output/error streaming.
- # http://mail.python.org/pipermail/python-list/2004-May/260937.html
+ """ A subclass of threading.Thread, with a kill() method.
+ To be used for foreground output/error streaming.
+ http://mail.python.org/pipermail/python-list/2004-May/260937.html
+ """
def __init__(self, program, ctl):
threading.Thread.__init__(self)
import http_client
self.killed = False
- self.program=program
- self.ctl=ctl
- self.listener=http_client.Listener()
- self.output_handler=http_client.HTTPHandler(self.listener,
- self.ctl.options.username,
- self.ctl.options.password)
- self.error_handler=http_client.HTTPHandler(self.listener,
- self.ctl.options.username,
- self.ctl.options.password)
+ self.program = program
+ self.ctl = ctl
+ self.listener = http_client.Listener()
+ self.output_handler = http_client.HTTPHandler(self.listener,
+ self.ctl.options.username,
+ self.ctl.options.password)
+ self.error_handler = http_client.HTTPHandler(self.listener,
+ self.ctl.options.username,
+ self.ctl.options.password)
def start(self):
# Start the thread
@@ -262,19 +262,19 @@
import readline
except ImportError:
return None
- line=readline.get_line_buffer()
+ line = readline.get_line_buffer()
if line == '':
results = [i+' ' for i in self.vocab if i.startswith(text)]+[None]
return results[state]
else:
- exp=line.split()[0]
+ exp = line.split()[0]
if exp in ['start','stop','restart','clear','status','tail','fg']:
if not line.endswith(' ') and len(line.split()) == 1:
- return [text+' ',None][state]
+ return [text + ' ', None][state]
if exp == 'fg':
if line.endswith(' ') and len(line.split()) > 1:
return None
- results=self.completionmatches(text,line)+[None]
+ results = self.completionmatches(text,line)+[None]
return results[state]
elif exp in ['maintail','pid','reload','shutdown','exit','open',
'quit','version','EOF']:
@@ -905,14 +905,14 @@
self.ctl.output('Error: no process name supplied')
self.help_fg()
return
- args=args.split()
- if len(args)>1:
+ args = args.split()
+ if len(args) > 1:
self.ctl.output('Error: too many process names supplied')
return
- program=args[0]
- supervisor=self.ctl.get_supervisor()
+ program = args[0]
+ supervisor = self.ctl.get_supervisor()
try:
- info=supervisor.getProcessInfo(program)
+ info = supervisor.getProcessInfo(program)
except xmlrpclib.Fault, msg:
if msg.faultCode == xmlrpc.Faults.BAD_NAME:
self.ctl.output('Error: bad process name supplied')
@@ -925,7 +925,7 @@
return
# everything good; continue
try:
- a=fgthread(program,self.ctl)
+ a = fgthread(program,self.ctl)
# this thread takes care of
# the output/error messages
a.start()
Modified: supervisor/trunk/src/supervisor/tests/test_supervisorctl.py
==============================================================================
--- supervisor/trunk/src/supervisor/tests/test_supervisorctl.py (original)
+++ supervisor/trunk/src/supervisor/tests/test_supervisorctl.py Mon Aug 18 12:57:02 2008
@@ -590,37 +590,37 @@
self.assertEqual(plugin.ctl.stdout.getvalue(),
'supervisord: ERROR (unknown error reading log)\n')
- def test_fg_toofewargs(self):
- plugin=self._makeOne()
- result=plugin.do_fg('')
- lines=plugin.ctl.stdout.getvalue().split('\n')
- self.assertEqual(result,None)
- self.assertEqual(lines[0],'Error: no process name supplied')
-
- def test_fg_toomanyargs(self):
- plugin=self._makeOne()
- result=plugin.do_fg('foo bar')
- line=plugin.ctl.stdout.getvalue()
- self.assertEqual(result,None)
- self.assertEqual(line,'Error: too many process names supplied\n')
+ def test_fg_too_few_args(self):
+ plugin = self._makeOne()
+ result = plugin.do_fg('')
+ lines = plugin.ctl.stdout.getvalue().split('\n')
+ self.assertEqual(result, None)
+ 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')
+ line = plugin.ctl.stdout.getvalue()
+ self.assertEqual(result, None)
+ self.assertEqual(line, 'Error: too many process names supplied\n')
def test_fg_badprocname(self):
- plugin=self._makeOne()
- result=plugin.do_fg('BAD_NAME')
- line=plugin.ctl.stdout.getvalue()
- self.assertEqual(result,None)
- self.assertEqual(line,'Error: bad process name supplied\n')
+ plugin = self._makeOne()
+ result = plugin.do_fg('BAD_NAME')
+ line = plugin.ctl.stdout.getvalue()
+ self.assertEqual(result, None)
+ self.assertEqual(line, 'Error: bad process name supplied\n')
def test_fg_procnotrunning(self):
- plugin=self._makeOne()
- result=plugin.do_fg('bar')
- line=plugin.ctl.stdout.getvalue()
- self.assertEqual(result,None)
- 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')
+ plugin = self._makeOne()
+ result = plugin.do_fg('bar')
+ line = plugin.ctl.stdout.getvalue()
+ self.assertEqual(result, None)
+ 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')
class DummyListener:
def __init__(self):
More information about the Supervisor-checkins
mailing list