Edit File: test_end_to_end.py
# ~*~ 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')
Back to File Manager