Gecko [ dataharbor.africa ]

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

Code Editor : test_dispatchers.py

import unittest
import os
import sys

from supervisor.tests.base import DummyOptions
from supervisor.tests.base import DummyProcess
from supervisor.tests.base import DummyPConfig
from supervisor.tests.base import DummyLogger
from supervisor.tests.base import DummyEvent

class POutputDispatcherTests(unittest.TestCase):
    def setUp(self):
        from supervisor.events import clear
        clear()

def tearDown(self):
        from supervisor.events import clear
        clear()

def _getTargetClass(self):
        from supervisor.dispatchers import POutputDispatcher
        return POutputDispatcher

def _makeOne(self, process, channel='stdout'):
        from supervisor import events
        events = {'stdout': events.ProcessCommunicationStdoutEvent,
                  'stderr': events.ProcessCommunicationStderrEvent}
        # dispatcher derives its channel from event class
        return self._getTargetClass()(process, events[channel], 0)

def test_writable(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.writable(), False)

def test_readable_open(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.closed = False
        self.assertEqual(dispatcher.readable(), True)

def test_readable_closed(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.closed = True
        self.assertEqual(dispatcher.readable(), False)

def test_handle_write_event(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertRaises(NotImplementedError, dispatcher.handle_write_event)

def test_handle_read_event(self):
        options = DummyOptions()
        options.readfd_result = 'abc'
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_capture_maxbytes=100)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.handle_read_event(), None)
        self.assertEqual(dispatcher.output_buffer, 'abc')

def test_handle_read_event_no_data_closes(self):
        options = DummyOptions()
        options.readfd_result = ''
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_capture_maxbytes=100)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertFalse(dispatcher.closed)
        self.assertEqual(dispatcher.handle_read_event(), None)
        self.assertEqual(dispatcher.output_buffer, '')
        self.assertTrue(dispatcher.closed)

def test_handle_error(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'test', '/test')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        try:
            raise ValueError('foo')
        except:
            dispatcher.handle_error()
        result = options.logger.data[0]
        self.assertTrue(result.startswith(
            'uncaptured python exception, closing channel'),result)

def test_toggle_capturemode_sends_event(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo',
                              stdout_capture_maxbytes=500)
        process = DummyProcess(config)
        process.pid = 4000
        dispatcher = self._makeOne(process)
        dispatcher.capturemode = True
        dispatcher.capturelog.data = ['hallooo']
        L = []
        def doit(event):
            L.append(event)
        from supervisor import events
        events.subscribe(events.EventTypes.PROCESS_COMMUNICATION, doit)
        dispatcher.toggle_capturemode()
        self.assertEqual(len(L), 1)
        event = L[0]
        self.assertEqual(event.process, process)
        self.assertEqual(event.pid, 4000)
        self.assertEqual(event.data, 'hallooo')

def test_removelogs(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.removelogs()
        self.assertEqual(dispatcher.mainlog.handlers[0].reopened, True)
        self.assertEqual(dispatcher.mainlog.handlers[0].removed, True)
        self.assertEqual(dispatcher.childlog.handlers[0].reopened, True)
        self.assertEqual(dispatcher.childlog.handlers[0].removed, True)

def test_reopenlogs(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.reopenlogs()
        self.assertEqual(dispatcher.childlog.handlers[0].reopened, True)
        self.assertEqual(dispatcher.mainlog.handlers[0].reopened, True)

def test_record_output_log_non_capturemode(self):
        # stdout/stderr goes to the process log and the main log,
        # in non-capturemode, the data length doesn't matter
        options = DummyOptions()
        from supervisor import loggers
        options.loglevel = loggers.LevelsByName.TRAC
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.output_buffer = 'a'
        dispatcher.record_output()
        self.assertEqual(dispatcher.childlog.data, ['a'])
        self.assertEqual(options.logger.data[0],
             "'process1' stdout output:\na")
        self.assertEqual(dispatcher.output_buffer, '')

def test_record_output_emits_stdout_event_when_enabled(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_events_enabled=True)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process, 'stdout')
        dispatcher.output_buffer = 'hello from stdout'

L = []
        def doit(event):
            L.append(event)
        from supervisor import events
        events.subscribe(events.EventTypes.PROCESS_LOG_STDOUT, doit)
        dispatcher.record_output()

self.assertEqual(len(L), 1)
        event = L[0]
        self.assertEqual(event.process, process)
        self.assertEqual(event.data, 'hello from stdout')

def test_record_output_does_not_emit_stdout_event_when_disabled(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_events_enabled=False)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process, 'stdout')
        dispatcher.output_buffer = 'hello from stdout'

