Server : Apache System : Linux server2.corals.io 4.18.0-348.2.1.el8_5.x86_64 #1 SMP Mon Nov 15 09:17:08 EST 2021 x86_64 User : corals ( 1002) PHP Version : 7.4.33 Disable Function : exec,passthru,shell_exec,system Directory : /lib/python3.6/site-packages/supervisor/tests/ |
# ~*~ 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')
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 = '%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)
# 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)
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')