Dre4m Shell
Server IP : 85.214.239.14  /  Your IP : 18.225.55.42
Web Server : Apache/2.4.62 (Debian)
System : Linux h2886529.stratoserver.net 4.9.0 #1 SMP Tue Jan 9 19:45:01 MSK 2024 x86_64
User : www-data ( 33)
PHP Version : 7.4.18
Disable Function : pcntl_alarm,pcntl_fork,pcntl_waitpid,pcntl_wait,pcntl_wifexited,pcntl_wifstopped,pcntl_wifsignaled,pcntl_wifcontinued,pcntl_wexitstatus,pcntl_wtermsig,pcntl_wstopsig,pcntl_signal,pcntl_signal_get_handler,pcntl_signal_dispatch,pcntl_get_last_error,pcntl_strerror,pcntl_sigprocmask,pcntl_sigwaitinfo,pcntl_sigtimedwait,pcntl_exec,pcntl_getpriority,pcntl_setpriority,pcntl_async_signals,pcntl_unshare,
MySQL : OFF  |  cURL : OFF  |  WGET : ON  |  Perl : ON  |  Python : ON  |  Sudo : ON  |  Pkexec : OFF
Directory :  /proc/3/root/proc/self/root/lib/python3/dist-packages/supervisor/tests/

Upload File :
current_dir [ Writeable ] document_root [ Writeable ]

 

Command :


[ HOME SHELL ]     

Current File : /proc/3/root/proc/self/root/lib/python3/dist-packages/supervisor/tests/test_childutils.py
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')

Anon7 - 2022
AnonSec Team