self.assertEqual(len(L), 0)

def test_record_output_emits_stderr_event_when_enabled(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stderr_events_enabled=True)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process, 'stderr')
        dispatcher.output_buffer = 'hello from stderr'

L = []
        def doit(event):
            L.append(event)
        from supervisor import events
        events.subscribe(events.EventTypes.PROCESS_LOG_STDERR, doit)
        dispatcher.record_output()

self.assertEqual(len(L), 1)
        event = L[0]
        self.assertEqual(event.process, process)
        self.assertEqual(event.data, 'hello from stderr')

def test_record_output_does_not_emit_stderr_event_when_disabled(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stderr_events_enabled=False)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process, 'stderr')
        dispatcher.output_buffer = 'hello from stderr'

self.assertEqual(len(L), 0)

def test_record_output_capturemode_string_longer_than_token(self):
        # stdout/stderr goes to the process log and the main log,
        # in capturemode, the length of the data needs to be longer
        # than the capture token to make it out.
        options = DummyOptions()
        from supervisor import loggers
        options.loglevel = loggers.LevelsByName.TRAC
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo',
                              stdout_capture_maxbytes=100)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.output_buffer = 'stdout string longer than a token'
        dispatcher.record_output()
        self.assertEqual(dispatcher.childlog.data,
                         ['stdout string longer than a token'])
        self.assertEqual(options.logger.data[0],
             "'process1' stdout output:\nstdout string longer than a token")

def test_record_output_capturemode_string_not_longer_than_token(self):
        # stdout/stderr goes to the process log and the main log,
        # in capturemode, the length of the data needs to be longer
        # than the capture token to make it out.
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo',
                              stdout_capture_maxbytes=100)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.output_buffer = 'a'
        dispatcher.record_output()
        self.assertEqual(dispatcher.childlog.data, [])
        self.assertEqual(dispatcher.output_buffer, 'a')

def test_stdout_capturemode_single_buffer(self):
        # mike reported that comm events that took place within a single
        # output buffer were broken 8/20/2007
        from supervisor.events import ProcessCommunicationEvent
        from supervisor.events import subscribe
        events = []
        def doit(event):
            events.append(event)
        subscribe(ProcessCommunicationEvent, doit)
        BEGIN_TOKEN = ProcessCommunicationEvent.BEGIN_TOKEN
        END_TOKEN = ProcessCommunicationEvent.END_TOKEN
        data = BEGIN_TOKEN + 'hello' + END_TOKEN
        options = DummyOptions()
        from supervisor.loggers import getLogger
        options.getLogger = getLogger # actually use real logger
        logfile = '/tmp/log'
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile=logfile,
                              stdout_capture_maxbytes=1000)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)

try:
            dispatcher.output_buffer = data
            dispatcher.record_output()
            self.assertEqual(open(logfile, 'r').read(), '')
            self.assertEqual(dispatcher.output_buffer, '')
            self.assertEqual(len(events), 1)

event = events[0]
            from supervisor.events import ProcessCommunicationStdoutEvent
            self.assertEqual(event.__class__, ProcessCommunicationStdoutEvent)
            self.assertEqual(event.process, process)
            self.assertEqual(event.channel, 'stdout')
            self.assertEqual(event.data, 'hello')

finally:
            try:
                dispatcher.capturelog.close()
                dispatcher.childlog.close()
                os.remove(logfile)
            except (OSError, IOError):
                pass

