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_childutils.py

import sys
import time
import unittest
from StringIO import StringIO

class ChildUtilsTests(unittest.TestCase):
    def test_getRPCInterface(self):
        from supervisor.childutils import getRPCInterface
        rpc = getRPCInterface({'SUPERVISOR_SERVER_URL':'http://localhost:9001'})
        # we can't really test this thing; its a magic object
        self.assertTrue(rpc is not None)

def test_getRPCTransport_no_uname_pass(self):
        from supervisor.childutils import getRPCTransport
        t = getRPCTransport({'SUPERVISOR_SERVER_URL':'http://localhost:9001'})
        self.assertEqual(t.username, '')
        self.assertEqual(t.password, '')
        self.assertEqual(t.serverurl, 'http://localhost:9001')

def test_getRPCTransport_with_uname_pass(self):
        from supervisor.childutils import getRPCTransport
        env = {'SUPERVISOR_SERVER_URL':'http://localhost:9001',
               'SUPERVISOR_USERNAME':'chrism',
               'SUPERVISOR_PASSWORD':'abc123'}
        t = getRPCTransport(env)
        self.assertEqual(t.username, 'chrism')
        self.assertEqual(t.password, 'abc123')
        self.assertEqual(t.serverurl, 'http://localhost:9001')

def test_get_headers(self):
        from supervisor.childutils import get_headers
        line = 'a:1 b:2'
        result = get_headers(line)
        self.assertEqual(result, {'a':'1', 'b':'2'})

def test_eventdata(self):
        from supervisor.childutils import eventdata
        payload = 'a:1 b:2\nthedata\n'
        headers, data = eventdata(payload)
        self.assertEqual(headers, {'a':'1', 'b':'2'})
        self.assertEqual(data, 'thedata\n')

def test_get_asctime(self):
        from supervisor.childutils import get_asctime
        timestamp = time.mktime((2009, 1, 18, 22, 14, 7, 0, 0, -1))
        result = get_asctime(timestamp)
        self.assertEqual(result, '2009-01-18 22:14:07,000')

class TestProcessCommunicationsProtocol(unittest.TestCase):
    def test_send(self):
        from supervisor.childutils import pcomm
        stdout = StringIO()
        pcomm.send('hello', stdout)
        from supervisor.events import ProcessCommunicationEvent
        begin = ProcessCommunicationEvent.BEGIN_TOKEN
        end = ProcessCommunicationEvent.END_TOKEN
        self.assertEqual(stdout.getvalue(), '%s%s%s' % (begin, 'hello', end))

def test_stdout(self):
        from supervisor.childutils import pcomm
        old = sys.stdout
        try:
            io = sys.stdout = StringIO()
            pcomm.stdout('hello')
            from supervisor.events import ProcessCommunicationEvent
            begin = ProcessCommunicationEvent.BEGIN_TOKEN
            end = ProcessCommunicationEvent.END_TOKEN
            self.assertEqual(io.getvalue(), '%s%s%s' % (begin, 'hello', end))
        finally:
            sys.stdout = old

def test_stderr(self):
        from supervisor.childutils import pcomm
        old = sys.stderr
        try:
            io = sys.stderr = StringIO()
            pcomm.stderr('hello')
            from supervisor.events import ProcessCommunicationEvent
            begin = ProcessCommunicationEvent.BEGIN_TOKEN
            end = ProcessCommunicationEvent.END_TOKEN
            self.assertEqual(io.getvalue(), '%s%s%s' % (begin, 'hello', end))
        finally:
            sys.stderr = old

class TestEventListenerProtocol(unittest.TestCase):
    def test_wait(self):
        from supervisor.childutils import listener
        class Dummy:
            def readline(self):
                return 'len:5'
            def read(self, *ignored):
                return 'hello'
        stdin = Dummy()
        stdout = StringIO()
        headers, payload = listener.wait(stdin, stdout)
        self.assertEqual(headers, {'len':'5'})
        self.assertEqual(payload, 'hello')
        self.assertEqual(stdout.getvalue(), 'READY\n')

def test_token(self):
        from supervisor.childutils import listener
        from supervisor.dispatchers import PEventListenerDispatcher
        token = PEventListenerDispatcher.READY_FOR_EVENTS_TOKEN
        stdout = StringIO()
        listener.ready(stdout)
        self.assertEqual(stdout.getvalue(), token)

def test_ok(self):
        from supervisor.childutils import listener
        from supervisor.dispatchers import PEventListenerDispatcher
        begin = PEventListenerDispatcher.RESULT_TOKEN_START
        stdout = StringIO()
        listener.ok(stdout)
        self.assertEqual(stdout.getvalue(), begin + '2\nOK')

def test_fail(self):
        from supervisor.childutils import listener
        from supervisor.dispatchers import PEventListenerDispatcher
        begin = PEventListenerDispatcher.RESULT_TOKEN_START
        stdout = StringIO()
        listener.fail(stdout)
        self.assertEqual(stdout.getvalue(), begin + '4\nFAIL')

def test_send(self):
        from supervisor.childutils import listener
        from supervisor.dispatchers import PEventListenerDispatcher
        begin = PEventListenerDispatcher.RESULT_TOKEN_START
        stdout = StringIO()
        msg = 'the body data ya fool\n'
        listener.send(msg, stdout)
        expected = '%s%s\n%s' % (begin, len(msg), msg)
        self.assertEqual(stdout.getvalue(), expected)

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

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

${this.title}

Code Editor : test_childutils.py