PHP Dosya Yöneticisi

# ~*~ coding: utf-8 ~*~
from __future__ import unicode_literals

import os
import sys
import signal
import unittest
import pkg_resources
from supervisor.compat import xmlrpclib
from supervisor.xmlrpc import SupervisorTransport

# end-to-test tests are slow so only run them when asked
if 'END_TO_END' in os.environ:
    import pexpect
    BaseTestCase = unittest.TestCase
else:
    BaseTestCase = object

class EndToEndTests(BaseTestCase):

def test_issue_550(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-550.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('success: print_env entered RUNNING state')
        supervisord.expect_exact('exited: print_env (exit status 0; expected)')

args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'tail -100000', 'print_env']
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisorctl.kill, signal.SIGINT)
        supervisorctl.expect_exact("THIS_SHOULD=BE_IN_CHILD_ENV")
        supervisorctl.expect(pexpect.EOF)

def test_issue_565(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-565.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('success: hello entered RUNNING state')

args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'tail', '-f', 'hello']
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisorctl.kill, signal.SIGINT)

for i in range(1, 4):
            line = 'The Øresund bridge ends in Malmö - %d' % i
            supervisorctl.expect_exact(line, timeout=30)

def test_issue_638(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-638.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        is_py2 = sys.version_info[0] < 3
        if is_py2:
            b_prefix = ''
        else:
            b_prefix = 'b'
        supervisord.expect_exact(r"Undecodable: %s'\x88\n'" % b_prefix, timeout=30)
        supervisord.expect('received SIGCH?LD indicating a child quit', timeout=30)
        if is_py2:
            # need to investigate why this message is only printed under 2.x
            supervisord.expect_exact('gave up: produce-unicode-error entered FATAL state, '
                                     'too many start retries too quickly', timeout=60)

def test_issue_663(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-663.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        for i in range(2):
            supervisord.expect_exact('OKREADY', timeout=60)
            supervisord.expect_exact('BUSY -> ACKNOWLEDGED', timeout=30)

def test_issue_664(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-664.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('test_öäü entered RUNNING state', timeout=60)
        args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'status']
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisorctl.kill, signal.SIGINT)
        try:
            supervisorctl.expect('test_öäü\\s+RUNNING', timeout=30)
            seen = True
        except pexpect.ExceptionPexpect:
            seen = False
        self.assertTrue(seen)

def test_issue_835(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-835.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('cat entered RUNNING state', timeout=60)
        transport = SupervisorTransport('', '', 'unix:///tmp/issue-835.sock')
        server = xmlrpclib.ServerProxy('http://anything/RPC2', transport)
        try:
            for s in ('The Øresund bridge ends in Malmö', 'hello'):
                result = server.supervisor.sendProcessStdin('cat', s)
                self.assertTrue(result)
                supervisord.expect_exact(s, timeout=30)
        finally:
            transport.connection.close()

def test_issue_836(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-836.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('cat entered RUNNING state', timeout=60)
        args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'fg', 'cat']
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisorctl.kill, signal.SIGINT)

try:
            for s in ('Hi', 'Hello', 'The Øresund bridge ends in Malmö'):
                supervisorctl.sendline(s)
                supervisord.expect_exact(s, timeout=60)
                supervisorctl.expect_exact(s) # echoed locally
                supervisorctl.expect_exact(s) # sent back by supervisord
            seen = True
        except pexpect.ExceptionPexpect:
            seen = False
        self.assertTrue(seen)

def test_issue_1054(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1054.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('cat entered RUNNING state', timeout=60)
        args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'avail']
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8')
        try:
            supervisorctl.expect('cat\\s+in use\\s+auto', timeout=30)
            seen = True
        except pexpect.ExceptionPexpect:
            seen = False
        self.assertTrue(seen)

def test_issue_1170a(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1170a.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact("set from [supervisord] section")

def test_issue_1170b(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1170b.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact("set from [program] section")

def test_issue_1170c(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1170c.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact("set from [eventlistener] section")

def test_issue_1224(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1224.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('cat entered RUNNING state', timeout=60)

def test_issue_1231a(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1231a.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('success: hello entered RUNNING state')

for i in range(1, 4):
            line = '%d - hash=57d94b…381088' % i
            supervisorctl.expect_exact(line, timeout=30)

def test_issue_1231b(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1231b.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('success: hello entered RUNNING state')

args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'tail', '-f', 'hello']
        env = os.environ.copy()
        env['LANG'] = 'oops'
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8',
                                      env=env)
        self.addCleanup(supervisorctl.kill, signal.SIGINT)

# For Python 3 < 3.7, LANG=oops leads to warnings because of the
        # stdout encoding. For 3.7 (and presumably later), the encoding is
        # utf-8 when LANG=oops.
        if sys.version_info[:2] < (3, 7):
            supervisorctl.expect('Warning: sys.stdout.encoding is set to ',
                                 timeout=30)
            supervisorctl.expect('Unicode output may fail.', timeout=30)

for i in range(1, 4):
            line = '%d - hash=57d94b…381088' % i
            try:
                supervisorctl.expect_exact(line, timeout=30)
            except pexpect.exceptions.EOF:
                self.assertIn('Unable to write Unicode to stdout because it '
                              'has encoding ',
                              supervisorctl.before)
                break

def test_issue_1231c(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1231c.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('success: hello entered RUNNING state')

args = ['-m', 'supervisor.supervisorctl', '-c', filename, 'tail', 'hello']
        env = os.environ.copy()
        env['LANG'] = 'oops'
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8',
                                      env=env)
        self.addCleanup(supervisorctl.kill, signal.SIGINT)

def test_issue_1251(self):
        args = ['-m', 'supervisor.supervisord', '-?']
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact("supervisord -- run a set of applications")
        supervisord.expect_exact("-l/--logfile FILENAME -- use FILENAME as")
        supervisord.expect(pexpect.EOF)

args = ['-m', 'supervisor.supervisorctl', '-?']
        supervisorctl = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisorctl.kill, signal.SIGINT)
        supervisorctl.expect_exact("supervisorctl -- control applications")
        supervisorctl.expect_exact("-i/--interactive -- start an interactive")
        supervisorctl.expect(pexpect.EOF)

def test_issue_1298(self):
        filename = pkg_resources.resource_filename(__name__, 'fixtures/issue-1298.conf')
        args = ['-m', 'supervisor.supervisord', '-c', filename]
        supervisord = pexpect.spawn(sys.executable, args, encoding='utf-8')
        self.addCleanup(supervisord.kill, signal.SIGINT)
        supervisord.expect_exact('success: spew entered RUNNING state')

cmd = "'%s' -m supervisor.supervisorctl -c '%s' tail -f spew | /bin/cat -u" % (
            sys.executable, filename
            )
        bash = pexpect.spawn('/bin/sh', ['-c', cmd], encoding='utf-8')
        self.addCleanup(bash.kill, signal.SIGINT)
        bash.expect('spewage 2', timeout=30)

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

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

İptal

📁 PHP Dosya Yöneticisi

📝 test_end_to_end.py

📝 test_end_to_end.py

📝 Yeniden Adlandır

🔐 Dosya İzinleri (chmod)