📁 PHP Dosya Yöneticisi
/
/
lib
/
python3.6
/
site-packages
/
supervisor
/
tests
📝
test_rpcinterfaces.py
← Geri Dön
# -*- coding: utf-8 -*- import unittest import sys import operator import os import time import errno from supervisor.tests.base import DummyOptions from supervisor.tests.base import DummySupervisor from supervisor.tests.base import DummyProcess from supervisor.tests.base import DummyPConfig from supervisor.tests.base import DummyPGroupConfig from supervisor.tests.base import DummyProcessGroup from supervisor.tests.base import PopulatedDummySupervisor from supervisor.tests.base import _NOW from supervisor.tests.base import _TIMEFORMAT from supervisor.compat import as_string, PY2 from supervisor.datatypes import Automatic class TestBase(unittest.TestCase): def setUp(self): pass def tearDown(self): pass def _assertRPCError(self, code, callable, *args, **kw): from supervisor import xmlrpc try: callable(*args, **kw) except xmlrpc.RPCError as inst: self.assertEqual(inst.code, code) else: raise AssertionError("Didn't raise") class MainXMLRPCInterfaceTests(TestBase): def _getTargetClass(self): from supervisor import xmlrpc return xmlrpc.RootRPCInterface def _makeOne(self, *args, **kw): return self._getTargetClass()(*args, **kw) def test_ctor(self): interface = self._makeOne([('supervisor', None)]) self.assertEqual(interface.supervisor, None) def test_traverse(self): dummy = DummyRPCInterface() interface = self._makeOne([('dummy', dummy)]) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.UNKNOWN_METHOD, xmlrpc.traverse, interface, 'notthere.hello', []) self._assertRPCError(xmlrpc.Faults.UNKNOWN_METHOD, xmlrpc.traverse, interface, 'supervisor._readFile', []) self._assertRPCError(xmlrpc.Faults.INCORRECT_PARAMETERS, xmlrpc.traverse, interface, 'dummy.hello', [1]) self.assertEqual(xmlrpc.traverse( interface, 'dummy.hello', []), 'Hello!') class SupervisorNamespaceXMLRPCInterfaceTests(TestBase): def _getTargetClass(self): from supervisor import rpcinterface return rpcinterface.SupervisorNamespaceRPCInterface def _makeOne(self, *args, **kw): return self._getTargetClass()(*args, **kw) def test_update(self): from supervisor import xmlrpc from supervisor.supervisord import SupervisorStates supervisord = DummySupervisor() interface = self._makeOne(supervisord) interface._update('foo') self.assertEqual(interface.update_text, 'foo') supervisord.options.mood = SupervisorStates.SHUTDOWN self._assertRPCError(xmlrpc.Faults.SHUTDOWN_STATE, interface._update, 'foo') def test_getAPIVersion(self): from supervisor import rpcinterface supervisord = DummySupervisor() interface = self._makeOne(supervisord) version = interface.getAPIVersion() self.assertEqual(version, rpcinterface.API_VERSION) self.assertEqual(interface.update_text, 'getAPIVersion') def test_getAPIVersion_aliased_to_deprecated_getVersion(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) self.assertEqual(interface.getAPIVersion, interface.getVersion) def test_getSupervisorVersion(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) version = interface.getSupervisorVersion() from supervisor import options self.assertEqual(version, options.VERSION) self.assertEqual(interface.update_text, 'getSupervisorVersion') def test_getIdentification(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) identifier = interface.getIdentification() self.assertEqual(identifier, supervisord.options.identifier) self.assertEqual(interface.update_text, 'getIdentification') def test_getState(self): from supervisor.states import getSupervisorStateDescription supervisord = DummySupervisor() interface = self._makeOne(supervisord) stateinfo = interface.getState() statecode = supervisord.options.mood statename = getSupervisorStateDescription(statecode) self.assertEqual(stateinfo['statecode'], statecode) self.assertEqual(stateinfo['statename'], statename) self.assertEqual(interface.update_text, 'getState') def test_getPID(self): options = DummyOptions() supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) self.assertEqual(interface.getPID(), options.get_pid()) self.assertEqual(interface.update_text, 'getPID') def test_readLog_aliased_to_deprecated_readMainLog(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) self.assertEqual(interface.readMainLog, interface.readLog) def test_readLog_unreadable(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.NO_FILE, interface.readLog, offset=0, length=1) def test_readLog_badargs(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) try: logfile = supervisord.options.logfile with open(logfile, 'w+') as f: f.write('x' * 2048) self._assertRPCError(xmlrpc.Faults.BAD_ARGUMENTS, interface.readLog, offset=-1, length=1) self._assertRPCError(xmlrpc.Faults.BAD_ARGUMENTS, interface.readLog, offset=-1, length=-1) finally: os.remove(logfile) def test_readLog(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) logfile = supervisord.options.logfile try: with open(logfile, 'w+') as f: f.write('x' * 2048) f.write('y' * 2048) data = interface.readLog(offset=0, length=0) self.assertEqual(interface.update_text, 'readLog') self.assertEqual(data, ('x' * 2048) + ('y' * 2048)) data = interface.readLog(offset=2048, length=0) self.assertEqual(data, 'y' * 2048) data = interface.readLog(offset=0, length=2048) self.assertEqual(data, 'x' * 2048) data = interface.readLog(offset=-4, length=0) self.assertEqual(data, 'y' * 4) finally: os.remove(logfile) def test_clearLog_unreadable(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.NO_FILE, interface.clearLog) def test_clearLog_unremoveable(self): from supervisor import xmlrpc options = DummyOptions() options.existing = [options.logfile] options.remove_exception = OSError(errno.EPERM, os.strerror(errno.EPERM)) supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) self.assertRaises(xmlrpc.RPCError, interface.clearLog) self.assertEqual(interface.update_text, 'clearLog') def test_clearLog(self): options = DummyOptions() options.existing = [options.logfile] supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) result = interface.clearLog() self.assertEqual(interface.update_text, 'clearLog') self.assertEqual(result, True) self.assertEqual(options.removed[0], options.logfile) for handler in supervisord.options.logger.handlers: self.assertEqual(handler.reopened, True) def test_shutdown(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) value = interface.shutdown() self.assertEqual(value, True) self.assertEqual(supervisord.options.mood, -1) def test_restart(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) value = interface.restart() self.assertEqual(value, True) self.assertEqual(supervisord.options.mood, 0) def test_reloadConfig(self): options = DummyOptions() supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) changes = [ [DummyPGroupConfig(options, 'added')], [DummyPGroupConfig(options, 'changed')], [DummyPGroupConfig(options, 'dropped')] ] supervisord.diff_to_active = lambda : changes value = interface.reloadConfig() self.assertEqual(value, [[['added'], ['changed'], ['dropped']]]) def test_reloadConfig_process_config_raises_ValueError(self): from supervisor import xmlrpc options = DummyOptions() def raise_exc(*arg, **kw): raise ValueError('foo') options.process_config = raise_exc supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.CANT_REREAD, interface.reloadConfig) def test_addProcessGroup(self): from supervisor.supervisord import Supervisor from supervisor import xmlrpc options = DummyOptions() supervisord = Supervisor(options) pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) gconfig = DummyPGroupConfig(options, 'group1', pconfigs=[pconfig]) supervisord.options.process_group_configs = [gconfig] interface = self._makeOne(supervisord) result = interface.addProcessGroup('group1') self.assertTrue(result) self.assertEqual(list(supervisord.process_groups.keys()), ['group1']) self._assertRPCError(xmlrpc.Faults.ALREADY_ADDED, interface.addProcessGroup, 'group1') self.assertEqual(list(supervisord.process_groups.keys()), ['group1']) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.addProcessGroup, 'asdf') self.assertEqual(list(supervisord.process_groups.keys()), ['group1']) def test_removeProcessGroup(self): from supervisor.supervisord import Supervisor options = DummyOptions() supervisord = Supervisor(options) pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) gconfig = DummyPGroupConfig(options, 'group1', pconfigs=[pconfig]) supervisord.options.process_group_configs = [gconfig] interface = self._makeOne(supervisord) interface.addProcessGroup('group1') result = interface.removeProcessGroup('group1') self.assertTrue(result) self.assertEqual(list(supervisord.process_groups.keys()), []) def test_removeProcessGroup_bad_name(self): from supervisor.supervisord import Supervisor from supervisor import xmlrpc options = DummyOptions() supervisord = Supervisor(options) pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) gconfig = DummyPGroupConfig(options, 'group1', pconfigs=[pconfig]) supervisord.options.process_group_configs = [gconfig] interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.removeProcessGroup, 'asdf') def test_removeProcessGroup_still_running(self): from supervisor.supervisord import Supervisor from supervisor import xmlrpc options = DummyOptions() supervisord = Supervisor(options) pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) gconfig = DummyPGroupConfig(options, 'group1', pconfigs=[pconfig]) supervisord.options.process_group_configs = [gconfig] process = DummyProcessGroup(gconfig) process.unstopped_processes = [123] supervisord.process_groups = {'group1':process} interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.STILL_RUNNING, interface.removeProcessGroup, 'group1') def test_startProcess_already_started(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'pid', 10) interface = self._makeOne(supervisord) self._assertRPCError( xmlrpc.Faults.ALREADY_STARTED, interface.startProcess, 'foo' ) def test_startProcess_unknown_state(self): from supervisor import xmlrpc from supervisor.states import ProcessStates options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'pid', 10) supervisord.set_procattr('foo', 'state', ProcessStates.UNKNOWN) interface = self._makeOne(supervisord) self._assertRPCError( xmlrpc.Faults.FAILED, interface.startProcess, 'foo' ) process = supervisord.process_groups['foo'].processes['foo'] self.assertEqual(process.spawned, False) def test_startProcess_bad_group_name(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) supervisord = PopulatedDummySupervisor(options, 'group1', pconfig) interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.startProcess, 'group2:foo') def test_startProcess_bad_process_name(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) supervisord = PopulatedDummySupervisor(options, 'group1', pconfig) interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.startProcess, 'group1:bar') def test_startProcess_file_not_found(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/foo/bar', autostart=False) from supervisor.options import NotFound supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) process = supervisord.process_groups['foo'].processes['foo'] process.execv_arg_exception = NotFound interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.NO_FILE, interface.startProcess, 'foo') def test_startProcess_file_not_executable(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/foo/bar', autostart=False) from supervisor.options import NotExecutable supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) process = supervisord.process_groups['foo'].processes['foo'] process.execv_arg_exception = NotExecutable interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.NOT_EXECUTABLE, interface.startProcess, 'foo') def test_startProcess_spawnerr(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) process = supervisord.process_groups['foo'].processes['foo'] process.spawnerr = 'abc' interface = self._makeOne(supervisord) self._assertRPCError( xmlrpc.Faults.SPAWN_ERROR, interface.startProcess, 'foo' ) def test_startProcess(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False, startsecs=.01) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) result = interface.startProcess('foo') process = supervisord.process_groups['foo'].processes['foo'] self.assertEqual(process.spawned, True) self.assertEqual(interface.update_text, 'startProcess') process.state = ProcessStates.RUNNING self.assertEqual(result, True) def test_startProcess_spawnerr_in_onwait(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) process = supervisord.process_groups['foo'].processes['foo'] def spawn(): process.spawned = True process.state = ProcessStates.STARTING def transition(): process.spawnerr = 'abc' process.spawn = spawn process.transition = transition interface = self._makeOne(supervisord) callback = interface.startProcess('foo') self._assertRPCError(xmlrpc.Faults.SPAWN_ERROR, callback) def test_startProcess_success_in_onwait(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) process = supervisord.process_groups['foo'].processes['foo'] def spawn(): process.spawned = True process.state = ProcessStates.STARTING process.spawn = spawn interface = self._makeOne(supervisord) callback = interface.startProcess('foo') process.state = ProcessStates.RUNNING self.assertEqual(callback(), True) def test_startProcess_nowait(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) result = interface.startProcess('foo', wait=False) self.assertEqual(result, True) process = supervisord.process_groups['foo'].processes['foo'] self.assertEqual(process.spawned, True) self.assertEqual(interface.update_text, 'startProcess') def test_startProcess_abnormal_term_process_not_running(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__, autostart=False) from supervisor.process import ProcessStates from supervisor import http supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] def spawn(): process.spawned = True process.state = ProcessStates.STARTING process.spawn = spawn callback = interface.startProcess('foo', 100) # milliseconds result = callback() self.assertEqual(result, http.NOT_DONE_YET) self.assertEqual(process.spawned, True) self.assertEqual(interface.update_text, 'startProcess') process.state = ProcessStates.BACKOFF from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.ABNORMAL_TERMINATION, callback) def test_startProcess_splat_calls_startProcessGroup(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, autostart=False, startsecs=.01) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2, startsecs=.01) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.STOPPED) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) interface.startProcess('foo:*') self.assertEqual(interface.update_text, 'startProcessGroup') def test_startProcessGroup(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, priority=1, startsecs=.01) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2, startsecs=.01) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates from supervisor.xmlrpc import Faults supervisord.set_procattr('process1', 'state', ProcessStates.STOPPED) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) callback = interface.startProcessGroup('foo') self.assertEqual( callback(), [{'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process1'}, {'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process2'} ] ) self.assertEqual(interface.update_text, 'startProcess') process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.spawned, True) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.spawned, True) def test_startProcessGroup_nowait(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, priority=1, startsecs=.01) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2, startsecs=.01) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.STOPPED) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) callback = interface.startProcessGroup('foo', wait=False) from supervisor.xmlrpc import Faults self.assertEqual( callback(), [{'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process1'}, {'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process2'} ] ) def test_startProcessGroup_badname(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.startProcessGroup, 'foo') def test_startAllProcesses(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, priority=1, startsecs=.01) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2, startsecs=.01) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.STOPPED) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) callback = interface.startAllProcesses() from supervisor.xmlrpc import Faults self.assertEqual( callback(), [{'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process1'}, {'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process2'} ] ) self.assertEqual(interface.update_text, 'startProcess') process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.spawned, True) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.spawned, True) def test_startAllProcesses_nowait(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, priority=1, startsecs=.01) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2, startsecs=.01) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.STOPPED) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) callback = interface.startAllProcesses(wait=False) from supervisor.xmlrpc import Faults self.assertEqual( callback(), [{'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process1'}, {'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process2'} ] ) def test_stopProcess(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) result = interface.stopProcess('foo') self.assertTrue(result) self.assertEqual(interface.update_text, 'stopProcess') process = supervisord.process_groups['foo'].processes['foo'] self.assertEqual(process.backoff, 0) self.assertEqual(process.delay, 0) self.assertFalse(process.killing) self.assertEqual(process.state, ProcessStates.STOPPED) self.assertTrue(process.stop_report_called) self.assertEqual(len(supervisord.process_groups['foo'].processes), 1) self.assertEqual(interface.update_text, 'stopProcess') def test_stopProcess_nowait(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', __file__) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) result = interface.stopProcess('foo', wait=False) self.assertEqual(result, True) process = supervisord.process_groups['foo'].processes['foo'] self.assertEqual(process.stop_called, True) self.assertTrue(process.stop_report_called) self.assertEqual(interface.update_text, 'stopProcess') def test_stopProcess_success_in_onwait(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) process = supervisord.process_groups['foo'].processes['foo'] L = [ ProcessStates.RUNNING, ProcessStates.STOPPING, ProcessStates.STOPPED ] def get_state(): return L.pop(0) process.get_state = get_state interface = self._makeOne(supervisord) callback = interface.stopProcess('foo') self.assertEqual(interface.update_text, 'stopProcess') self.assertTrue(callback()) def test_stopProcess_NDY_in_onwait(self): from supervisor.http import NOT_DONE_YET options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) process = supervisord.process_groups['foo'].processes['foo'] L = [ ProcessStates.RUNNING, ProcessStates.STOPPING, ProcessStates.STOPPING ] def get_state(): return L.pop(0) process.get_state = get_state interface = self._makeOne(supervisord) callback = interface.stopProcess('foo') self.assertEqual(callback(), NOT_DONE_YET) self.assertEqual(interface.update_text, 'stopProcess') def test_stopProcess_bad_name(self): from supervisor.xmlrpc import Faults supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(Faults.BAD_NAME, interface.stopProcess, 'foo') def test_stopProcess_not_running(self): from supervisor.states import ProcessStates from supervisor.xmlrpc import Faults options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.EXITED) interface = self._makeOne(supervisord) self._assertRPCError(Faults.NOT_RUNNING, interface.stopProcess, 'foo') def test_stopProcess_failed(self): from supervisor.xmlrpc import Faults options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'stop', lambda: 'unstoppable') interface = self._makeOne(supervisord) self._assertRPCError(Faults.FAILED, interface.stopProcess, 'foo') def test_stopProcessGroup(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo', priority=1) pconfig2 = DummyPConfig(options, 'process2', '/bin/foo2', priority=2) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) callback = interface.stopProcessGroup('foo') self.assertEqual(interface.update_text, 'stopProcessGroup') from supervisor import http value = http.NOT_DONE_YET while 1: value = callback() if value is not http.NOT_DONE_YET: break from supervisor.xmlrpc import Faults self.assertEqual(value, [ {'status': Faults.SUCCESS, 'group':'foo', 'name': 'process1', 'description': 'OK'}, {'status': Faults.SUCCESS, 'group':'foo', 'name': 'process2', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.stop_called, True) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.stop_called, True) def test_stopProcessGroup_nowait(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, priority=1) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) callback = interface.stopProcessGroup('foo', wait=False) from supervisor.xmlrpc import Faults self.assertEqual( callback(), [ {'name': 'process1', 'description': 'OK', 'group': 'foo', 'status': Faults.SUCCESS}, {'name': 'process2', 'description': 'OK', 'group': 'foo', 'status': Faults.SUCCESS} ] ) def test_stopProcessGroup_badname(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.stopProcessGroup, 'foo') def test_stopProcess_splat_calls_stopProcessGroup(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, autostart=False, startsecs=.01) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2, startsecs=.01) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.STOPPED) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) interface.stopProcess('foo:*') self.assertEqual(interface.update_text, 'stopProcessGroup') def test_stopAllProcesses(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo', priority=1) pconfig2 = DummyPConfig(options, 'process2', '/bin/foo2', priority=2) from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) callback = interface.stopAllProcesses() self.assertEqual(interface.update_text, 'stopAllProcesses') from supervisor import http value = http.NOT_DONE_YET while 1: value = callback() if value is not http.NOT_DONE_YET: break from supervisor.xmlrpc import Faults self.assertEqual(value, [ {'status': Faults.SUCCESS, 'group':'foo', 'name': 'process1', 'description': 'OK'}, {'status': Faults.SUCCESS, 'group':'foo', 'name': 'process2', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.stop_called, True) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.stop_called, True) def test_stopAllProcesses_nowait(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', __file__, priority=1) pconfig2 = DummyPConfig(options, 'process2', __file__, priority=2) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) from supervisor.process import ProcessStates supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) callback = interface.stopAllProcesses(wait=False) from supervisor.xmlrpc import Faults self.assertEqual( callback(), [{'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process1'}, {'group': 'foo', 'status': Faults.SUCCESS, 'description': 'OK', 'name': 'process2'} ] ) def test_signalProcess_with_signal_number(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) result = interface.signalProcess('foo', 10) self.assertEqual(interface.update_text, 'signalProcess') self.assertEqual(result, True) p = supervisord.process_groups[supervisord.group_name].processes['foo'] self.assertEqual(p.sent_signal, 10) def test_signalProcess_with_signal_name(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.RUNNING) from supervisor.datatypes import signal_number signame = 'quit' signum = signal_number(signame) interface = self._makeOne(supervisord) result = interface.signalProcess('foo', signame) self.assertEqual(interface.update_text, 'signalProcess') self.assertEqual(result, True) p = supervisord.process_groups[supervisord.group_name].processes['foo'] self.assertEqual(p.sent_signal, signum) def test_signalProcess_stopping(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPING) interface = self._makeOne(supervisord) result = interface.signalProcess('foo', 10) self.assertEqual(interface.update_text, 'signalProcess') self.assertEqual(result, True) p = supervisord.process_groups[supervisord.group_name].processes['foo'] self.assertEqual(p.sent_signal, 10) def test_signalProcess_badsignal(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates from supervisor import xmlrpc supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) self._assertRPCError( xmlrpc.Faults.BAD_SIGNAL, interface.signalProcess, 'foo', 155 ) def test_signalProcess_notrunning(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates from supervisor import xmlrpc supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) self._assertRPCError( xmlrpc.Faults.NOT_RUNNING, interface.signalProcess, 'foo', 10 ) def test_signalProcess_signal_returns_message(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') from supervisor.process import ProcessStates from supervisor import xmlrpc supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.RUNNING) def signalreturn(sig): return 'msg' pgroup = supervisord.process_groups[supervisord.group_name] proc = pgroup.processes['foo'] proc.signal = signalreturn interface = self._makeOne(supervisord) self._assertRPCError( xmlrpc.Faults.FAILED, interface.signalProcess, 'foo', 10 ) def test_signalProcess_withgroup(self): """ Test that sending foo:* works """ options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo') pconfig2 = DummyPConfig(options, 'process2', '/bin/foo2') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPING) interface = self._makeOne(supervisord) result = interface.signalProcess('foo:*', 10) self.assertEqual(interface.update_text, 'signalProcessGroup') # Sort so we get deterministic results despite hash randomization result = sorted(result, key=operator.itemgetter('name')) from supervisor.xmlrpc import Faults self.assertEqual(result, [ {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process1', 'description': 'OK'}, {'status': Faults.SUCCESS, 'group':'foo', 'name': 'process2', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.sent_signal, 10) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.sent_signal, 10) def test_signalProcessGroup_with_signal_number(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo') pconfig2 = DummyPConfig(options, 'process2', '/bin/foo2') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) result = interface.signalProcessGroup('foo', 10) self.assertEqual(interface.update_text, 'signalProcessGroup') # Sort so we get deterministic results despite hash randomization result = sorted(result, key=operator.itemgetter('name')) from supervisor.xmlrpc import Faults self.assertEqual(result, [ {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process1', 'description': 'OK'}, {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process2', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.sent_signal, 10) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.sent_signal, 10) def test_signalProcessGroup_with_signal_name(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) from supervisor.datatypes import signal_number signame = 'term' signum = signal_number(signame) interface = self._makeOne(supervisord) result = interface.signalProcessGroup('foo', signame) self.assertEqual(interface.update_text, 'signalProcessGroup') from supervisor.xmlrpc import Faults self.assertEqual(result, [ {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process1', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.sent_signal, signum) def test_signalProcessGroup_nosuchgroup(self): from supervisor import xmlrpc options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo') pconfig2 = DummyPConfig(options, 'process2', '/bin/foo2') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.signalProcessGroup, 'bar', 10 ) def test_signalAllProcesses_with_signal_number(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo') pconfig2 = DummyPConfig(options, 'process2', '/bin/foo2') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'state', ProcessStates.RUNNING) interface = self._makeOne(supervisord) result = interface.signalAllProcesses(10) self.assertEqual(interface.update_text, 'signalAllProcesses') # Sort so we get deterministic results despite hash randomization result = sorted(result, key=operator.itemgetter('name')) from supervisor.xmlrpc import Faults self.assertEqual(result, [ {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process1', 'description': 'OK'}, {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process2', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.sent_signal, 10) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.sent_signal, 10) def test_signalAllProcesses_with_signal_name(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', '/bin/foo') from supervisor.process import ProcessStates supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) from supervisor.datatypes import signal_number signame = 'hup' signum = signal_number(signame) interface = self._makeOne(supervisord) result = interface.signalAllProcesses(signame) self.assertEqual(interface.update_text, 'signalAllProcesses') from supervisor.xmlrpc import Faults self.assertEqual(result, [ {'status': Faults.SUCCESS, 'group': 'foo', 'name': 'process1', 'description': 'OK'}, ] ) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.sent_signal, signum) def test_getAllConfigInfo(self): options = DummyOptions() supervisord = DummySupervisor(options, 'foo') pconfig1 = DummyPConfig(options, 'process1', __file__, stdout_logfile=Automatic, stderr_logfile=Automatic, ) pconfig2 = DummyPConfig(options, 'process2', __file__, stdout_logfile=None, stderr_logfile=None, ) gconfig = DummyPGroupConfig(options, 'group1', pconfigs=[pconfig1, pconfig2]) supervisord.process_groups = {'group1': DummyProcessGroup(gconfig)} supervisord.options.process_group_configs = [gconfig] interface = self._makeOne(supervisord) configs = interface.getAllConfigInfo() self.assertEqual(configs[0]['autostart'], True) self.assertEqual(configs[0]['stopwaitsecs'], 10) self.assertEqual(configs[0]['stdout_events_enabled'], False) self.assertEqual(configs[0]['stderr_events_enabled'], False) self.assertEqual(configs[0]['group'], 'group1') self.assertEqual(configs[0]['stdout_capture_maxbytes'], 0) self.assertEqual(configs[0]['name'], 'process1') self.assertEqual(configs[0]['stopsignal'], 15) self.assertEqual(configs[0]['stderr_syslog'], False) self.assertEqual(configs[0]['stdout_logfile_maxbytes'], 0) self.assertEqual(configs[0]['group_prio'], 999) self.assertEqual(configs[0]['killasgroup'], False) self.assertEqual(configs[0]['process_prio'], 999) self.assertEqual(configs[0]['stdout_syslog'], False) self.assertEqual(configs[0]['stderr_logfile_maxbytes'], 0) self.assertEqual(configs[0]['startsecs'], 10) self.assertEqual(configs[0]['redirect_stderr'], False) self.assertEqual(configs[0]['stdout_logfile'], 'auto') self.assertEqual(configs[0]['exitcodes'], (0,)) self.assertEqual(configs[0]['stderr_capture_maxbytes'], 0) self.assertEqual(configs[0]['startretries'], 999) self.assertEqual(configs[0]['stderr_logfile_maxbytes'], 0) self.assertEqual(configs[0]['inuse'], True) self.assertEqual(configs[0]['stderr_logfile'], 'auto') self.assertEqual(configs[0]['stdout_logfile_backups'], 0) assert 'test_rpcinterfaces.py' in configs[0]['command'] self.assertEqual(configs[1]['autostart'], True) self.assertEqual(configs[1]['stopwaitsecs'], 10) self.assertEqual(configs[1]['stdout_events_enabled'], False) self.assertEqual(configs[1]['stderr_events_enabled'], False) self.assertEqual(configs[1]['group'], 'group1') self.assertEqual(configs[1]['stdout_capture_maxbytes'], 0) self.assertEqual(configs[1]['name'], 'process2') self.assertEqual(configs[1]['stopsignal'], 15) self.assertEqual(configs[1]['stderr_syslog'], False) self.assertEqual(configs[1]['stdout_logfile_maxbytes'], 0) self.assertEqual(configs[1]['group_prio'], 999) self.assertEqual(configs[1]['killasgroup'], False) self.assertEqual(configs[1]['process_prio'], 999) self.assertEqual(configs[1]['stdout_syslog'], False) self.assertEqual(configs[1]['stderr_logfile_maxbytes'], 0) self.assertEqual(configs[1]['startsecs'], 10) self.assertEqual(configs[1]['redirect_stderr'], False) self.assertEqual(configs[1]['stdout_logfile'], 'none') self.assertEqual(configs[1]['exitcodes'], (0,)) self.assertEqual(configs[1]['stderr_capture_maxbytes'], 0) self.assertEqual(configs[1]['startretries'], 999) self.assertEqual(configs[1]['stderr_logfile_maxbytes'], 0) self.assertEqual(configs[1]['inuse'], True) self.assertEqual(configs[1]['stderr_logfile'], 'none') self.assertEqual(configs[1]['stdout_logfile_backups'], 0) assert 'test_rpcinterfaces.py' in configs[0]['command'] def test_getAllConfigInfo_filters_types_not_compatible_with_xmlrpc(self): options = DummyOptions() supervisord = DummySupervisor(options, 'foo') pconfig1 = DummyPConfig(options, 'process1', __file__) pconfig2 = DummyPConfig(options, 'process2', __file__) gconfig = DummyPGroupConfig(options, 'group1', pconfigs=[pconfig1, pconfig2]) supervisord.process_groups = {'group1': DummyProcessGroup(gconfig)} supervisord.options.process_group_configs = [gconfig] interface = self._makeOne(supervisord) unmarshallables = [type(None)] try: from enum import Enum unmarshallables.append(Enum) except ImportError: # python 2 pass for typ in unmarshallables: for config in interface.getAllConfigInfo(): for k, v in config.items(): self.assertFalse(isinstance(v, typ), k) def test__interpretProcessInfo(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) start = _NOW -100 stop = _NOW -1 from supervisor.process import ProcessStates running = {'name':'running', 'pid':1, 'state':ProcessStates.RUNNING, 'start':start, 'stop':stop, 'now':_NOW} description = interface._interpretProcessInfo(running) self.assertEqual(description, 'pid 1, uptime 0:01:40') fatal = {'name':'fatal', 'pid':2, 'state':ProcessStates.FATAL, 'start':start, 'stop':stop, 'now':_NOW, 'spawnerr':'Hosed'} description = interface._interpretProcessInfo(fatal) self.assertEqual(description, 'Hosed') fatal2 = {'name':'fatal', 'pid':2, 'state':ProcessStates.FATAL, 'start':start, 'stop':stop, 'now':_NOW, 'spawnerr':'',} description = interface._interpretProcessInfo(fatal2) self.assertEqual(description, 'unknown error (try "tail fatal")') stopped = {'name':'stopped', 'pid':3, 'state':ProcessStates.STOPPED, 'start':start, 'stop':stop, 'now':_NOW, 'spawnerr':'',} description = interface._interpretProcessInfo(stopped) from datetime import datetime stoptime = datetime(*time.localtime(stop)[:7]) self.assertEqual(description, stoptime.strftime(_TIMEFORMAT)) stopped2 = {'name':'stopped', 'pid':3, 'state':ProcessStates.STOPPED, 'start':0, 'stop':stop, 'now':_NOW, 'spawnerr':'',} description = interface._interpretProcessInfo(stopped2) self.assertEqual(description, 'Not started') def test__interpretProcessInfo_doesnt_report_negative_uptime(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) from supervisor.process import ProcessStates running = {'name': 'running', 'pid': 42, 'state': ProcessStates.RUNNING, 'start': _NOW + 10, # started in the future 'stop': None, 'now': _NOW} description = interface._interpretProcessInfo(running) self.assertEqual(description, 'pid 42, uptime 0:00:00') def test_getProcessInfo(self): from supervisor.process import ProcessStates options = DummyOptions() config = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile='/tmp/fleeb.bar') process = DummyProcess(config) process.pid = 111 process.laststart = 10 process.laststop = 11 pgroup_config = DummyPGroupConfig(options, name='foo') pgroup = DummyProcessGroup(pgroup_config) pgroup.processes = {'foo':process} supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) data = interface.getProcessInfo('foo') self.assertEqual(interface.update_text, 'getProcessInfo') self.assertEqual(data['logfile'], '/tmp/fleeb.bar') self.assertEqual(data['stdout_logfile'], '/tmp/fleeb.bar') self.assertEqual(data['stderr_logfile'], '') self.assertEqual(data['name'], 'foo') self.assertEqual(data['pid'], 111) self.assertEqual(data['start'], 10) self.assertEqual(data['stop'], 11) self.assertEqual(data['state'], ProcessStates.RUNNING) self.assertEqual(data['statename'], 'RUNNING') self.assertEqual(data['exitstatus'], 0) self.assertEqual(data['spawnerr'], '') self.assertTrue(data['description'].startswith('pid 111')) def test_getProcessInfo_logfile_NONE(self): options = DummyOptions() config = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile=None) process = DummyProcess(config) process.pid = 111 process.laststart = 10 process.laststop = 11 pgroup_config = DummyPGroupConfig(options, name='foo') pgroup = DummyProcessGroup(pgroup_config) pgroup.processes = {'foo':process} supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) data = interface.getProcessInfo('foo') self.assertEqual(data['logfile'], '') self.assertEqual(data['stdout_logfile'], '') def test_getProcessInfo_unknown_state(self): from supervisor.states import ProcessStates options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) supervisord.set_procattr('foo', 'state', ProcessStates.UNKNOWN) interface = self._makeOne(supervisord) data = interface.getProcessInfo('foo') self.assertEqual(data['statename'], 'UNKNOWN') self.assertEqual(data['description'], '') def test_getProcessInfo_bad_name_when_bad_process(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.getProcessInfo, 'nonexistent') def test_getProcessInfo_bad_name_when_no_process(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.getProcessInfo, 'foo:') def test_getProcessInfo_caps_timestamps_exceeding_xmlrpc_maxint(self): from supervisor.compat import xmlrpclib options = DummyOptions() config = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile=None) process = DummyProcess(config) process.laststart = float(xmlrpclib.MAXINT + 1) process.laststop = float(xmlrpclib.MAXINT + 1) pgroup_config = DummyPGroupConfig(options, name='foo') pgroup = DummyProcessGroup(pgroup_config) pgroup.processes = {'foo':process} supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) interface._now = lambda: float(xmlrpclib.MAXINT + 1) data = interface.getProcessInfo('foo') self.assertEqual(data['start'], xmlrpclib.MAXINT) self.assertEqual(data['stop'], xmlrpclib.MAXINT) self.assertEqual(data['now'], xmlrpclib.MAXINT) def test_getAllProcessInfo(self): from supervisor.process import ProcessStates options = DummyOptions() p1config = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') p2config = DummyPConfig(options, 'process2', '/bin/process2', priority=2, stdout_logfile='/tmp/process2.log') supervisord = PopulatedDummySupervisor(options, 'gname', p1config, p2config) supervisord.set_procattr('process1', 'pid', 111) supervisord.set_procattr('process1', 'laststart', 10) supervisord.set_procattr('process1', 'laststop', 11) supervisord.set_procattr('process1', 'state', ProcessStates.RUNNING) supervisord.set_procattr('process2', 'pid', 0) supervisord.set_procattr('process2', 'laststart', 20) supervisord.set_procattr('process2', 'laststop', 11) supervisord.set_procattr('process2', 'state', ProcessStates.STOPPED) interface = self._makeOne(supervisord) info = interface.getAllProcessInfo() self.assertEqual(interface.update_text, 'getProcessInfo') self.assertEqual(len(info), 2) p1info = info[0] self.assertEqual(p1info['logfile'], '/tmp/process1.log') self.assertEqual(p1info['stdout_logfile'], '/tmp/process1.log') self.assertEqual(p1info['stderr_logfile'], '') self.assertEqual(p1info['name'], 'process1') self.assertEqual(p1info['pid'], 111) self.assertEqual(p1info['start'], 10) self.assertEqual(p1info['stop'], 11) self.assertEqual(p1info['state'], ProcessStates.RUNNING) self.assertEqual(p1info['statename'], 'RUNNING') self.assertEqual(p1info['exitstatus'], 0) self.assertEqual(p1info['spawnerr'], '') self.assertEqual(p1info['group'], 'gname') self.assertTrue(p1info['description'].startswith('pid 111')) p2info = info[1] process2 = supervisord.process_groups['gname'].processes['process2'] self.assertEqual(p2info['logfile'], '/tmp/process2.log') self.assertEqual(p2info['stdout_logfile'], '/tmp/process2.log') self.assertEqual(p1info['stderr_logfile'], '') self.assertEqual(p2info['name'], 'process2') self.assertEqual(p2info['pid'], 0) self.assertEqual(p2info['start'], process2.laststart) self.assertEqual(p2info['stop'], 11) self.assertEqual(p2info['state'], ProcessStates.STOPPED) self.assertEqual(p2info['statename'], 'STOPPED') self.assertEqual(p2info['exitstatus'], 0) self.assertEqual(p2info['spawnerr'], '') self.assertEqual(p1info['group'], 'gname') from datetime import datetime starttime = datetime(*time.localtime(process2.laststart)[:7]) self.assertEqual(p2info['description'], starttime.strftime(_TIMEFORMAT)) def test_readProcessStdoutLog_unreadable(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.NO_FILE, interface.readProcessStdoutLog, 'process1', offset=0, length=1) def test_readProcessStdoutLog_badargs(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['process1'].processes['process1'] logfile = process.config.stdout_logfile try: with open(logfile, 'w+') as f: f.write('x' * 2048) self._assertRPCError(xmlrpc.Faults.BAD_ARGUMENTS, interface.readProcessStdoutLog, 'process1', offset=-1, length=1) self._assertRPCError(xmlrpc.Faults.BAD_ARGUMENTS, interface.readProcessStdoutLog, 'process1', offset=-1, length=-1) finally: os.remove(logfile) def test_readProcessStdoutLog_bad_name_no_process(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.readProcessStdoutLog, 'foo:*', offset=0, length=1) def test_readProcessStdoutLog(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stdout_logfile try: with open(logfile, 'w+') as f: f.write('x' * 2048) f.write('y' * 2048) data = interface.readProcessStdoutLog('foo', offset=0, length=0) self.assertEqual(interface.update_text, 'readProcessStdoutLog') self.assertEqual(data, ('x' * 2048) + ('y' * 2048)) data = interface.readProcessStdoutLog('foo', offset=2048, length=0) self.assertEqual(data, 'y' * 2048) data = interface.readProcessStdoutLog('foo', offset=0, length=2048) self.assertEqual(data, 'x' * 2048) data = interface.readProcessStdoutLog('foo', offset=-4, length=0) self.assertEqual(data, 'y' * 4) finally: os.remove(logfile) def test_readProcessLogAliasedTo_readProcessStdoutLog(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self.assertEqual(interface.readProcessLog, interface.readProcessStdoutLog) def test_readProcessStderrLog_unreadable(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stderr_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.NO_FILE, interface.readProcessStderrLog, 'process1', offset=0, length=1) def test_readProcessStderrLog_badargs(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stderr_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['process1'].processes['process1'] logfile = process.config.stderr_logfile try: with open(logfile, 'w+') as f: f.write('x' * 2048) self._assertRPCError(xmlrpc.Faults.BAD_ARGUMENTS, interface.readProcessStderrLog, 'process1', offset=-1, length=1) self._assertRPCError(xmlrpc.Faults.BAD_ARGUMENTS, interface.readProcessStderrLog, 'process1', offset=-1, length=-1) finally: os.remove(logfile) def test_readProcessStderrLog_bad_name_no_process(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.readProcessStderrLog, 'foo:*', offset=0, length=1) def test_readProcessStderrLog(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stderr_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stderr_logfile try: with open(logfile, 'w+') as f: f.write('x' * 2048) f.write('y' * 2048) data = interface.readProcessStderrLog('foo', offset=0, length=0) self.assertEqual(interface.update_text, 'readProcessStderrLog') self.assertEqual(data, ('x' * 2048) + ('y' * 2048)) data = interface.readProcessStderrLog('foo', offset=2048, length=0) self.assertEqual(data, 'y' * 2048) data = interface.readProcessStderrLog('foo', offset=0, length=2048) self.assertEqual(data, 'x' * 2048) data = interface.readProcessStderrLog('foo', offset=-4, length=0) self.assertEqual(data, 'y' * 4) finally: os.remove(logfile) def test_tailProcessStdoutLog_bad_name(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.tailProcessStdoutLog, 'BAD_NAME', 0, 10) def test_tailProcessStdoutLog_bad_name_no_process(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.tailProcessStdoutLog, 'foo:*', 0, 10) def test_tailProcessStdoutLog_all(self): # test entire log is returned when offset==0 and logsize < length from supervisor.compat import letters options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stdout_logfile try: with open(logfile, 'w+') as f: f.write(letters) data, offset, overflow = interface.tailProcessStdoutLog('foo', offset=0, length=len(letters)) self.assertEqual(interface.update_text, 'tailProcessStdoutLog') self.assertEqual(overflow, False) self.assertEqual(offset, len(letters)) self.assertEqual(data, letters) finally: os.remove(logfile) def test_tailProcessStdoutLog_none(self): # test nothing is returned when offset <= logsize from supervisor.compat import letters options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stdout_logfile try: with open(logfile, 'w+') as f: f.write(letters) # offset==logsize data, offset, overflow = interface.tailProcessStdoutLog('foo', offset=len(letters), length=100) self.assertEqual(interface.update_text, 'tailProcessStdoutLog') self.assertEqual(overflow, False) self.assertEqual(offset, len(letters)) self.assertEqual(data, '') # offset > logsize data, offset, overflow = interface.tailProcessStdoutLog('foo', offset=len(letters)+5, length=100) self.assertEqual(interface.update_text, 'tailProcessStdoutLog') self.assertEqual(overflow, False) self.assertEqual(offset, len(letters)) self.assertEqual(data, '') finally: os.remove(logfile) def test_tailProcessStdoutLog_overflow(self): # test buffer overflow occurs when logsize > offset+length from supervisor.compat import letters options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stdout_logfile try: with open(logfile, 'w+') as f: f.write(letters) data, offset, overflow = interface.tailProcessStdoutLog('foo', offset=0, length=5) self.assertEqual(interface.update_text, 'tailProcessStdoutLog') self.assertEqual(overflow, True) self.assertEqual(offset, len(letters)) self.assertEqual(data, letters[-5:]) finally: os.remove(logfile) def test_tailProcessStdoutLog_unreadable(self): # test nothing is returned if the log doesn't exist yet options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stdout_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) data, offset, overflow = interface.tailProcessStdoutLog('foo', offset=0, length=100) self.assertEqual(interface.update_text, 'tailProcessStdoutLog') self.assertEqual(overflow, False) self.assertEqual(offset, 0) self.assertEqual(data, '') def test_tailProcessLogAliasedTo_tailProcessStdoutLog(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self.assertEqual(interface.tailProcessLog, interface.tailProcessStdoutLog) def test_tailProcessStderrLog_bad_name(self): from supervisor import xmlrpc supervisord = DummySupervisor() interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.tailProcessStderrLog, 'BAD_NAME', 0, 10) def test_tailProcessStderrLog_bad_name_no_process(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'process1', '/bin/process1', priority=1, stdout_logfile='/tmp/process1.log') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.tailProcessStderrLog, 'foo:*', 0, 10) def test_tailProcessStderrLog_all(self): # test entire log is returned when offset==0 and logsize < length from supervisor.compat import letters options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stderr_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stderr_logfile try: with open(logfile, 'w+') as f: f.write(letters) data, offset, overflow = interface.tailProcessStderrLog('foo', offset=0, length=len(letters)) self.assertEqual(interface.update_text, 'tailProcessStderrLog') self.assertEqual(overflow, False) self.assertEqual(offset, len(letters)) self.assertEqual(data, letters) finally: os.remove(logfile) def test_tailProcessStderrLog_none(self): # test nothing is returned when offset <= logsize from supervisor.compat import letters options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stderr_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stderr_logfile try: with open(logfile, 'w+') as f: f.write(letters) # offset==logsize data, offset, overflow = interface.tailProcessStderrLog('foo', offset=len(letters), length=100) self.assertEqual(interface.update_text, 'tailProcessStderrLog') self.assertEqual(overflow, False) self.assertEqual(offset, len(letters)) self.assertEqual(data, '') # offset > logsize data, offset, overflow = interface.tailProcessStderrLog('foo', offset=len(letters)+5, length=100) self.assertEqual(interface.update_text, 'tailProcessStderrLog') self.assertEqual(overflow, False) self.assertEqual(offset, len(letters)) self.assertEqual(data, '') finally: os.remove(logfile) def test_tailProcessStderrLog_overflow(self): # test buffer overflow occurs when logsize > offset+length from supervisor.compat import letters options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stderr_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) process = supervisord.process_groups['foo'].processes['foo'] logfile = process.config.stderr_logfile try: with open(logfile, 'w+') as f: f.write(letters) data, offset, overflow = interface.tailProcessStderrLog('foo', offset=0, length=5) self.assertEqual(interface.update_text, 'tailProcessStderrLog') self.assertEqual(overflow, True) self.assertEqual(offset, len(letters)) self.assertEqual(data, letters[-5:]) finally: os.remove(logfile) def test_tailProcessStderrLog_unreadable(self): # test nothing is returned if the log doesn't exist yet options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo', stderr_logfile='/tmp/fooooooo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) data, offset, overflow = interface.tailProcessStderrLog('foo', offset=0, length=100) self.assertEqual(interface.update_text, 'tailProcessStderrLog') self.assertEqual(overflow, False) self.assertEqual(offset, 0) self.assertEqual(data, '') def test_clearProcessLogs_bad_name_no_group(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', 'foo') process = DummyProcess(pconfig) pgroup = DummyProcessGroup(None) pgroup.processes = {'foo': process} supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.clearProcessLogs, 'badgroup') def test_clearProcessLogs_bad_name_no_process(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', 'foo') process = DummyProcess(pconfig) pgroup = DummyProcessGroup(None) pgroup.processes = {'foo': process} supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.clearProcessLogs, 'foo:*') def test_clearProcessLogs(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', 'foo') process = DummyProcess(pconfig) pgroup = DummyProcessGroup(None) pgroup.processes = {'foo': process} supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) interface.clearProcessLogs('foo') self.assertEqual(process.logsremoved, True) def test_clearProcessLogs_failed(self): from supervisor import xmlrpc options = DummyOptions() pconfig = DummyPConfig(options, 'foo', 'foo') process = DummyProcess(pconfig) pgroup = DummyProcessGroup(None) pgroup.processes = {'foo': process} process.error_at_clear = True supervisord = DummySupervisor(process_groups={'foo':pgroup}) interface = self._makeOne(supervisord) self.assertRaises(xmlrpc.RPCError, interface.clearProcessLogs, 'foo') def test_clearProcessLogAliasedTo_clearProcessLogs(self): options = DummyOptions() pconfig = DummyPConfig(options, 'foo', '/bin/foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig) interface = self._makeOne(supervisord) self.assertEqual(interface.clearProcessLog, interface.clearProcessLogs) def test_clearAllProcessLogs(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo', priority=1) pconfig2 = DummyPConfig(options, 'process2', 'bar', priority=2) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) interface = self._makeOne(supervisord) callback = interface.clearAllProcessLogs() callback() results = callback() from supervisor import xmlrpc self.assertEqual(results[0], {'name':'process1', 'group':'foo', 'status':xmlrpc.Faults.SUCCESS, 'description':'OK'}) self.assertEqual(results[1], {'name':'process2', 'group':'foo', 'status':xmlrpc.Faults.SUCCESS, 'description':'OK'}) process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.logsremoved, True) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.logsremoved, True) def test_clearAllProcessLogs_onefails(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo', priority=1) pconfig2 = DummyPConfig(options, 'process2', 'bar', priority=2) supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1, pconfig2) supervisord.set_procattr('process1', 'error_at_clear', True) interface = self._makeOne(supervisord) callback = interface.clearAllProcessLogs() callback() results = callback() process1 = supervisord.process_groups['foo'].processes['process1'] self.assertEqual(process1.logsremoved, False) process2 = supervisord.process_groups['foo'].processes['process2'] self.assertEqual(process2.logsremoved, True) self.assertEqual(len(results), 2) from supervisor import xmlrpc self.assertEqual(results[0], {'name':'process1', 'group':'foo', 'status':xmlrpc.Faults.FAILED, 'description':'FAILED: foo:process1'}) self.assertEqual(results[1], {'name':'process2', 'group':'foo', 'status':xmlrpc.Faults.SUCCESS, 'description':'OK'}) def test_clearAllProcessLogs_no_processes(self): supervisord = DummySupervisor() self.assertEqual(supervisord.process_groups, {}) interface = self._makeOne(supervisord) callback = interface.clearAllProcessLogs() results = callback() self.assertEqual(results, []) def test_sendProcessStdin_raises_incorrect_params_when_not_chars(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'foo', pconfig1) interface = self._makeOne(supervisord) thing_not_chars = 42 from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.INCORRECT_PARAMETERS, interface.sendProcessStdin, 'process1', thing_not_chars) def test_sendProcessStdin_raises_bad_name_when_bad_process(self): supervisord = DummySupervisor() interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.sendProcessStdin, 'nonexistent', 'chars for stdin') def test_sendProcessStdin_raises_bad_name_when_no_process(self): options = DummyOptions() supervisord = PopulatedDummySupervisor(options, 'foo') interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.BAD_NAME, interface.sendProcessStdin, 'foo:*', 'chars for stdin') def test_sendProcessStdin_raises_not_running_when_not_process_pid(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig1) supervisord.set_procattr('process1', 'pid', 0) interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.NOT_RUNNING, interface.sendProcessStdin, 'process1', 'chars for stdin') def test_sendProcessStdin_raises_not_running_when_killing(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig1) supervisord.set_procattr('process1', 'pid', 42) supervisord.set_procattr('process1', 'killing', True) interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.NOT_RUNNING, interface.sendProcessStdin, 'process1', 'chars for stdin') def test_sendProcessStdin_raises_no_file_when_write_raises_epipe(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig1) supervisord.set_procattr('process1', 'pid', 42) supervisord.set_procattr('process1', 'killing', False) supervisord.set_procattr('process1', 'write_exception', OSError(errno.EPIPE, os.strerror(errno.EPIPE))) interface = self._makeOne(supervisord) from supervisor import xmlrpc self._assertRPCError(xmlrpc.Faults.NO_FILE, interface.sendProcessStdin, 'process1', 'chars for stdin') def test_sendProcessStdin_reraises_other_oserrors(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig1) supervisord.set_procattr('process1', 'pid', 42) supervisord.set_procattr('process1', 'killing', False) supervisord.set_procattr('process1', 'write_exception', OSError(errno.EINTR, os.strerror(errno.EINTR))) interface = self._makeOne(supervisord) self.assertRaises(OSError, interface.sendProcessStdin, 'process1', 'chars for stdin') def test_sendProcessStdin_writes_chars_and_returns_true(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig1) supervisord.set_procattr('process1', 'pid', 42) interface = self._makeOne(supervisord) chars = b'chars for stdin' self.assertTrue(interface.sendProcessStdin('process1', chars)) self.assertEqual('sendProcessStdin', interface.update_text) process1 = supervisord.process_groups['process1'].processes['process1'] self.assertEqual(process1.stdin_buffer, chars) def test_sendProcessStdin_unicode_encoded_to_utf8(self): options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') supervisord = PopulatedDummySupervisor(options, 'process1', pconfig1) supervisord.set_procattr('process1', 'pid', 42) interface = self._makeOne(supervisord) interface.sendProcessStdin('process1', as_string(b'fi\xc3\xad')) process1 = supervisord.process_groups['process1'].processes['process1'] self.assertEqual(process1.stdin_buffer, b'fi\xc3\xad') def test_sendRemoteCommEvent_notifies_subscribers(self): options = DummyOptions() supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) from supervisor import events L = [] def callback(event): L.append(event) try: events.callbacks[:] = [(events.RemoteCommunicationEvent, callback)] result = interface.sendRemoteCommEvent('foo', 'bar') finally: events.callbacks[:] = [] events.clear() self.assertTrue(result) self.assertEqual(len(L), 1) event = L[0] self.assertEqual(event.type, 'foo') self.assertEqual(event.data, 'bar') def test_sendRemoteCommEvent_unicode_encoded_to_utf8(self): options = DummyOptions() supervisord = DummySupervisor(options) interface = self._makeOne(supervisord) from supervisor import events L = [] def callback(event): L.append(event) try: events.callbacks[:] = [(events.RemoteCommunicationEvent, callback)] result = interface.sendRemoteCommEvent( as_string('fií once'), as_string('fií twice'), ) finally: events.callbacks[:] = [] events.clear() self.assertTrue(result) self.assertEqual(len(L), 1) event = L[0] if PY2: self.assertEqual(event.type, 'fi\xc3\xad once') self.assertEqual(event.data, 'fi\xc3\xad twice') else: self.assertEqual(event.type, 'fií once') self.assertEqual(event.data, 'fií twice') class SystemNamespaceXMLRPCInterfaceTests(TestBase): def _getTargetClass(self): from supervisor import xmlrpc return xmlrpc.SystemNamespaceRPCInterface def _makeOne(self): from supervisor import rpcinterface supervisord = DummySupervisor() supervisor = rpcinterface.SupervisorNamespaceRPCInterface(supervisord) return self._getTargetClass()( [('supervisor', supervisor), ] ) def test_ctor(self): interface = self._makeOne() self.assertTrue(interface.namespaces['supervisor']) self.assertTrue(interface.namespaces['system']) def test_listMethods(self): interface = self._makeOne() methods = interface.listMethods() methods.sort() keys = list(interface._listMethods().keys()) keys.sort() self.assertEqual(methods, keys) def test_methodSignature(self): from supervisor import xmlrpc interface = self._makeOne() self._assertRPCError(xmlrpc.Faults.SIGNATURE_UNSUPPORTED, interface.methodSignature, ['foo.bar']) result = interface.methodSignature('system.methodSignature') self.assertEqual(result, ['array', 'string']) def test_allMethodDocs(self): from supervisor import xmlrpc # belt-and-suspenders test for docstring-as-typing parsing correctness # and documentation validity vs. implementation _RPCTYPES = ['int', 'double', 'string', 'boolean', 'dateTime.iso8601', 'base64', 'binary', 'array', 'struct'] interface = self._makeOne() methods = interface._listMethods() for k in methods.keys(): # if a method doesn't have a @return value, an RPCError is raised. # Detect that here. try: interface.methodSignature(k) except xmlrpc.RPCError: raise AssertionError('methodSignature for %s raises ' 'RPCError (missing @return doc?)' % k) # we want to test that the number of arguments implemented in # the function is the same as the number of arguments implied by # the doc @params, and that they show up in the same order. ns_name, method_name = k.split('.', 1) namespace = interface.namespaces[ns_name] meth = getattr(namespace, method_name) try: code = meth.func_code except Exception: code = meth.__code__ argnames = code.co_varnames[1:code.co_argcount] parsed = xmlrpc.gettags(str(meth.__doc__)) plines = [] ptypes = [] pnames = [] ptexts = [] rlines = [] rtypes = [] rnames = [] rtexts = [] for thing in parsed: if thing[1] == 'param': # tag name plines.append(thing[0]) # doc line number ptypes.append(thing[2]) # data type pnames.append(thing[3]) # function name ptexts.append(thing[4]) # description elif thing[1] == 'return': # tag name rlines.append(thing[0]) # doc line number rtypes.append(thing[2]) # data type rnames.append(thing[3]) # function name rtexts.append(thing[4]) # description elif thing[1] is not None: raise AssertionError( 'unknown tag type %s for %s, parsed %s' % (thing[1], k, parsed)) # param tokens if len(argnames) != len(pnames): raise AssertionError('Incorrect documentation ' '(%s args, %s doc params) in %s' % (len(argnames), len(pnames), k)) for docline in plines: self.assertTrue(type(docline) == int, (docline, type(docline), k, parsed)) for doctype in ptypes: self.assertTrue(doctype in _RPCTYPES, doctype) for x in range(len(pnames)): if pnames[x] != argnames[x]: msg = 'Name wrong: (%s vs. %s in %s)\n%s' % (pnames[x], argnames[x], k, parsed) raise AssertionError(msg) for doctext in ptexts: self.assertTrue(type(doctext) == type(''), doctext) # result tokens if len(rlines) > 1: raise AssertionError( 'Duplicate @return values in docs for %s' % k) for docline in rlines: self.assertTrue(type(docline) == int, (docline, type(docline), k)) for doctype in rtypes: self.assertTrue(doctype in _RPCTYPES, doctype) for docname in rnames: self.assertTrue(type(docname) == type(''), (docname, type(docname), k)) for doctext in rtexts: self.assertTrue(type(doctext) == type(''), (doctext, type(doctext), k)) def test_multicall_simplevals(self): interface = self._makeOne() results = interface.multicall([ {'methodName':'system.methodHelp', 'params':['system.methodHelp']}, {'methodName':'system.listMethods', 'params':[]}, ]) self.assertEqual(results[0], interface.methodHelp('system.methodHelp')) self.assertEqual(results[1], interface.listMethods()) def test_multicall_recursion_guard(self): from supervisor import xmlrpc interface = self._makeOne() results = interface.multicall([ {'methodName': 'system.multicall', 'params': []}, ]) e = xmlrpc.RPCError(xmlrpc.Faults.INCORRECT_PARAMETERS, 'Recursive system.multicall forbidden') recursion_fault = {'faultCode': e.code, 'faultString': e.text} self.assertEqual(results, [recursion_fault]) def test_multicall_nested_callback(self): from supervisor import http interface = self._makeOne() callback = interface.multicall([ {'methodName':'supervisor.stopAllProcesses'}]) results = http.NOT_DONE_YET while results is http.NOT_DONE_YET: results = callback() self.assertEqual(results[0], []) def test_methodHelp(self): from supervisor import xmlrpc interface = self._makeOne() self._assertRPCError(xmlrpc.Faults.SIGNATURE_UNSUPPORTED, interface.methodHelp, ['foo.bar']) help = interface.methodHelp('system.methodHelp') self.assertEqual(help, interface.methodHelp.__doc__) class Test_make_allfunc(unittest.TestCase): def _callFUT(self, processes, predicate, func, **extra_kwargs): from supervisor.rpcinterface import make_allfunc return make_allfunc(processes, predicate, func, **extra_kwargs) def test_rpcerror_nocallbacks(self): from supervisor import xmlrpc def cb(name, **kw): raise xmlrpc.RPCError(xmlrpc.Faults.FAILED) options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') proc = DummyProcess(pconfig1) group = DummyProcessGroup(pconfig1) def pred(proc): return True af = self._callFUT([(group, proc)], pred, cb) result = af() self.assertEqual(result, [{'description': 'FAILED', 'group': 'process1', 'name': 'process1', 'status': xmlrpc.Faults.FAILED}]) def test_func_callback_normal_return_val(self): def cb(name, **kw): return lambda: 1 options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') proc = DummyProcess(pconfig1) group = DummyProcessGroup(pconfig1) def pred(proc): return True af = self._callFUT([(group, proc)], pred, cb) result = af() self.assertEqual( result, [{'group': 'process1', 'description': 'OK', 'status': 80, 'name': 'process1'}] ) def test_func_callback_raises_RPCError(self): from supervisor import xmlrpc def cb(name, **kw): def inner(): raise xmlrpc.RPCError(xmlrpc.Faults.FAILED) return inner options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') proc = DummyProcess(pconfig1) group = DummyProcessGroup(pconfig1) def pred(proc): return True af = self._callFUT([(group, proc)], pred, cb) result = af() self.assertEqual( result, [{'description': 'FAILED', 'group': 'process1', 'status': 30, 'name': 'process1'}] ) def test_func_callback_returns_NOT_DONE_YET(self): from supervisor.http import NOT_DONE_YET def cb(name, **kw): def inner(): return NOT_DONE_YET return inner options = DummyOptions() pconfig1 = DummyPConfig(options, 'process1', 'foo') proc = DummyProcess(pconfig1) group = DummyProcessGroup(pconfig1) def pred(proc): return True af = self._callFUT([(group, proc)], pred, cb) result = af() self.assertEqual( result, NOT_DONE_YET, ) class Test_make_main_rpcinterface(unittest.TestCase): def _callFUT(self, supervisord): from supervisor.rpcinterface import make_main_rpcinterface return make_main_rpcinterface(supervisord) def test_it(self): inst = self._callFUT(None) self.assertEqual( inst.__class__.__name__, 'SupervisorNamespaceRPCInterface' ) class DummyRPCInterface: def hello(self): return 'Hello!' def test_suite(): return unittest.findTestCases(sys.modules[__name__]) if __name__ == '__main__': unittest.main(defaultTest='test_suite')
💾 Kaydet
İptal
📝 Yeniden Adlandır
İptal
Kaydet
🔐 Dosya İzinleri (chmod)
İzin Değeri:
Hızlı Seçim:
777
755
644
600
777
= Herkes okur/yazar/çalıştırır
755
= Sahip tam, diğerleri okur/çalıştırır
644
= Sahip okur/yazar, diğerleri okur
600
= Sadece sahip okur/yazar
İptal
Uygula