فهرست منبع

Add test coverage for signal command

Mike Naberezny 10 سال پیش
والد
کامیت
fa72f360c7
2فایلهای تغییر یافته به همراه21 افزوده شده و 0 حذف شده
  1. 2 0
      supervisor/tests/base.py
  2. 19 0
      supervisor/tests/test_supervisorctl.py

+ 2 - 0
supervisor/tests/base.py

@@ -944,6 +944,8 @@ class DummySupervisorRPCNamespace:
 
     def signalProcessGroup(self, name, signal):
         from supervisor import xmlrpc
+        if name == 'BAD_NAME':
+            raise Fault(xmlrpc.Faults.BAD_NAME, 'BAD_NAME')
         return [
             {'name':'foo_00',
              'group':'foo',

+ 19 - 0
supervisor/tests/test_supervisorctl.py

@@ -894,6 +894,13 @@ class TestDefaultControllerPlugin(unittest.TestCase):
         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')
@@ -938,6 +945,18 @@ class TestDefaultControllerPlugin(unittest.TestCase):
                          '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.stopAllProcesses = f
+        supervisor.stopProcessGroup = f
+        plugin.do_signal('term foo')
+        self.assertEqual(called, [])
+
     def test_restart_help(self):
         plugin = self._makeOne()
         plugin.help_restart()