osmocom
/
wireshark
11
0
Fork 1
Code Issues Pull Requests Projects Releases Wiki Activity

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 <peter@lekensteyn.nl>
Reviewed-by: Peter Wu <peter@lekensteyn.nl>
This commit is contained in:
Peter Wu 2018-10-13 18:08:43 +02:00
parent 50a0a78bd0
commit 7f2cf2fcca
6 changed files with 39 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
test/config.py
View File

@ -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:

61
test/subprocesstest.py
View File

@ -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 <test id>.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

2
test/suite_decryption.py
View File

@ -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',

15
test/suite_sharkd.py
View File

@ -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)

16
test/test.py
View File

@ -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)

6
test/util_dump_dhcp_pcap.py
View File

@ -1,4 +1,4 @@
#!/usr/bin/env python
#!/usr/bin/env python3
#
# Wireshark tests
# By Gerald Combs <gerald@wireshark.org>
@ -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')