def test_stdout_capturemode_multiple_buffers(self):
        from supervisor.events import ProcessCommunicationEvent
        from supervisor.events import subscribe
        events = []
        def doit(event):
            events.append(event)
        subscribe(ProcessCommunicationEvent, doit)
        import string
        # ascii_letters for python 3
        letters = getattr(string, "letters", string.ascii_letters)
        digits = string.digits * 4
        BEGIN_TOKEN = ProcessCommunicationEvent.BEGIN_TOKEN
        END_TOKEN = ProcessCommunicationEvent.END_TOKEN
        data = (letters +  BEGIN_TOKEN + digits + END_TOKEN + letters)

# boundaries that split tokens
        broken = data.split(':')
        first = broken[0] + ':'
        second = broken[1] + ':'
        third = broken[2]

options = DummyOptions()
        from supervisor.loggers import getLogger
        options.getLogger = getLogger # actually use real logger
        logfile = '/tmp/log'
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile=logfile,
                              stdout_capture_maxbytes=10000)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        try:
            dispatcher.output_buffer = first
            dispatcher.record_output()
            [ x.flush() for x in dispatcher.childlog.handlers]
            self.assertEqual(open(logfile, 'r').read(), letters)
            self.assertEqual(dispatcher.output_buffer, first[len(letters):])
            self.assertEqual(len(events), 0)

dispatcher.output_buffer += second
            dispatcher.record_output()
            self.assertEqual(len(events), 0)
            [ x.flush() for x in dispatcher.childlog.handlers]
            self.assertEqual(open(logfile, 'r').read(), letters)
            self.assertEqual(dispatcher.output_buffer, first[len(letters):])
            self.assertEqual(len(events), 0)

dispatcher.output_buffer += third
            dispatcher.record_output()
            [ x.flush() for x in dispatcher.childlog.handlers]
            self.assertEqual(open(logfile, 'r').read(), letters *2)
            self.assertEqual(len(events), 1)
            event = events[0]
            from supervisor.events import ProcessCommunicationStdoutEvent
            self.assertEqual(event.__class__, ProcessCommunicationStdoutEvent)
            self.assertEqual(event.process, process)
            self.assertEqual(event.channel, 'stdout')
            self.assertEqual(event.data, digits)

def test_strip_ansi(self):
        options = DummyOptions()
        options.strip_ansi = True
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        ansi = '\x1b[34mHello world... this is longer than a token!\x1b[0m'
        noansi = 'Hello world... this is longer than a token!'

dispatcher.output_buffer = ansi
        dispatcher.record_output()
        self.assertEqual(len(dispatcher.childlog.data), 1)
        self.assertEqual(dispatcher.childlog.data[0], noansi)

options.strip_ansi = False

dispatcher.output_buffer = ansi
        dispatcher.record_output()
        self.assertEqual(len(dispatcher.childlog.data), 2)
        self.assertEqual(dispatcher.childlog.data[1], ansi)

def test_ctor_nologfiles(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.process, process)
        self.assertEqual(dispatcher.channel, 'stdout')
        self.assertEqual(dispatcher.fd, 0)
        self.assertEqual(dispatcher.capturelog, None)
        self.assertEqual(dispatcher.mainlog, None)
        self.assertEqual(dispatcher.childlog, None)

def test_ctor_logfile_only(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.process, process)
        self.assertEqual(dispatcher.channel, 'stdout')
        self.assertEqual(dispatcher.fd, 0)
        self.assertEqual(dispatcher.capturelog, None)
        self.assertEqual(dispatcher.mainlog.__class__, DummyLogger)
        self.assertEqual(dispatcher.childlog, dispatcher.mainlog)

def test_ctor_capturelog_only(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_capture_maxbytes=300)
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.process, process)
        self.assertEqual(dispatcher.channel, 'stdout')
        self.assertEqual(dispatcher.fd, 0)
        self.assertEqual(dispatcher.capturelog.__class__,DummyLogger)
        self.assertEqual(dispatcher.mainlog, None)
        self.assertEqual(dispatcher.childlog, None)

