This doesn't fix everything in here, but it does help clean up the
pylint report considerably.
This should be 100% style changes only; the intent is to make pylint
more useful by working on establishing a baseline for iotests that we
can gate against in the future. This will be important if (when?) we
begin adding type hints to our code base.
Signed-off-by: John Snow <jsnow@redhat.com>
---
tests/qemu-iotests/iotests.py | 83 ++++++++++++++++++-----------------
1 file changed, 43 insertions(+), 40 deletions(-)
diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
index 8815052eb5..c3aa857140 100644
--- a/tests/qemu-iotests/iotests.py
+++ b/tests/qemu-iotests/iotests.py
@@ -16,11 +16,9 @@
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#
-import errno
import os
import re
import subprocess
-import string
import unittest
import sys
import struct
@@ -34,7 +32,7 @@
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
from qemu import qtest
-assert sys.version_info >= (3,6)
+assert sys.version_info >= (3, 6)
# This will not work if arguments contain spaces but is necessary if we
# want to support the override options that ./check supports.
@@ -138,11 +136,11 @@ def qemu_img_log(*args):
return result
def img_info_log(filename, filter_path=None, imgopts=False, extra_args=[]):
- args = [ 'info' ]
+ args = ['info']
if imgopts:
args.append('--image-opts')
else:
- args += [ '-f', imgfmt ]
+ args += ['-f', imgfmt]
args += extra_args
args.append(filename)
@@ -221,7 +219,7 @@ def cmd(self, cmd):
# quit command is in close(), '\n' is added automatically
assert '\n' not in cmd
cmd = cmd.strip()
- assert cmd != 'q' and cmd != 'quit'
+ assert cmd not in ('q', 'quit')
self._p.stdin.write(cmd + '\n')
self._p.stdin.flush()
return self._read_output()
@@ -243,10 +241,8 @@ def qemu_nbd_early_pipe(*args):
sys.stderr.write('qemu-nbd received signal %i: %s\n' %
(-exitcode,
' '.join(qemu_nbd_args + ['--fork'] + list(args))))
- if exitcode == 0:
- return exitcode, ''
- else:
- return exitcode, subp.communicate()[0]
+
+ return exitcode, subp.communicate()[0] if exitcode else ''
def qemu_nbd_popen(*args):
'''Run qemu-nbd in daemon mode and return the parent's exit code'''
@@ -310,7 +306,7 @@ def filter_qmp(qmsg, filter_fn):
items = qmsg.items()
for k, v in items:
- if isinstance(v, list) or isinstance(v, dict):
+ if isinstance(v, (dict, list)):
qmsg[k] = filter_qmp(v, filter_fn)
else:
qmsg[k] = filter_fn(k, v)
@@ -321,7 +317,7 @@ def filter_testfiles(msg):
return msg.replace(prefix, 'TEST_DIR/PID-')
def filter_qmp_testfiles(qmsg):
- def _filter(key, value):
+ def _filter(_key, value):
if is_str(value):
return filter_testfiles(value)
return value
@@ -347,7 +343,7 @@ def filter_imgfmt(msg):
return msg.replace(imgfmt, 'IMGFMT')
def filter_qmp_imgfmt(qmsg):
- def _filter(key, value):
+ def _filter(_key, value):
if is_str(value):
return filter_imgfmt(value)
return value
@@ -358,7 +354,7 @@ def log(msg, filters=[], indent=None):
If indent is provided, JSON serializable messages are pretty-printed.'''
for flt in filters:
msg = flt(msg)
- if isinstance(msg, dict) or isinstance(msg, list):
+ if isinstance(msg, (dict, list)):
# Python < 3.4 needs to know not to add whitespace when pretty-printing:
separators = (', ', ': ') if indent is None else (',', ': ')
# Don't sort if it's already sorted
@@ -369,14 +365,14 @@ def log(msg, filters=[], indent=None):
print(msg)
class Timeout:
- def __init__(self, seconds, errmsg = "Timeout"):
+ def __init__(self, seconds, errmsg="Timeout"):
self.seconds = seconds
self.errmsg = errmsg
def __enter__(self):
signal.signal(signal.SIGALRM, self.timeout)
signal.setitimer(signal.ITIMER_REAL, self.seconds)
return self
- def __exit__(self, type, value, traceback):
+ def __exit__(self, exc_type, value, traceback):
signal.setitimer(signal.ITIMER_REAL, 0)
return False
def timeout(self, signum, frame):
@@ -385,7 +381,7 @@ def timeout(self, signum, frame):
def file_pattern(name):
return "{0}-{1}".format(os.getpid(), name)
-class FilePaths(object):
+class FilePaths:
"""
FilePaths is an auto-generated filename that cleans itself up.
@@ -532,11 +528,11 @@ def pause_drive(self, drive, event=None):
self.pause_drive(drive, "write_aio")
return
self.qmp('human-monitor-command',
- command_line='qemu-io %s "break %s bp_%s"' % (drive, event, drive))
+ command_line='qemu-io %s "break %s bp_%s"' % (drive, event, drive))
def resume_drive(self, drive):
self.qmp('human-monitor-command',
- command_line='qemu-io %s "remove_break bp_%s"' % (drive, drive))
+ command_line='qemu-io %s "remove_break bp_%s"' % (drive, drive))
def hmp_qemu_io(self, drive, cmd):
'''Write to a given drive using an HMP command'''
@@ -547,8 +543,8 @@ def flatten_qmp_object(self, obj, output=None, basestr=''):
if output is None:
output = dict()
if isinstance(obj, list):
- for i in range(len(obj)):
- self.flatten_qmp_object(obj[i], output, basestr + str(i) + '.')
+ for i, atom in enumerate(obj):
+ self.flatten_qmp_object(atom, output, basestr + str(i) + '.')
elif isinstance(obj, dict):
for key in obj:
self.flatten_qmp_object(obj[key], output, basestr + key + '.')
@@ -703,9 +699,7 @@ def get_bitmap(self, node_name, bitmap_name, recording=None, bitmaps=None):
for bitmap in bitmaps[node_name]:
if bitmap.get('name', '') == bitmap_name:
- if recording is None:
- return bitmap
- elif bitmap.get('recording') == recording:
+ if recording is None or bitmap.get('recording') == recording:
return bitmap
return None
@@ -756,12 +750,13 @@ def assert_block_path(self, root, path, expected_node, graph=None):
assert node is not None, 'Cannot follow path %s%s' % (root, path)
try:
- node_id = next(edge['child'] for edge in graph['edges'] \
- if edge['parent'] == node['id'] and
- edge['name'] == child_name)
+ node_id = next(edge['child'] for edge in graph['edges']
+ if (edge['parent'] == node['id'] and
+ edge['name'] == child_name))
+
+ node = next(node for node in graph['nodes']
+ if node['id'] == node_id)
- node = next(node for node in graph['nodes'] \
- if node['id'] == node_id)
except StopIteration:
node = None
@@ -779,6 +774,12 @@ def assert_block_path(self, root, path, expected_node, graph=None):
class QMPTestCase(unittest.TestCase):
'''Abstract base class for QMP test cases'''
+ def __init__(self, *args, **kwargs):
+ super().__init__(*args, **kwargs)
+ # Many users of this class set a VM property we rely on heavily
+ # in the methods below.
+ self.vm = None
+
def dictpath(self, d, path):
'''Traverse a path in a nested dict'''
for component in path.split('/'):
@@ -824,7 +825,7 @@ def assert_qmp(self, d, path, value):
else:
self.assertEqual(result, value,
'"%s" is "%s", expected "%s"'
- % (path, str(result), str(value)))
+ % (path, str(result), str(value)))
def assert_no_active_block_jobs(self):
result = self.vm.qmp('query-block-jobs')
@@ -834,15 +835,15 @@ def assert_has_block_node(self, node_name=None, file_name=None):
"""Issue a query-named-block-nodes and assert node_name and/or
file_name is present in the result"""
def check_equal_or_none(a, b):
- return a == None or b == None or a == b
+ return a is None or b is None or a == b
assert node_name or file_name
result = self.vm.qmp('query-named-block-nodes')
for x in result["return"]:
if check_equal_or_none(x.get("node-name"), node_name) and \
check_equal_or_none(x.get("file"), file_name):
return
- self.assertTrue(False, "Cannot find %s %s in result:\n%s" % \
- (node_name, file_name, result))
+ self.fail("Cannot find %s %s in result:\n%s" %
+ (node_name, file_name, result))
def assert_json_filename_equal(self, json_filename, reference):
'''Asserts that the given filename is a json: filename and that its
@@ -891,13 +892,13 @@ def wait_until_completed(self, drive='drive0', check_offset=True, wait=60.0,
self.assert_qmp(event, 'data/error', error)
self.assert_no_active_block_jobs()
return event
- elif event['event'] == 'JOB_STATUS_CHANGE':
+ if event['event'] == 'JOB_STATUS_CHANGE':
self.assert_qmp(event, 'data/id', drive)
def wait_ready(self, drive='drive0'):
- '''Wait until a block job BLOCK_JOB_READY event'''
- f = {'data': {'type': 'mirror', 'device': drive } }
- event = self.vm.event_wait(name='BLOCK_JOB_READY', match=f)
+ """Wait until a BLOCK_JOB_READY event, and return the event."""
+ f = {'data': {'type': 'mirror', 'device': drive}}
+ return self.vm.event_wait(name='BLOCK_JOB_READY', match=f)
def wait_ready_and_cancel(self, drive='drive0'):
self.wait_ready(drive=drive)
@@ -926,7 +927,7 @@ def pause_wait(self, job_id='job0'):
for job in result['return']:
if job['device'] == job_id:
found = True
- if job['paused'] == True and job['busy'] == False:
+ if job['paused'] and not job['busy']:
return job
break
assert found
@@ -1023,8 +1024,8 @@ def qemu_pipe(*args):
universal_newlines=True)
exitcode = subp.wait()
if exitcode < 0:
- sys.stderr.write('qemu received signal %i: %s\n' % (-exitcode,
- ' '.join(args)))
+ sys.stderr.write('qemu received signal %i: %s\n' %
+ (-exitcode, ' '.join(args)))
return subp.communicate()[0]
def supported_formats(read_only=False):
@@ -1056,6 +1057,7 @@ def func_wrapper(test_case: QMPTestCase, *args, **kwargs):
if usf_list:
test_case.case_skip('{}: formats {} are not whitelisted'.format(
test_case, usf_list))
+ return None
else:
return func(test_case, *args, **kwargs)
return func_wrapper
@@ -1067,6 +1069,7 @@ def skip_if_user_is_root(func):
def func_wrapper(*args, **kwargs):
if os.getuid() == 0:
case_notrun('{}: cannot be run as root'.format(args[0]))
+ return None
else:
return func(*args, **kwargs)
return func_wrapper
--
2.21.1
On 3/4/20 10:38 PM, John Snow wrote:
> This doesn't fix everything in here, but it does help clean up the
> pylint report considerably.
>
> This should be 100% style changes only; the intent is to make pylint
> more useful by working on establishing a baseline for iotests that we
> can gate against in the future. This will be important if (when?) we
> begin adding type hints to our code base.
>
> Signed-off-by: John Snow <jsnow@redhat.com>
> ---
> tests/qemu-iotests/iotests.py | 83 ++++++++++++++++++-----------------
> 1 file changed, 43 insertions(+), 40 deletions(-)
>
> diff --git a/tests/qemu-iotests/iotests.py b/tests/qemu-iotests/iotests.py
> index 8815052eb5..c3aa857140 100644
> --- a/tests/qemu-iotests/iotests.py
> +++ b/tests/qemu-iotests/iotests.py
> @@ -16,11 +16,9 @@
> # along with this program. If not, see <http://www.gnu.org/licenses/>.
> #
>
> -import errno
> import os
> import re
> import subprocess
> -import string
> import unittest
> import sys
> import struct
> @@ -34,7 +32,7 @@
> sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
> from qemu import qtest
>
> -assert sys.version_info >= (3,6)
> +assert sys.version_info >= (3, 6)
>
> # This will not work if arguments contain spaces but is necessary if we
> # want to support the override options that ./check supports.
> @@ -138,11 +136,11 @@ def qemu_img_log(*args):
> return result
>
> def img_info_log(filename, filter_path=None, imgopts=False, extra_args=[]):
> - args = [ 'info' ]
> + args = ['info']
> if imgopts:
> args.append('--image-opts')
> else:
> - args += [ '-f', imgfmt ]
> + args += ['-f', imgfmt]
> args += extra_args
> args.append(filename)
>
> @@ -221,7 +219,7 @@ def cmd(self, cmd):
> # quit command is in close(), '\n' is added automatically
> assert '\n' not in cmd
> cmd = cmd.strip()
> - assert cmd != 'q' and cmd != 'quit'
> + assert cmd not in ('q', 'quit')
> self._p.stdin.write(cmd + '\n')
> self._p.stdin.flush()
> return self._read_output()
> @@ -243,10 +241,8 @@ def qemu_nbd_early_pipe(*args):
> sys.stderr.write('qemu-nbd received signal %i: %s\n' %
> (-exitcode,
> ' '.join(qemu_nbd_args + ['--fork'] + list(args))))
> - if exitcode == 0:
> - return exitcode, ''
> - else:
> - return exitcode, subp.communicate()[0]
> +
> + return exitcode, subp.communicate()[0] if exitcode else ''
>
> def qemu_nbd_popen(*args):
> '''Run qemu-nbd in daemon mode and return the parent's exit code'''
> @@ -310,7 +306,7 @@ def filter_qmp(qmsg, filter_fn):
> items = qmsg.items()
>
> for k, v in items:
> - if isinstance(v, list) or isinstance(v, dict):
> + if isinstance(v, (dict, list)):
> qmsg[k] = filter_qmp(v, filter_fn)
> else:
> qmsg[k] = filter_fn(k, v)
> @@ -321,7 +317,7 @@ def filter_testfiles(msg):
> return msg.replace(prefix, 'TEST_DIR/PID-')
>
> def filter_qmp_testfiles(qmsg):
> - def _filter(key, value):
> + def _filter(_key, value):
> if is_str(value):
> return filter_testfiles(value)
> return value
> @@ -347,7 +343,7 @@ def filter_imgfmt(msg):
> return msg.replace(imgfmt, 'IMGFMT')
>
> def filter_qmp_imgfmt(qmsg):
> - def _filter(key, value):
> + def _filter(_key, value):
> if is_str(value):
> return filter_imgfmt(value)
> return value
> @@ -358,7 +354,7 @@ def log(msg, filters=[], indent=None):
> If indent is provided, JSON serializable messages are pretty-printed.'''
> for flt in filters:
> msg = flt(msg)
> - if isinstance(msg, dict) or isinstance(msg, list):
> + if isinstance(msg, (dict, list)):
> # Python < 3.4 needs to know not to add whitespace when pretty-printing:
> separators = (', ', ': ') if indent is None else (',', ': ')
> # Don't sort if it's already sorted
> @@ -369,14 +365,14 @@ def log(msg, filters=[], indent=None):
> print(msg)
>
> class Timeout:
> - def __init__(self, seconds, errmsg = "Timeout"):
> + def __init__(self, seconds, errmsg="Timeout"):
> self.seconds = seconds
> self.errmsg = errmsg
> def __enter__(self):
> signal.signal(signal.SIGALRM, self.timeout)
> signal.setitimer(signal.ITIMER_REAL, self.seconds)
> return self
> - def __exit__(self, type, value, traceback):
> + def __exit__(self, exc_type, value, traceback):
> signal.setitimer(signal.ITIMER_REAL, 0)
> return False
> def timeout(self, signum, frame):
> @@ -385,7 +381,7 @@ def timeout(self, signum, frame):
> def file_pattern(name):
> return "{0}-{1}".format(os.getpid(), name)
>
> -class FilePaths(object):
> +class FilePaths:
> """
> FilePaths is an auto-generated filename that cleans itself up.
>
> @@ -532,11 +528,11 @@ def pause_drive(self, drive, event=None):
> self.pause_drive(drive, "write_aio")
> return
> self.qmp('human-monitor-command',
> - command_line='qemu-io %s "break %s bp_%s"' % (drive, event, drive))
> + command_line='qemu-io %s "break %s bp_%s"' % (drive, event, drive))
>
> def resume_drive(self, drive):
> self.qmp('human-monitor-command',
> - command_line='qemu-io %s "remove_break bp_%s"' % (drive, drive))
> + command_line='qemu-io %s "remove_break bp_%s"' % (drive, drive))
>
> def hmp_qemu_io(self, drive, cmd):
> '''Write to a given drive using an HMP command'''
> @@ -547,8 +543,8 @@ def flatten_qmp_object(self, obj, output=None, basestr=''):
> if output is None:
> output = dict()
> if isinstance(obj, list):
> - for i in range(len(obj)):
> - self.flatten_qmp_object(obj[i], output, basestr + str(i) + '.')
> + for i, atom in enumerate(obj):
> + self.flatten_qmp_object(atom, output, basestr + str(i) + '.')
> elif isinstance(obj, dict):
> for key in obj:
> self.flatten_qmp_object(obj[key], output, basestr + key + '.')
> @@ -703,9 +699,7 @@ def get_bitmap(self, node_name, bitmap_name, recording=None, bitmaps=None):
>
> for bitmap in bitmaps[node_name]:
> if bitmap.get('name', '') == bitmap_name:
> - if recording is None:
> - return bitmap
> - elif bitmap.get('recording') == recording:
> + if recording is None or bitmap.get('recording') == recording:
> return bitmap
> return None
>
> @@ -756,12 +750,13 @@ def assert_block_path(self, root, path, expected_node, graph=None):
> assert node is not None, 'Cannot follow path %s%s' % (root, path)
>
> try:
> - node_id = next(edge['child'] for edge in graph['edges'] \
> - if edge['parent'] == node['id'] and
> - edge['name'] == child_name)
> + node_id = next(edge['child'] for edge in graph['edges']
> + if (edge['parent'] == node['id'] and
> + edge['name'] == child_name))
> +
> + node = next(node for node in graph['nodes']
> + if node['id'] == node_id)
>
> - node = next(node for node in graph['nodes'] \
> - if node['id'] == node_id)
> except StopIteration:
> node = None
>
> @@ -779,6 +774,12 @@ def assert_block_path(self, root, path, expected_node, graph=None):
> class QMPTestCase(unittest.TestCase):
> '''Abstract base class for QMP test cases'''
>
> + def __init__(self, *args, **kwargs):
> + super().__init__(*args, **kwargs)
> + # Many users of this class set a VM property we rely on heavily
> + # in the methods below.
> + self.vm = None
> +
> def dictpath(self, d, path):
> '''Traverse a path in a nested dict'''
> for component in path.split('/'):
> @@ -824,7 +825,7 @@ def assert_qmp(self, d, path, value):
> else:
> self.assertEqual(result, value,
> '"%s" is "%s", expected "%s"'
> - % (path, str(result), str(value)))
> + % (path, str(result), str(value)))
>
> def assert_no_active_block_jobs(self):
> result = self.vm.qmp('query-block-jobs')
> @@ -834,15 +835,15 @@ def assert_has_block_node(self, node_name=None, file_name=None):
> """Issue a query-named-block-nodes and assert node_name and/or
> file_name is present in the result"""
> def check_equal_or_none(a, b):
> - return a == None or b == None or a == b
> + return a is None or b is None or a == b
> assert node_name or file_name
> result = self.vm.qmp('query-named-block-nodes')
> for x in result["return"]:
> if check_equal_or_none(x.get("node-name"), node_name) and \
> check_equal_or_none(x.get("file"), file_name):
> return
> - self.assertTrue(False, "Cannot find %s %s in result:\n%s" % \
> - (node_name, file_name, result))
> + self.fail("Cannot find %s %s in result:\n%s" %
> + (node_name, file_name, result))
>
> def assert_json_filename_equal(self, json_filename, reference):
> '''Asserts that the given filename is a json: filename and that its
> @@ -891,13 +892,13 @@ def wait_until_completed(self, drive='drive0', check_offset=True, wait=60.0,
> self.assert_qmp(event, 'data/error', error)
> self.assert_no_active_block_jobs()
> return event
> - elif event['event'] == 'JOB_STATUS_CHANGE':
> + if event['event'] == 'JOB_STATUS_CHANGE':
I don't understand the rational for this change, but the result is
correct anyway, so thanks for this nice cleanup!
Reviewed-by: Philippe Mathieu-Daudé <philmd@redhat.com>
> self.assert_qmp(event, 'data/id', drive)
>
> def wait_ready(self, drive='drive0'):
> - '''Wait until a block job BLOCK_JOB_READY event'''
> - f = {'data': {'type': 'mirror', 'device': drive } }
> - event = self.vm.event_wait(name='BLOCK_JOB_READY', match=f)
> + """Wait until a BLOCK_JOB_READY event, and return the event."""
> + f = {'data': {'type': 'mirror', 'device': drive}}
> + return self.vm.event_wait(name='BLOCK_JOB_READY', match=f)
>
> def wait_ready_and_cancel(self, drive='drive0'):
> self.wait_ready(drive=drive)
> @@ -926,7 +927,7 @@ def pause_wait(self, job_id='job0'):
> for job in result['return']:
> if job['device'] == job_id:
> found = True
> - if job['paused'] == True and job['busy'] == False:
> + if job['paused'] and not job['busy']:
> return job
> break
> assert found
> @@ -1023,8 +1024,8 @@ def qemu_pipe(*args):
> universal_newlines=True)
> exitcode = subp.wait()
> if exitcode < 0:
> - sys.stderr.write('qemu received signal %i: %s\n' % (-exitcode,
> - ' '.join(args)))
> + sys.stderr.write('qemu received signal %i: %s\n' %
> + (-exitcode, ' '.join(args)))
> return subp.communicate()[0]
>
> def supported_formats(read_only=False):
> @@ -1056,6 +1057,7 @@ def func_wrapper(test_case: QMPTestCase, *args, **kwargs):
> if usf_list:
> test_case.case_skip('{}: formats {} are not whitelisted'.format(
> test_case, usf_list))
> + return None
> else:
> return func(test_case, *args, **kwargs)
> return func_wrapper
> @@ -1067,6 +1069,7 @@ def skip_if_user_is_root(func):
> def func_wrapper(*args, **kwargs):
> if os.getuid() == 0:
> case_notrun('{}: cannot be run as root'.format(args[0]))
> + return None
> else:
> return func(*args, **kwargs)
> return func_wrapper
>
On 3/4/20 4:45 PM, Philippe Mathieu-Daudé wrote: > I don't understand the rational for this change, but the result is > correct anyway, so thanks for this nice cleanup! Leftover from when 'no-else-return' was enabled. It's mostly harmless.
© 2016 - 2025 Red Hat, Inc.