Add tests for signal command (based on a patch by Moriyoshi Koizumi)

supervisor/tests/base.py

@@ -917,13 +917,16 @@ class DummySupervisorRPCNamespace:
     def clearAllProcessLogs(self):
         from supervisor import xmlrpc
         return [
-            {'name':'foo', 'group':'foo',
+            {'name':'foo',
+             'group':'foo',
              'status':xmlrpc.Faults.SUCCESS,
              'description': 'OK'},
-            {'name':'foo2', 'group':'foo2',
+            {'name':'foo2',
+             'group':'foo2',
              'status':xmlrpc.Faults.SUCCESS,
              'description': 'OK'},
-            {'name':'failed', 'group':'failed_group',
+            {'name':'failed',
+             'group':'failed_group',
              'status':xmlrpc.Faults.FAILED,
              'description':'FAILED'}
             ]
@@ -939,6 +942,52 @@ class DummySupervisorRPCNamespace:
             raise Fault(self._readlog_error, '')
         return 'mainlogdata'
 
+    def signalGroup(self, name, signal):
+        from supervisor import xmlrpc
+        return [
+            {'name':'foo_00',
+             'group':'foo',
+             'status': xmlrpc.Faults.SUCCESS,
+             'description': 'OK'},
+            {'name':'foo_01',
+             'group':'foo',
+             'status':xmlrpc.Faults.SUCCESS,
+             'description': 'OK'},
+            ]
+
+    def signalProcess(self, name, signal):
+        from supervisor import xmlrpc
+        from xmlrpclib import Fault
+        if signal == 'BAD_SIGNAL':
+            raise Fault(xmlrpc.Faults.BAD_SIGNAL, 'BAD_SIGNAL')
+        if name == 'BAD_NAME:BAD_NAME':
+            raise Fault(xmlrpc.Faults.BAD_NAME, 'BAD_NAME:BAD_NAME')
+        if name == 'BAD_NAME':
+            raise Fault(xmlrpc.Faults.BAD_NAME, 'BAD_NAME')
+        if name == 'NOT_RUNNING':
+            raise Fault(xmlrpc.Faults.NOT_RUNNING, 'NOT_RUNNING')
+        if name == 'FAILED':
+            raise Fault(xmlrpc.Faults.FAILED, 'FAILED')
+
+        return True
+
+    def signalAllProcesses(self, signal):
+        from supervisor import xmlrpc
+        return [
+            {'name':'foo',
+             'group':'foo',
+             'status': xmlrpc.Faults.SUCCESS,
+             'description': 'OK'},
+            {'name':'foo2',
+             'group':'foo2',
+             'status':xmlrpc.Faults.SUCCESS,
+             'description': 'OK'},
+            {'name':'failed',
+             'group':'failed_group',
+             'status':xmlrpc.Faults.BAD_NAME,
+             'description':'FAILED'}
+            ]
+
 class DummyPGroupConfig:
     def __init__(self, options, name='whatever', priority=999, pconfigs=None):
         self.options = options
@@ -1114,7 +1163,7 @@ class DummyPoller:
 
     def poll(self, timeout):
         return self.result
-        
+
 def dummy_handler(event, result):
     pass
 

supervisor/tests/test_supervisorctl.py

@@ -866,20 +866,78 @@ class TestDefaultControllerPlugin(unittest.TestCase):
         out = plugin.ctl.stdout.getvalue()
         self.assertTrue("signal <signal name> <name>" in out)
 
-    def test_signal_no_arg(self):
+    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_one_arg(self):
+    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_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_restart_help(self):
         plugin = self._makeOne()
         plugin.help_restart()