def test_ctor_nologs(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.process, process)
        self.assertEqual(dispatcher.channel, 'stdout')
        self.assertEqual(dispatcher.fd, 0)
        self.assertEqual(dispatcher.capturelog, None)
        self.assertEqual(dispatcher.mainlog, None)
        self.assertEqual(dispatcher.childlog, None)

def test_repr(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        drepr = repr(dispatcher)
        self.assertTrue(drepr.startswith('<POutputDispatcher at'), drepr)
        self.assertNotEqual(
            drepr.find('<supervisor.tests.base.DummyProcess instance at'),
            -1)
        self.assertTrue(drepr.endswith('(stdout)>'), drepr)

def test_close(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.close()
        self.assertEqual(dispatcher.closed, True)
        dispatcher.close() # make sure we don't error if we try to close twice
        self.assertEqual(dispatcher.closed, True)

class PInputDispatcherTests(unittest.TestCase):
    def _getTargetClass(self):
        from supervisor.dispatchers import PInputDispatcher
        return PInputDispatcher

def _makeOne(self, process):
        channel = 'stdin'
        return self._getTargetClass()(process, channel, 0)

def test_writable_open_nodata(self):
        process = DummyProcess(None)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = 'a'
        dispatcher.closed = False
        self.assertEqual(dispatcher.writable(), True)

def test_writable_open_withdata(self):
        process = DummyProcess(None)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = ''
        dispatcher.closed = False
        self.assertEqual(dispatcher.writable(), False)

def test_writable_closed_nodata(self):
        process = DummyProcess(None)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = 'a'
        dispatcher.closed = True
        self.assertEqual(dispatcher.writable(), False)

def test_writable_closed_withdata(self):
        process = DummyProcess(None)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = ''
        dispatcher.closed = True
        self.assertEqual(dispatcher.writable(), False)

def test_readable(self):
        process = DummyProcess(None)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.readable(), False)

def test_handle_write_event(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = 'halloooo'
        self.assertEqual(dispatcher.handle_write_event(), None)
        self.assertEqual(options.written[0], 'halloooo')

def test_handle_write_event_nodata(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'test', '/test')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.input_buffer, '')
        dispatcher.handle_write_event()
        self.assertEqual(dispatcher.input_buffer, '')
        self.assertEqual(options.written, {})

def test_handle_write_event_epipe_raised(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'test', '/test')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = 'halloooo'
        import errno
        options.write_error = errno.EPIPE
        dispatcher.handle_write_event()
        self.assertEqual(dispatcher.input_buffer, '')
        self.assertTrue(options.logger.data[0].startswith(
            'fd 0 closed, stopped monitoring'))
        self.assertTrue(options.logger.data[0].endswith('(stdin)>'))

def test_handle_write_event_uncaught_raised(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'test', '/test')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.input_buffer = 'halloooo'
        import errno
        options.write_error = errno.EBADF
        self.assertRaises(OSError, dispatcher.handle_write_event)

def test_handle_write_event_over_os_limit(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'test', '/test')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        options.write_accept = 1
        dispatcher.input_buffer = 'a' * 50
        dispatcher.handle_write_event()
        self.assertEqual(len(dispatcher.input_buffer), 49)
        self.assertEqual(options.written[0], 'a')

def test_handle_read_event(self):
        process = DummyProcess(None)
        dispatcher = self._makeOne(process)
        self.assertRaises(NotImplementedError, dispatcher.handle_read_event)

def test_repr(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        drepr = repr(dispatcher)
        self.assertTrue(drepr.startswith('<PInputDispatcher at'), drepr)
        self.assertNotEqual(
            drepr.find('<supervisor.tests.base.DummyProcess instance at'),
            -1)
        self.assertTrue(drepr.endswith('(stdin)>'), drepr)

class PEventListenerDispatcherTests(unittest.TestCase):
    def setUp(self):
        from supervisor.events import clear
        clear()

def tearDown(self):
        from supervisor.events import clear
        clear()

def _getTargetClass(self):
        from supervisor.dispatchers import PEventListenerDispatcher
        return PEventListenerDispatcher

def _makeOne(self, process):
        channel = 'stdout'
        return self._getTargetClass()(process, channel, 0)

def test_handle_read_event_calls_handle_listener_state_change(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        process.listener_state = EventListenerStates.ACKNOWLEDGED
        dispatcher = self._makeOne(process)
        options.readfd_result = dispatcher.READY_FOR_EVENTS_TOKEN
        self.assertEqual(dispatcher.handle_read_event(), None)
        self.assertEqual(process.listener_state, EventListenerStates.READY)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(len(dispatcher.childlog.data), 1)
        self.assertEqual(dispatcher.childlog.data[0],
                         dispatcher.READY_FOR_EVENTS_TOKEN)

def test_handle_read_event_nodata(self):
        options = DummyOptions()
        options.readfd_result = ''
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.handle_read_event(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        from supervisor.dispatchers import EventListenerStates
        self.assertEqual(dispatcher.process.listener_state,
                         EventListenerStates.ACKNOWLEDGED)

def test_handle_read_event_logging_nologs(self):
        options = DummyOptions()
        options.readfd_result = 'supercalifragilisticexpialidocious'
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        # just make sure there are no errors if a child logger doesnt
        # exist
        self.assertEqual(dispatcher.handle_read_event(), None)
        self.assertEqual(dispatcher.childlog, None)

def test_handle_read_event_logging_childlog(self):
        options = DummyOptions()
        options.readfd_result = 'supercalifragilisticexpialidocious'
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        self.assertEqual(dispatcher.handle_read_event(), None)
        self.assertEqual(len(dispatcher.childlog.data), 1)
        self.assertEqual(dispatcher.childlog.data[0],
                         'supercalifragilisticexpialidocious')

def test_handle_listener_state_change_from_unknown(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.UNKNOWN
        dispatcher.state_buffer = 'whatever'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data, [])
        self.assertEqual(process.listener_state, EventListenerStates.UNKNOWN)

def test_handle_listener_state_change_acknowledged_to_ready(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.ACKNOWLEDGED
        dispatcher.state_buffer = 'READY\n'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data[0],
                         'process1: ACKNOWLEDGED -> READY')
        self.assertEqual(process.listener_state, EventListenerStates.READY)

def test_handle_listener_state_change_acknowledged_gobbles(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.ACKNOWLEDGED
        dispatcher.state_buffer = 'READY\ngarbage\n'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data[0],
                         'process1: ACKNOWLEDGED -> READY')
        self.assertEqual(options.logger.data[1],
                         'process1: READY -> UNKNOWN')
        self.assertEqual(process.listener_state, EventListenerStates.UNKNOWN)

def test_handle_listener_state_change_acknowledged_to_insufficient(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.ACKNOWLEDGED
        dispatcher.state_buffer = 'RE'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, 'RE')
        self.assertEqual(options.logger.data, [])
        self.assertEqual(process.listener_state,
                         EventListenerStates.ACKNOWLEDGED)

def test_handle_listener_state_change_acknowledged_to_unknown(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.ACKNOWLEDGED
        dispatcher.state_buffer = 'bogus data yo'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data[0],
                         'process1: ACKNOWLEDGED -> UNKNOWN')
        self.assertEqual(options.logger.data[1],
                         'process1: has entered the UNKNOWN state and will '
                         'no longer receive events, this usually indicates '
                         'the process violated the eventlistener protocol')
        self.assertEqual(process.listener_state, EventListenerStates.UNKNOWN)

def test_handle_listener_state_change_ready_to_unknown(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.READY
        dispatcher.state_buffer = 'bogus data yo'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data[0],
                         'process1: READY -> UNKNOWN')
        self.assertEqual(options.logger.data[1],
                         'process1: has entered the UNKNOWN state and will '
                         'no longer receive events, this usually indicates '
                         'the process violated the eventlistener protocol')
        self.assertEqual(process.listener_state, EventListenerStates.UNKNOWN)

def test_handle_listener_state_change_busy_to_insufficient(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.BUSY
        dispatcher.state_buffer = 'bogus data yo'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, 'bogus data yo')
        self.assertEqual(process.listener_state, EventListenerStates.BUSY)

def test_handle_listener_state_change_busy_to_acknowledged_procd(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.BUSY
        class Dummy:
            pass
        process.group = Dummy()
        process.group.config = Dummy()
        from supervisor.dispatchers import default_handler
        process.group.config.result_handler = default_handler
        dispatcher.state_buffer = 'RESULT 2\nOKabc'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, 'abc')
        self.assertEqual(options.logger.data[0],
                         'process1: event was processed')
        self.assertEqual(options.logger.data[1],
                         'process1: BUSY -> ACKNOWLEDGED')
        self.assertEqual(process.listener_state,
                         EventListenerStates.ACKNOWLEDGED)

def test_handle_listener_state_change_busy_to_acknowledged_rejected(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.BUSY
        class Dummy:
            pass
        process.group = Dummy()
        process.group.config = Dummy()
        from supervisor.dispatchers import default_handler
        process.group.config.result_handler = default_handler
        dispatcher.state_buffer = 'RESULT 4\nFAILabc'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, 'abc')
        self.assertEqual(options.logger.data[0],
                         'process1: event was rejected')
        self.assertEqual(options.logger.data[1],
                         'process1: BUSY -> ACKNOWLEDGED')
        self.assertEqual(process.listener_state,
                         EventListenerStates.ACKNOWLEDGED)

def test_handle_listener_state_change_busy_to_unknown(self):
        from supervisor.events import EventRejectedEvent
        from supervisor.events import subscribe
        events = []
        def doit(event):
            events.append(event)
        subscribe(EventRejectedEvent, doit)
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.BUSY
        current_event = DummyEvent()
        process.event = current_event
        dispatcher.state_buffer = 'bogus data\n'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data[0],
                "process1: bad result line: 'bogus data'")
        self.assertEqual(options.logger.data[1],
                'process1: BUSY -> UNKNOWN')
        self.assertEqual(options.logger.data[2],
                         'process1: has entered the UNKNOWN state and will '
                         'no longer receive events, this usually indicates '
                         'the process violated the eventlistener protocol')
        self.assertEqual(process.listener_state,
                         EventListenerStates.UNKNOWN)
        self.assertEqual(events[0].process, process)
        self.assertEqual(events[0].event, current_event)

def test_handle_listener_state_busy_gobbles(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        process.listener_state = EventListenerStates.BUSY
        class Dummy:
            pass
        process.group = Dummy()
        process.group.config = Dummy()
        from supervisor.dispatchers import default_handler
        process.group.config.result_handler = default_handler
        dispatcher.state_buffer = 'RESULT 2\nOKbogus data\n'
        self.assertEqual(dispatcher.handle_listener_state_change(), None)
        self.assertEqual(dispatcher.state_buffer, '')
        self.assertEqual(options.logger.data[0],
                         'process1: event was processed')
        self.assertEqual(options.logger.data[1],
                         'process1: BUSY -> ACKNOWLEDGED')
        self.assertEqual(options.logger.data[2],
                         'process1: ACKNOWLEDGED -> UNKNOWN')
        self.assertEqual(options.logger.data[3],
                         'process1: has entered the UNKNOWN state and will '
                         'no longer receive events, this usually indicates '
                         'the process violated the eventlistener protocol')
        self.assertEqual(process.listener_state,
                         EventListenerStates.UNKNOWN)

def test_handle_result_accept(self):
        from supervisor.events import subscribe
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        L = []
        def doit(event):
            L.append(event)
        from supervisor import events
        subscribe(events.EventRejectedEvent, doit)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        def handle(event, result):
            pass
        class Dummy:
            pass
        process.group = Dummy()
        process.group.config = Dummy()
        process.group.config.result_handler = handle
        process.listener_state = EventListenerStates.BUSY
        dispatcher.handle_result('foo')
        self.assertEqual(len(L), 0)
        self.assertEqual(process.listener_state,
                         EventListenerStates.ACKNOWLEDGED)
        self.assertEqual(options.logger.data[0],
                         'process1: event was processed')
        self.assertEqual(options.logger.data[1],
                         'process1: BUSY -> ACKNOWLEDGED')

def test_handle_result_rejectevent(self):
        from supervisor.events import subscribe
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        L = []
        def doit(event):
            L.append(event)
        from supervisor import events
        subscribe(events.EventRejectedEvent, doit)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        def rejected(event, result):
            from supervisor.dispatchers import RejectEvent
            raise RejectEvent(result)
        class Dummy:
            pass
        process.group = Dummy()
        process.group.config = Dummy()
        process.group.config.result_handler = rejected
        process.listener_state = EventListenerStates.BUSY
        dispatcher.handle_result('foo')
        self.assertEqual(len(L), 1)
        self.assertEqual(L[0].__class__, events.EventRejectedEvent)
        self.assertEqual(process.listener_state,
                         EventListenerStates.ACKNOWLEDGED)
        self.assertEqual(options.logger.data[0],
                         'process1: event was rejected')
        self.assertEqual(options.logger.data[1],
                         'process1: BUSY -> ACKNOWLEDGED')

def test_handle_result_exception(self):
        from supervisor.events import subscribe
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        L = []
        def doit(event):
            L.append(event)
        from supervisor import events
        subscribe(events.EventRejectedEvent, doit)
        from supervisor.dispatchers import EventListenerStates
        dispatcher = self._makeOne(process)
        def exception(event, result):
            raise ValueError
        class Dummy:
            pass
        process.group = Dummy()
        process.group.config = Dummy()
        process.group.config.result_handler = exception
        process.group.result_handler = exception
        process.listener_state = EventListenerStates.BUSY
        dispatcher.handle_result('foo')
        self.assertEqual(len(L), 1)
        self.assertEqual(L[0].__class__, events.EventRejectedEvent)
        self.assertEqual(process.listener_state,
                         EventListenerStates.UNKNOWN)
        self.assertEqual(options.logger.data[0],
                         'process1: event caused an error')
        self.assertEqual(options.logger.data[1],
                         'process1: BUSY -> UNKNOWN')
        self.assertEqual(options.logger.data[2],
                         'process1: has entered the UNKNOWN state and will '
                         'no longer receive events, this usually indicates '
                         'the process violated the eventlistener protocol')

def test_removelogs(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1',
                              stdout_logfile='/tmp/foo')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        dispatcher.removelogs()
        self.assertEqual(dispatcher.childlog.handlers[0].reopened, True)
        self.assertEqual(dispatcher.childlog.handlers[0].removed, True)

options.readfd_result = ansi
        dispatcher.handle_read_event()
        self.assertEqual(len(dispatcher.childlog.data), 1)
        self.assertEqual(dispatcher.childlog.data[0], noansi)

options.strip_ansi = False

options.readfd_result = ansi
        dispatcher.handle_read_event()
        self.assertEqual(len(dispatcher.childlog.data), 2)
        self.assertEqual(dispatcher.childlog.data[1], ansi)

def test_repr(self):
        options = DummyOptions()
        config = DummyPConfig(options, 'process1', '/bin/process1')
        process = DummyProcess(config)
        dispatcher = self._makeOne(process)
        drepr = repr(dispatcher)
        self.assertTrue(drepr.startswith('<PEventListenerDispatcher at'), drepr)
        self.assertNotEqual(
            drepr.find('<supervisor.tests.base.DummyProcess instance at'),
            -1)
        self.assertTrue(drepr.endswith('(stdout)>'), drepr)

class stripEscapeTests(unittest.TestCase):
    def _callFUT(self, s):
        from supervisor.dispatchers import stripEscapes
        return stripEscapes(s)

def test_zero_length_string(self):
        self.assertEqual(self._callFUT(''), '')

def test_ansi(self):
        ansi = '\x1b[34mHello world... this is longer than a token!\x1b[0m'
        noansi = 'Hello world... this is longer than a token!'
        self.assertEqual(self._callFUT(ansi), noansi)

def test_noansi(self):
        noansi = 'Hello world... this is longer than a token!'
        self.assertEqual(self._callFUT(noansi), noansi)

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

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

${this.title}

Code Editor : test_dispatchers.py