Gecko [ tsuru-no-tsurugi.tokyo ]

Name	Size	Permission	Date
__pycache__	[ DIR ]	drwxr-xr-x	2025-02-27 19:58
fixtures	[ DIR ]	drwxr-xr-x	2025-02-27 19:58
__init__.py	20 B	-rw-r--r--	2017-07-25 03:57
base.py	35.16 KB	-rw-r--r--	2020-08-07 05:05
test_childutils.py	5.35 KB	-rw-r--r--	2019-04-06 06:19
test_confecho.py	549 B	-rw-r--r--	2019-04-06 06:19
test_datatypes.py	26.61 KB	-rw-r--r--	2019-04-06 06:19
test_dispatchers.py	49.02 KB	-rw-r--r--	2019-04-06 06:19
test_end_to_end.py	10.47 KB	-rw-r--r--	2020-02-16 04:41
test_events.py	20.43 KB	-rw-r--r--	2019-04-06 06:19
test_http.py	24.82 KB	-rw-r--r--	2019-11-28 04:11
test_http_client.py	13.5 KB	-rw-r--r--	2019-04-06 06:19
test_loggers.py	20.94 KB	-rw-r--r--	2019-10-20 02:06
test_options.py	136.94 KB	-rw-r--r--	2020-04-10 05:49
test_poller.py	16.3 KB	-rw-r--r--	2019-04-06 06:19
test_process.py	97.58 KB	-rw-r--r--	2019-04-11 06:22
test_rpcinterfaces.py	101.81 KB	-rw-r--r--	2019-04-06 06:19
test_socket_manager.py	8.39 KB	-rw-r--r--	2020-08-07 06:07
test_states.py	2.22 KB	-rw-r--r--	2017-07-25 03:57
test_supervisorctl.py	80.15 KB	-rw-r--r--	2019-06-17 02:45
test_supervisord.py	33.72 KB	-rw-r--r--	2019-12-10 02:49
test_templating.py	62.18 KB	-rw-r--r--	2019-09-17 02:23
test_web.py	7 KB	-rw-r--r--	2019-04-06 06:19
test_xmlrpc.py	34.96 KB	-rw-r--r--	2019-04-06 06:19

Rename

from io import BytesIO
import sys
import time
import unittest
from supervisor.compat import StringIO
from supervisor.compat import as_string

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 = BytesIO()
        pcomm.send(b'hello', stdout)
        from supervisor.events import ProcessCommunicationEvent
        begin = ProcessCommunicationEvent.BEGIN_TOKEN
        end = ProcessCommunicationEvent.END_TOKEN
        self.assertEqual(stdout.getvalue(), begin + b'hello' + end)

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

def test_stderr(self):
        from supervisor.childutils import pcomm
        old = sys.stderr
        try:
            io = sys.stderr = BytesIO()
            pcomm.stderr(b'hello')
            from supervisor.events import ProcessCommunicationEvent
            begin = ProcessCommunicationEvent.BEGIN_TOKEN
            end = ProcessCommunicationEvent.END_TOKEN
            self.assertEqual(io.getvalue(), begin + b'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 = as_string(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 = as_string(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 = as_string(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 = as_string(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')