This is an automated email from the ASF dual-hosted git repository.
astitcher pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/qpid-proton.git
The following commit(s) were added to refs/heads/master by this push:
new 8a0d586 PROTON-2099: [Python] fix example runner to work with python2
- Make the tests a whole lot more readable while we're at it.
8a0d586 is described below
commit 8a0d586357e2e9fdb00dacaf12d6129720b01e67
Author: Andrew Stitcher <[email protected]>
AuthorDate: Thu Dec 5 15:55:58 2019 -0500
PROTON-2099: [Python] fix example runner to work with python2
- Make the tests a whole lot more readable while we're at it.
---
python/examples/test_examples.py | 78 +++++++++++++++++++++++-----------------
1 file changed, 45 insertions(+), 33 deletions(-)
diff --git a/python/examples/test_examples.py b/python/examples/test_examples.py
index c33156d..04ed8c6 100644
--- a/python/examples/test_examples.py
+++ b/python/examples/test_examples.py
@@ -23,14 +23,42 @@ import time
import unittest
+class Popen(subprocess.Popen):
+
+ # We always use these options
+ def __init__(self, args, **kwargs):
+ super(Popen, self).\
+ __init__(args,
+ stderr=subprocess.STDOUT,
+ stdout=subprocess.PIPE,
+ universal_newlines=True, **kwargs)
+
+ # For Python 2 compatibility add context manager support to Popen if it's
not there
+ if not hasattr(subprocess.Popen, '__enter__'):
+ def __enter__(self):
+ return self
+
+ # For Python 2 compatibility add context manager support to Popen if it's
not there
+ if not hasattr(subprocess.Popen, '__exit__'):
+ def __exit__(self, exc_type, exc_val, exc_tb):
+ try:
+ if self.stdin:
+ self.stdin.close()
+ if self.stdout:
+ self.stdout.close()
+ if self.stderr:
+ self.stderr.close()
+ finally:
+ self.wait()
+
+
def remove_unicode_prefix(line):
return re.sub(r"u(['\"])", r"\1", line)
class ExamplesTest(unittest.TestCase):
def test_helloworld(self, example="helloworld.py"):
- with subprocess.Popen([example], stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True) as p:
+ with Popen([example]) as p:
p.wait()
output = [l.strip() for l in p.stdout]
self.assertEqual(output, ['Hello World!'])
@@ -48,22 +76,18 @@ class ExamplesTest(unittest.TestCase):
self.test_helloworld('helloworld_direct_tornado.py')
def test_simple_send_recv(self, recv='simple_recv.py',
send='simple_send.py'):
- with subprocess.Popen([recv], stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True) as r:
- with subprocess.Popen([send], stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True):
+ with Popen([recv]) as r:
+ with Popen([send]):
pass
actual = [remove_unicode_prefix(l.strip()) for l in r.stdout]
expected = ["{'sequence': %i}" % (i+1,) for i in range(100)]
self.assertEqual(actual, expected)
def test_client_server(self, client=['client.py'], server=['server.py'],
sleep=0):
- with subprocess.Popen(server, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True) as s:
+ with Popen(server) as s:
if sleep:
time.sleep(sleep)
- with subprocess.Popen(client, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True) as c:
+ with Popen(client) as c:
c.wait()
actual = [l.strip() for l in c.stdout]
inputs = ["Twas brillig, and the slithy toves",
@@ -94,18 +118,14 @@ class ExamplesTest(unittest.TestCase):
# setup databases
subprocess.check_call(['db_ctrl.py', 'init', './src_db'])
subprocess.check_call(['db_ctrl.py', 'init', './dst_db'])
- with subprocess.Popen(['db_ctrl.py', 'insert', './src_db'],
- stdin=subprocess.PIPE, stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True) as fill:
+ with Popen(['db_ctrl.py', 'insert', './src_db'],
stdin=subprocess.PIPE) as fill:
for i in range(100):
fill.stdin.write("Message-%i\n" % (i+1))
fill.stdin.close()
# run send and recv
- with subprocess.Popen(['db_recv.py', '-m', '100'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True) as r:
- with subprocess.Popen(['db_send.py', '-m', '100'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True):
+ with Popen(['db_recv.py', '-m', '100']) as r:
+ with Popen(['db_send.py', '-m', '100']):
pass
r.wait()
# verify output of receive
@@ -114,8 +134,7 @@ class ExamplesTest(unittest.TestCase):
self.assertEqual(actual, expected)
# verify state of databases
- with subprocess.Popen(['db_ctrl.py', 'list', './dst_db'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True) as v:
+ with Popen(['db_ctrl.py', 'list', './dst_db']) as v:
v.wait()
expected = ["(%i, 'Message-%i')" % (i+1, i+1) for i in range(100)]
actual = [remove_unicode_prefix(l.strip()) for l in v.stdout]
@@ -126,11 +145,9 @@ class ExamplesTest(unittest.TestCase):
def test_simple_send_direct_recv(self):
self.maxDiff = None
- with subprocess.Popen(['direct_recv.py', '-a', 'localhost:8888'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True) as r:
+ with Popen(['direct_recv.py', '-a', 'localhost:8888']) as r:
time.sleep(0.5)
- with subprocess.Popen(['simple_send.py', '-a', 'localhost:8888'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True):
+ with Popen(['simple_send.py', '-a', 'localhost:8888']):
pass
r.wait()
actual = [remove_unicode_prefix(l.strip()) for l in r.stdout]
@@ -138,30 +155,25 @@ class ExamplesTest(unittest.TestCase):
self.assertEqual(actual, expected)
def test_direct_send_simple_recv(self):
- with subprocess.Popen(['direct_send.py', '-a', 'localhost:8888'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True):
+ with Popen(['direct_send.py', '-a', 'localhost:8888']):
time.sleep(0.5)
- with subprocess.Popen(['simple_recv.py', '-a', 'localhost:8888'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True) as r:
+ with Popen(['simple_recv.py', '-a', 'localhost:8888']) as r:
r.wait()
actual = [remove_unicode_prefix(l.strip()) for l in r.stdout]
expected = ["{'sequence': %i}" % (i+1,) for i in range(100)]
self.assertEqual(actual, expected)
def test_selected_recv(self):
- with subprocess.Popen(['colour_send.py'], stderr=subprocess.STDOUT,
stdout=subprocess.PIPE,
- universal_newlines=True):
+ with Popen(['colour_send.py']):
pass
- with subprocess.Popen(['selected_recv.py', '-m', '50'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True) as r:
+ with Popen(['selected_recv.py', '-m', '50']) as r:
r.wait()
actual = [l.strip() for l in r.stdout]
expected = ["green %i" % (i+1) for i in range(100) if i % 2 == 0]
self.assertEqual(actual, expected)
- with subprocess.Popen(['simple_recv.py', '-m', '50'],
stderr=subprocess.STDOUT, stdout=subprocess.PIPE,
- universal_newlines=True) as r:
+ with Popen(['simple_recv.py', '-m', '50']) as r:
r.wait()
actual = [l.strip() for l in r.stdout]
expected = ["red %i" % (i+1) for i in range(100) if i % 2 == 1]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
