From ab2229451827f530959d554920619d87daa34586 Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Jan=20Pokorn=C3=BD?= <jpokorny@redhat.com>
Date: Wed, 11 Jul 2018 16:18:25 +0200
Subject: [PATCH] test: make Python files supported _also_ with Python 3.3+
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit
- use "print" like a function rather than a statement
. where implicit newline is to be suppressed, don't do that
and rather strip it from the string to be printed instead
- use 2+3 compatible convention for parametrizing exceptions
- Python 3 doesn't recognize "basestring" class, and at the place
of use (pre Python 2.7 only), unicode string is really not expected
(also re.UNICODE flag is not used...)
- Python 3 doesn't recognize "xrange" function, but the surrounding
code can be reasonably simplified using "enumerate" function
- arrange dict treatment in a compatible way:
. d.has_key(k) -> k in d
. d.iteritems() -> custom "iter_items", always efficient wrapper
. d.iterkeys(), here incl. lazy mapping and filtering
-> rewrite while retaining laziness
. optimize UT.merge_dicts in script/unit-test.py along
- also in three instances, deal with string/uninterpreted bytes proper
dichotomy introduced in Python 3, and related to that, "string"
module only supports "ascii_lowercase" attribute in Python 3
(as opposed to system-specific plain "lowercase" one)
Note that script/unit-test.py has a pre-existing issue (regardless
of which Python version interpreter is used), so at least document
that in the header for now.
Signed-off-by: Jan Pokorný <jpokorny@redhat.com>
---
script/unit-test.py | 65 ++++++++++++++++++++++++++------------------
test/assertions.py | 4 +--
test/boothrunner.py | 32 ++++++++++++----------
test/boothtestenv.py | 6 ++--
test/clienttests.py | 4 +--
test/runtests.py | 2 +-
test/serverenv.py | 18 ++++++------
test/servertests.py | 10 +++----
test/utils.py | 10 +++++--
9 files changed, 84 insertions(+), 67 deletions(-)
diff --git a/script/unit-test.py b/script/unit-test.py
index 6871930..399528e 100755
--- a/script/unit-test.py
+++ b/script/unit-test.py
@@ -1,6 +1,8 @@
#!/usr/bin/python
# vim: fileencoding=utf-8
# see http://stackoverflow.com/questions/728891/correct-way-to-define-python-source-code-encoding
+# NOTE: setting the encoding is needed as non-ASCII characters are contained
+# FIXME: apparently, pexpect.EOF is not being excepted properly
import os, sys, time, signal, tempfile, socket, posix, time
import re, shutil, pexpect, logging, pprint
@@ -16,6 +18,16 @@ default_log_format = '%(asctime)s: : %(message)s'
default_log_datefmt = '%b %d %H:%M:%S'
+# Compatibility with dictionary methods not present in Python 3;
+# https://www.python.org/dev/peps/pep-0469/#migrating-to-the-common-subset-of-python-2-and-3
+try:
+ dict.iteritems
+except AttributeError: # Python 3
+ iter_items = lambda d: iter(d.items())
+else: # Python 2
+ iter_items = lambda d: d.iteritems()
+
+
# {{{ pexpect-logging glue
# needed for use as pexpect.logfile, to relay into existing logfiles
class expect_logging():
@@ -28,9 +40,12 @@ class expect_logging():
def flush(self, *arg):
pass
+
def write(self, stg):
if self.test.dont_log_expect == 0:
# TODO: split by input/output, give program
+ if sys.version_info[0] >= 3:
+ stg = str(stg, 'UTF-8')
for line in re.split(r"[\r\n]+", stg):
if line == self.test.prompt:
continue
@@ -110,7 +125,7 @@ class UT():
res = re.match(r"^\s*(\w+)\s*:(?:\s*(#.*?\S))?\s*$", line)
if res:
state = res.group(1)
- if not m.has_key(state):
+ if state not in m:
m[state] = dict_plus()
if res.group(2):
m[state].aux["comment"] = res.group(2)
@@ -188,17 +203,15 @@ class UT():
name = re.sub(r".*/", "", bin)
# How to get stderr, too?
expct = pexpect.spawn(bin,
- env = dict( os.environ.items() +
- [('PATH',
- self.test_base + "/bin/:" +
- os.getenv('PATH')),
- ('UNIT_TEST_PATH', self.test_base),
- ('LC_ALL', 'C'),
- ('LANG', 'C')] +
- env_add ),
- timeout = 30,
- maxread = 32768,
- **args)
+ env=dict(os.environ, **dict({
+ 'PATH': ':'.join((self.test_base + "/bin/",
+ os.getenv('PATH'))),
+ 'UNIT_TEST_PATH': self.test_base,
+ 'LC_ALL': 'C',
+ 'LANG': 'C'}, **dict(env_add))),
+ timeout=30,
+ maxread=32768,
+ **args)
expct.setecho(False)
expct.logfile_read = expect_logging("<- %s" % name, self)
expct.logfile_send = expect_logging(" -> %s" % name, self)
@@ -361,7 +374,7 @@ class UT():
self.current_nr = kv.aux.get("line")
#os.system("strace -f -tt -s 2000 -e write -p" + str(self.gdb.pid) + " &")
- for n, v in kv.iteritems():
+ for n, v in iter_items(kv):
self.set_val( self.translate_shorthand(n, "ticket"), v)
logging.info("set state")
@@ -372,7 +385,7 @@ class UT():
if not sys.stdin.isatty():
logging.error("Not a terminal, stopping.")
else:
- print "\n\nEntering interactive mode.\n\n"
+ print("\n\nEntering interactive mode.\n\n")
self.gdb.sendline("set prompt GDB> \n")
self.gdb.setecho(True)
# can't use send_cmd, doesn't reply with expected prompt anymore.
@@ -415,7 +428,7 @@ class UT():
self.send_cmd("next")
# push message.
- for (n, v) in msg.iteritems():
+ for (n, v) in iter_items(msg):
self.set_val( self.translate_shorthand(n, "message"), v, "htonl")
# set "received" length
@@ -426,7 +439,7 @@ class UT():
def wait_outgoing(self, msg):
self.wait_for_function("booth_udp_send")
ok = True
- for (n, v) in msg.iteritems():
+ for (n, v) in iter_items(msg):
if re.search(r"\.", n):
ok = self.check_value( self.translate_shorthand(n, "inject"), v) and ok
else:
@@ -438,14 +451,12 @@ class UT():
#stopped_at = self.sync()
def merge_dicts(self, base, overlay):
- return dict(base.items() + overlay.items())
+ return dict(base, **overlay)
def loop(self, fn, data):
- matches = map(lambda k: re.match(r"^(outgoing|message)(\d+)$", k), data.iterkeys())
- valid_matches = filter(None, matches)
- nums = map(lambda m: int(m.group(2)), valid_matches)
- loop_max = max(nums)
+ matches = (re.match(r"^(outgoing|message)(\d+)$", k) for k in data)
+ loop_max = max(int(m.group(2)) for m in matches if m)
for counter in range(0, loop_max+1): # incl. last message
kmsg = 'message%d' % counter
@@ -471,14 +482,14 @@ class UT():
logging.info("ticket change %s (%s:%d) %s" % (ktkt, fn, self.current_nr, comment))
self.set_state(tkt)
if gdb:
- for (k, v) in gdb.iteritems():
+ for (k, v) in iter_items(gdb):
self.send_cmd(k + " " + v.replace("§", "\n"))
if msg:
self.current_nr = msg.aux.get("line")
comment = msg.aux.get("comment", "")
logging.info("sending %s (%s:%d) %s" % (kmsg, fn, self.current_nr, comment))
self.send_message(self.merge_dicts(data["message"], msg))
- if data.has_key(kgdb) and len(gdb) == 0:
+ if kgdb in data and len(gdb) == 0:
self.user_debug("manual override")
if out:
self.current_nr = out.aux.get("line")
@@ -520,7 +531,7 @@ class UT():
self.let_booth_go_a_bit()
ok = True
- for (n, v) in data.iteritems():
+ for (n, v) in iter_items(data):
ok = self.check_value( self.translate_shorthand(n, "ticket"), v) and ok
if not ok:
sys.exit(1)
@@ -529,8 +540,8 @@ class UT():
def run(self, start_from="000", end_with="999"):
os.chdir(self.test_base)
# TODO: sorted, random order
- tests = filter( (lambda f: re.match(r"^\d\d\d_.*\.txt$", f)), glob.glob("*"))
- tests.sort()
+ tests = sorted(f for f in glob.glob("*")
+ if re.match(r"^\d\d\d_.*\.txt$", f))
failed = 0
for f in tests:
if f[0:3] < start_from:
@@ -561,7 +572,7 @@ class UT():
except:
failed += 1
logging.error(self.colored_string("Broke in %s:%s %s" % (f, self.current_nr, sys.exc_info()), self.RED))
- for frame in traceback.format_tb(sys.exc_traceback):
+ for frame in traceback.format_tb(sys.exc_info()[2]):
logging.info(" - %s " % frame.rstrip())
finally:
self.stop_processes()
diff --git a/test/assertions.py b/test/assertions.py
index 0b7f995..34333ca 100644
--- a/test/assertions.py
+++ b/test/assertions.py
@@ -21,7 +21,7 @@ class BoothAssertions:
# backported from 2.7 just in case we're running on an older Python
def assertRegexpMatches(self, text, expected_regexp, msg=None):
"""Fail the test unless the text matches the regular expression."""
- if isinstance(expected_regexp, basestring):
+ if isinstance(expected_regexp, str):
expected_regexp = re.compile(expected_regexp)
if not expected_regexp.search(text, MULTILINE):
msg = msg or "Regexp didn't match"
@@ -30,7 +30,7 @@ class BoothAssertions:
def assertNotRegexpMatches(self, text, unexpected_regexp, msg=None):
"""Fail the test if the text matches the regular expression."""
- if isinstance(unexpected_regexp, basestring):
+ if isinstance(unexpected_regexp, str):
unexpected_regexp = re.compile(unexpected_regexp)
match = unexpected_regexp.search(text)
if match:
diff --git a/test/boothrunner.py b/test/boothrunner.py
index d981183..347912b 100644
--- a/test/boothrunner.py
+++ b/test/boothrunner.py
@@ -1,4 +1,5 @@
import os
+import sys
import subprocess
import time
import unittest
@@ -37,14 +38,14 @@ class BoothRunner:
def show_output(self, stdout, stderr):
if stdout:
- print "STDOUT:"
- print "------"
- print stdout,
+ print("STDOUT:")
+ print("------")
+ print(stdout.rstrip('\n'))
if stderr:
- print "STDERR: (N.B. crm_ticket failures indicate daemon started correctly)"
- print "------"
- print stderr,
- print "-" * 70
+ print("STDERR: (N.B. crm_ticket failures indicate daemon started correctly)")
+ print("------")
+ print(stderr.rstrip('\n'))
+ print("-" * 70)
def subproc_completed_within(self, p, timeout):
start = time.time()
@@ -55,7 +56,7 @@ class BoothRunner:
elapsed = time.time() - start
if elapsed + wait > timeout:
wait = timeout - elapsed
- print "Waiting on %d for %.1fs ..." % (p.pid, wait)
+ print("Waiting on %d for %.1fs ..." % (p.pid, wait))
time.sleep(wait)
elapsed = time.time() - start
if elapsed >= timeout:
@@ -83,26 +84,29 @@ class BoothRunner:
return text
def show_args(self):
- print "\n"
- print "-" * 70
- print "Running", ' '.join(self.all_args())
+ print("\n")
+ print("-" * 70)
+ print("Running", ' '.join(self.all_args()))
msg = "with config from %s" % self.config_file_used()
config_text = self.config_text_used()
if config_text is not None:
msg += ": [%s]" % config_text
- print msg
+ print(msg)
def run(self):
p = subprocess.Popen(self.all_args(), stdout=subprocess.PIPE, stderr=subprocess.PIPE)
if not p:
- raise RuntimeError, "failed to start subprocess"
+ raise RuntimeError("failed to start subprocess")
- print "Started subprocess pid %d" % p.pid
+ print("Started subprocess pid %d" % p.pid)
completed = self.subproc_completed_within(p, 2)
if completed:
(stdout, stderr) = p.communicate()
+ if sys.version_info[0] >= 3:
+ # only expect ASCII/UTF-8 encodings for the obtained input bytes
+ stdout, stderr = str(stdout, 'UTF-8'), str(stderr, 'UTF-8')
self.show_output(stdout, stderr)
return (p.pid, p.returncode, stdout, stderr)
diff --git a/test/boothtestenv.py b/test/boothtestenv.py
index fcd0c4d..59e25c3 100644
--- a/test/boothtestenv.py
+++ b/test/boothtestenv.py
@@ -17,7 +17,7 @@ class BoothTestEnvironment(unittest.TestCase, BoothAssertions):
def setUp(self):
if not self._testMethodName.startswith('test_'):
- raise RuntimeError, "unexpected test method name: " + self._testMethodName
+ raise RuntimeError("unexpected test method name: " + self._testMethodName)
self.test_name = self._testMethodName[5:]
self.test_path = os.path.join(self.test_run_path, self.test_name)
os.makedirs(self.test_path)
@@ -54,11 +54,11 @@ class BoothTestEnvironment(unittest.TestCase, BoothAssertions):
def check_return_code(self, pid, return_code, expected_exitcode):
if return_code is None:
- print "pid %d still running" % pid
+ print("pid %d still running" % pid)
if expected_exitcode is not None:
self.fail("expected exit code %d, not long-running process" % expected_exitcode)
else:
- print "pid %d exited with code %d" % (pid, return_code)
+ print("pid %d exited with code %d" % (pid, return_code))
if expected_exitcode is None:
msg = "should not exit"
else:
diff --git a/test/clienttests.py b/test/clienttests.py
index c4b9d8a..512620e 100644
--- a/test/clienttests.py
+++ b/test/clienttests.py
@@ -7,14 +7,14 @@ class ClientConfigTests(ClientTestEnvironment):
def test_site_buffer_overflow(self):
# https://bugzilla.novell.com/show_bug.cgi?id=750256
- longfile = (string.lowercase * 3)[:63]
+ longfile = (string.ascii_lowercase * 3)[:63]
expected_error = "'%s' exceeds maximum site name length" % longfile
args = [ 'grant', '-s', longfile, '-t', 'ticket' ]
self._test_buffer_overflow(expected_error, args=args)
def test_ticket_buffer_overflow(self):
# https://bugzilla.novell.com/show_bug.cgi?id=750256
- longfile = (string.lowercase * 3)[:63]
+ longfile = (string.ascii_lowercase * 3)[:63]
expected_error = "'%s' exceeds maximum ticket name length" % longfile
args = [ 'grant', '-s', 'site', '-t', longfile ]
self._test_buffer_overflow(expected_error, args=args)
diff --git a/test/runtests.py b/test/runtests.py
index 0532c01..833b1a7 100755
--- a/test/runtests.py
+++ b/test/runtests.py
@@ -53,5 +53,5 @@ if __name__ == '__main__':
shutil.rmtree(test_run_path)
sys.exit(0)
else:
- print "Left %s for debugging" % test_run_path
+ print("Left %s for debugging" % test_run_path)
sys.exit(1)
diff --git a/test/serverenv.py b/test/serverenv.py
index c6d4e30..5d6c6c4 100644
--- a/test/serverenv.py
+++ b/test/serverenv.py
@@ -73,12 +73,10 @@ ticket="ticketB"
where return_code/stdout/stderr are None iff pid is still running.
'''
if expected_daemon and expected_exitcode is not None and expected_exitcode != 0:
- raise RuntimeError, \
- "Shouldn't ever expect daemon to start and then failure"
+ raise RuntimeError("Shouldn't ever expect daemon to start and then failure")
if not expected_daemon and expected_exitcode == 0:
- raise RuntimeError, \
- "Shouldn't ever expect success without starting daemon"
+ raise RuntimeError("Shouldn't ever expect success without starting daemon")
self.init_log()
@@ -122,9 +120,9 @@ ticket="ticketB"
return config_file
def kill_pid(self, pid):
- print "killing %d ..." % pid
+ print("killing %d ..." % pid)
os.kill(pid, 15)
- print "killed"
+ print("killed")
def check_daemon_handling(self, runner, expected_daemon):
'''
@@ -154,7 +152,7 @@ ticket="ticketB"
Returns the pid contained in lock_file, or None if it doesn't exist.
'''
if not os.path.exists(lock_file):
- print "%s does not exist" % lock_file
+ print("%s does not exist" % lock_file)
return None
l = open(lock_file)
@@ -162,7 +160,7 @@ ticket="ticketB"
l.close()
self.assertEqual(len(lines), 1, "Lock file should contain one line")
pid = re.search('\\bbooth_pid="?(\\d+)"?', lines[0]).group(1)
- print "lockfile contains: <%s>" % pid
+ print("lockfile contains: <%s>" % pid)
return pid
def is_pid_running_daemon(self, pid):
@@ -185,11 +183,11 @@ ticket="ticketB"
c = open("/proc/%s/cmdline" % pid)
cmdline = "".join(c.readlines())
- print cmdline
+ print(cmdline)
c.close()
if cmdline.find('boothd') == -1:
- print 'no boothd in cmdline:', cmdline
+ print('no boothd in cmdline:', cmdline)
return False
# self.assertRegexpMatches(
diff --git a/test/servertests.py b/test/servertests.py
index 71e808e..288d19f 100644
--- a/test/servertests.py
+++ b/test/servertests.py
@@ -35,13 +35,13 @@ class ServerTests(ServerTestEnvironment):
def test_config_file_buffer_overflow(self):
# https://bugzilla.novell.com/show_bug.cgi?id=750256
- longfile = (string.lowercase * 5)[:127]
+ longfile = (string.ascii_lowercase * 5)[:127]
expected_error = "'%s' exceeds maximum config name length" % longfile
self._test_buffer_overflow(expected_error, config_file=longfile)
def test_lock_file_buffer_overflow(self):
# https://bugzilla.novell.com/show_bug.cgi?id=750256
- longfile = (string.lowercase * 5)[:127]
+ longfile = (string.ascii_lowercase * 5)[:127]
expected_error = "'%s' exceeds maximum lock file length" % longfile
self._test_buffer_overflow(expected_error, lock_file=longfile)
@@ -54,12 +54,12 @@ class ServerTests(ServerTestEnvironment):
# quotes no longer required
return True
orig_lines = self.working_config.split("\n")
- for i in xrange(len(orig_lines)):
+ for (i, line) in enumerate(orig_lines):
new_lines = copy.copy(orig_lines)
- new_lines[i] = new_lines[i].replace('"', '')
+ new_lines[i] = line.replace('"', '')
new_config = "\n".join(new_lines)
- line_contains_IP = re.search('^\s*(site|arbitrator)=.*[0-9]\.', orig_lines[i])
+ line_contains_IP = re.search('^\s*(site|arbitrator)=.*[0-9]\.', line)
if line_contains_IP:
# IP addresses need to be surrounded by quotes,
# so stripping them should cause it to fail
diff --git a/test/utils.py b/test/utils.py
index 5b70cfc..aca3592 100644
--- a/test/utils.py
+++ b/test/utils.py
@@ -1,5 +1,6 @@
import subprocess
import re
+import sys
def run_cmd(cmd):
p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
@@ -9,8 +10,11 @@ def run_cmd(cmd):
def get_IP():
(stdout, stderr, returncode) = run_cmd(['hostname', '-i'])
if returncode != 0:
- raise RuntimeError, "Failed to run hostname -i:\n" + stderr
+ raise RuntimeError("Failed to run hostname -i:\n" + stderr)
# in case multiple IP addresses are returned, use only the first
- # and also strip '%<device>' part possibly present with IPv6 address
- ret = re.sub(r'\s.*', '', stdout)
+ # and also strip '%<device>' part possibly present with IPv6 address;
+ # in Python 3 context, only expect ASCII/UTF-8 encodings for the
+ # obtained input bytes
+ ret = re.sub(r'\s.*', '',
+ stdout if sys.version_info[0] < 3 else str(stdout, 'UTF-8'))
return "::1" if '%' in ret else ret
--
2.18.0.rc2