From 7f2cf2fccaff3cbdfc0f3a43a10667269f4ea4f6 Mon Sep 17 00:00:00 2001 From: Peter Wu Date: Sat, 13 Oct 2018 18:08:43 +0200 Subject: [PATCH] test: drop Python 2 compatibility, use more Python 3 features Reduces maintenance costs and makes it possible to simplify code. pytest supports Python 2.7 and Python 3.4 (or newer), so that is more or less the minimum target for now. Change-Id: I0347b6c334bf2fc6c9480ff56e9ccfcd48886dde Reviewed-on: https://code.wireshark.org/review/30193 Tested-by: Petri Dish Buildbot Petri-Dish: Peter Wu Reviewed-by: Peter Wu --- test/config.py | 5 +-- test/subprocesstest.py | 61 ++++++++++++++----------------------- test/suite_decryption.py | 2 +- test/suite_sharkd.py | 15 ++++----- test/test.py | 16 +++++----- test/util_dump_dhcp_pcap.py | 6 +--- 6 files changed, 39 insertions(+), 66 deletions(-) diff --git a/test/config.py b/test/config.py index b1a76772ff..9fd5f71940 100644 --- a/test/config.py +++ b/test/config.py @@ -127,10 +127,7 @@ def getDefaultCaptureInterface(): return try: dumpcap_d_data = subprocess.check_output((cmd_dumpcap, '-D'), stderr=subprocess.PIPE) - if sys.version_info[0] >= 3: - dumpcap_d_stdout = dumpcap_d_data.decode('UTF-8', 'replace') - else: - dumpcap_d_stdout = unicode(dumpcap_d_data, 'UTF-8', 'replace') + dumpcap_d_stdout = dumpcap_d_data.decode('UTF-8', 'replace') for d_line in dumpcap_d_stdout.splitlines(): iface_m = re.search('(\d+)\..*(Ethernet|Network Connection|VMware|Intel)', d_line) if iface_m: diff --git a/test/subprocesstest.py b/test/subprocesstest.py index 96edb259c1..c16c67db02 100644 --- a/test/subprocesstest.py +++ b/test/subprocesstest.py @@ -25,8 +25,7 @@ import unittest # XXX This should probably be in config.py and settable from # the command line. -if sys.version_info[0] >= 3: - process_timeout = 300 # Seconds +process_timeout = 300 # Seconds def capture_command(cmd, *args, **kwargs): '''Convert the supplied arguments into a command suitable for SubprocessTestCase. @@ -67,39 +66,33 @@ class LoggingPopen(subprocess.Popen): # Make sure communicate() gives us bytes. kwargs['universal_newlines'] = False self.cmd_str = 'command ' + repr(proc_args) - super(LoggingPopen, self).__init__(proc_args, *args, **kwargs) + super().__init__(proc_args, *args, **kwargs) self.stdout_str = '' self.stderr_str = '' def wait_and_log(self): '''Wait for the process to finish and log its output.''' - # Wherein we navigate the Python 2 and 3 Unicode compatibility maze. - if sys.version_info[0] >= 3: - out_data, err_data = self.communicate(timeout=process_timeout) - out_log = out_data.decode('UTF-8', 'replace') - err_log = err_data.decode('UTF-8', 'replace') - else: - out_data, err_data = self.communicate() - out_log = unicode(out_data, 'UTF-8', 'replace') - err_log = unicode(err_data, 'UTF-8', 'replace') + out_data, err_data = self.communicate(timeout=process_timeout) + out_log = out_data.decode('UTF-8', 'replace') + err_log = err_data.decode('UTF-8', 'replace') + self.log_fd.flush() + self.log_fd.write('-- Begin stdout for {} --\n'.format(self.cmd_str)) + self.log_fd.write(out_log) + self.log_fd.write('-- End stdout for {} --\n'.format(self.cmd_str)) + self.log_fd.write('-- Begin stderr for {} --\n'.format(self.cmd_str)) + self.log_fd.write(err_log) + self.log_fd.write('-- End stderr for {} --\n'.format(self.cmd_str)) + self.log_fd.flush() # Throwing a UnicodeDecodeError exception here is arguably a good thing. self.stdout_str = out_data.decode('UTF-8', 'strict') self.stderr_str = err_data.decode('UTF-8', 'strict') - self.log_fd.flush() - self.log_fd.write(u'-- Begin stdout for {} --\n'.format(self.cmd_str)) - self.log_fd.write(out_log) - self.log_fd.write(u'-- End stdout for {} --\n'.format(self.cmd_str)) - self.log_fd.write(u'-- Begin stderr for {} --\n'.format(self.cmd_str)) - self.log_fd.write(err_log) - self.log_fd.write(u'-- End stderr for {} --\n'.format(self.cmd_str)) - self.log_fd.flush() def stop_process(self, kill=False): '''Stop the process immediately.''' if kill: - super(LoggingPopen, self).kill() + super().kill() else: - super(LoggingPopen, self).terminate() + super().terminate() def terminate(self): '''Terminate the process. Do not log its output.''' @@ -114,7 +107,7 @@ class SubprocessTestCase(unittest.TestCase): '''Run a program and gather its stdout and stderr.''' def __init__(self, *args, **kwargs): - super(SubprocessTestCase, self).__init__(*args, **kwargs) + super().__init__(*args, **kwargs) self.exit_ok = 0 self.exit_command_line = 1 self.exit_error = 2 @@ -126,10 +119,7 @@ class SubprocessTestCase(unittest.TestCase): self.dump_files = [] def log_fd_write_bytes(self, log_data): - if sys.version_info[0] >= 3: - self.log_fd.write(log_data) - else: - self.log_fd.write(unicode(log_data, 'UTF-8', 'replace')) + self.log_fd.write(log_data) def filename_from_id(self, filename): '''Generate a filename prefixed with our test ID.''' @@ -175,7 +165,7 @@ class SubprocessTestCase(unittest.TestCase): self.cleanup_files.append(self.log_fname) pre_run_problem_count = self._error_count(result) try: - super(SubprocessTestCase, self).run(result=result) + super().run(result=result) except KeyboardInterrupt: # XXX This doesn't seem to work on Windows, which is where we need it the most. self.kill_processes() @@ -208,16 +198,13 @@ class SubprocessTestCase(unittest.TestCase): Default cap_file is .testout.pcap.''' if not cap_file: cap_file = self.filename_from_id('testout.pcap') - self.log_fd.write(u'\nOutput of {0} {1}:\n'.format(config.cmd_capinfos, cap_file)) + self.log_fd.write('\nOutput of {0} {1}:\n'.format(config.cmd_capinfos, cap_file)) capinfos_cmd = [config.cmd_capinfos] if capinfos_args is not None: capinfos_cmd += capinfos_args capinfos_cmd.append(cap_file) capinfos_data = subprocess.check_output(capinfos_cmd) - if sys.version_info[0] >= 3: - capinfos_stdout = capinfos_data.decode('UTF-8', 'replace') - else: - capinfos_stdout = unicode(capinfos_data, 'UTF-8', 'replace') + capinfos_stdout = capinfos_data.decode('UTF-8', 'replace') self.log_fd.write(capinfos_stdout) return capinfos_stdout @@ -238,7 +225,7 @@ class SubprocessTestCase(unittest.TestCase): if proc is None: proc = self.processes[-1] - out_data = u'' + out_data = '' if count_stdout: out_data = proc.stdout_str if count_stderr: @@ -265,12 +252,10 @@ class SubprocessTestCase(unittest.TestCase): lines_b = blob_b.splitlines() diff = '\n'.join(list(difflib.unified_diff(lines_a, lines_b, *args, **kwargs))) if len(diff) > 0: - if sys.version_info[0] < 3 and not isinstance(diff, unicode): - diff = unicode(diff, 'UTF-8', 'replace') self.log_fd.flush() - self.log_fd.write(u'-- Begin diff output --\n') + self.log_fd.write('-- Begin diff output --\n') self.log_fd.writelines(diff) - self.log_fd.write(u'-- End diff output --\n') + self.log_fd.write('-- End diff output --\n') return False return True diff --git a/test/suite_decryption.py b/test/suite_decryption.py index bc343a78af..606c89b894 100644 --- a/test/suite_decryption.py +++ b/test/suite_decryption.py @@ -512,7 +512,7 @@ class case_decrypt_http2(subprocesstest.SubprocessTestCase): env=config.test_env) test_passed = self.grepOutput('nghttp2') if not test_passed: - self.log_fd.write(u'\n\n-- Verbose output --\n\n') + self.log_fd.write('\n\n-- Verbose output --\n\n') self.runProcess((config.cmd_tshark, '-r', capture_file, '-V', diff --git a/test/suite_sharkd.py b/test/suite_sharkd.py index 27eb9d58ce..64cba38aec 100644 --- a/test/suite_sharkd.py +++ b/test/suite_sharkd.py @@ -24,9 +24,7 @@ class case_sharkd(subprocesstest.SubprocessTestCase): stdin=subprocess.PIPE ) - sharkd_commands = '{"req":"status"}\n' - if sys.version_info[0] >= 3: - sharkd_commands = sharkd_commands.encode('UTF-8') + sharkd_commands = b'{"req":"status"}\n' sharkd_proc.stdin.write(sharkd_commands) self.waitProcess(sharkd_proc) @@ -44,12 +42,11 @@ class case_sharkd(subprocesstest.SubprocessTestCase): stdin=subprocess.PIPE ) - sharkd_commands = '' - sharkd_commands = '{"req":"load","file":' + json.JSONEncoder().encode(dhcp_pcap) + '}\n' - sharkd_commands += '{"req":"status"}\n' - sharkd_commands += '{"req":"frames"}\n' - if sys.version_info[0] >= 3: - sharkd_commands = sharkd_commands.encode('UTF-8') + sharkd_commands = b'{"req":"load","file":' + sharkd_commands += json.dumps(dhcp_pcap).encode('utf8') + sharkd_commands += b'}\n' + sharkd_commands += b'{"req":"status"}\n' + sharkd_commands += b'{"req":"frames"}\n' sharkd_proc.stdin.write(sharkd_commands) self.waitProcess(sharkd_proc) diff --git a/test/test.py b/test/test.py index a1c0b059fc..1b13e12103 100755 --- a/test/test.py +++ b/test/test.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # -*- coding: utf-8 -*- # # Wireshark tests @@ -13,8 +13,6 @@ # To do: # - Avoid printing Python tracebacks when we assert? It looks like we'd need # to override unittest.TextTestResult.addFailure(). -# - Switch to Python 3 only? [Windows, Linux, macOS] x [Python 2, Python 3] -# is painful. # - Remove BIN_PATH/hosts via config.tearDownHostFiles + case_name_resolution.tearDownClass? @@ -41,6 +39,10 @@ def dump_failed_output(suite): dump_failed_output(s) def main(): + if sys.version_info[0] < 3: + print("Unit tests require Python 3") + sys.exit(2) + parser = argparse.ArgumentParser(description='Wireshark unit tests') cap_group = parser.add_mutually_exclusive_group() cap_group.add_argument('-e', '--enable-capture', action='store_true', help='Enable capture tests') @@ -132,12 +134,8 @@ def main(): import codecs import locale sys.stderr.write('Warning: Output encoding is {0} and not UTF-8.\n'.format(sys.stdout.encoding)) - if sys.version_info[0] >= 3: - sys.stdout = codecs.getwriter(locale.getpreferredencoding())(sys.stdout.buffer, 'backslashreplace') - sys.stderr = codecs.getwriter(locale.getpreferredencoding())(sys.stderr.buffer, 'backslashreplace') - else: - sys.stdout = codecs.getwriter(locale.getpreferredencoding())(sys.stdout, 'backslashreplace') - sys.stderr = codecs.getwriter(locale.getpreferredencoding())(sys.stderr, 'backslashreplace') + sys.stdout = codecs.getwriter(locale.getpreferredencoding())(sys.stdout.buffer, 'backslashreplace') + sys.stderr = codecs.getwriter(locale.getpreferredencoding())(sys.stderr.buffer, 'backslashreplace') run_suite = unittest.defaultTestLoader.loadTestsFromNames(run_ids) runner = unittest.TextTestRunner(verbosity=args.verbose) diff --git a/test/util_dump_dhcp_pcap.py b/test/util_dump_dhcp_pcap.py index 2fbd7fa138..2e0b242818 100755 --- a/test/util_dump_dhcp_pcap.py +++ b/test/util_dump_dhcp_pcap.py @@ -1,4 +1,4 @@ -#!/usr/bin/env python +#!/usr/bin/env python3 # # Wireshark tests # By Gerald Combs @@ -21,10 +21,6 @@ def main(): help='cat: Just dump the file. slow: Dump the file, pause, and dump its packet records. raw: Dump only the packet records.') args = parser.parse_args() - if sys.version_info[0] < 3 and sys.platform == "win32": - import msvcrt - msvcrt.setmode(sys.stdout.fileno(), os.O_BINARY) - dhcp_pcap = os.path.join(os.path.dirname(__file__), 'captures', 'dhcp.pcap') dhcp_fd = open(dhcp_pcap, 'rb')