1
0
mirror of https://gitlab.com/libvirt/libvirt.git synced 2024-12-22 17:34:18 +03:00
libvirt/scripts/qemu-replies-tool.py

241 lines
7.1 KiB
Python
Raw Normal View History

#!/usr/bin/env python3
#
# SPDX-License-Identifier: LGPL-2.1-or-later
#
# A "swiss army knife" tool for qemu capability probing '.replies' files. See
# below in 'description' for more information.
from pathlib import Path
import argparse
import json
import os
import sys
class qrtException(Exception):
pass
# Load the 'replies' file into a list of (command, reply) tuples of parsed JSON
def qemu_replies_load(filename):
conv = []
with open(filename, "r") as fh:
command = None
jsonstr = ''
try:
for line in fh:
jsonstr += line
if line == '\n':
if command is None:
command = json.loads(jsonstr)
else:
conv.append((command, json.loads(jsonstr)))
command = None
jsonstr = ''
if command is not None and jsonstr != '':
conv.append((command, json.loads(jsonstr)))
command = None
jsonstr = ''
except json.decoder.JSONDecodeError as je:
raise qrtException("JSON error:\n'%s'\nwhile processing snippet:\n'%s'" % (je, jsonstr))
if command is not None or jsonstr != '':
if command is not None:
errorstr = json.dumps(command, indent=2)
else:
errorstr = jsonstr
raise qrtException("replies file error: Missing reply for command:\n'%s'" % errorstr)
return conv
# Format the list of (command, reply) tuples into a string and compare it with
# the 'replies' file. Optionally regenerate the replies file if the output doesn't match
def qemu_replies_compare_or_replace(filename, conv, regenerate_on_error):
actual = ''
seq = 9999 # poison the initial counter state
# possibly fix mis-ordererd 'id' fields
for (cmd, rep) in conv:
# 'qmp_capabilities' command restarts the numbering sequence
if cmd['execute'] == 'qmp_capabilities':
seq = 1
newid = 'libvirt-%d' % seq
cmd['id'] = newid
rep['id'] = newid
seq += 1
# format the output string
if len(actual) != 0:
actual += '\n\n'
actual += json.dumps(cmd, indent=2) + '\n\n' + json.dumps(rep, indent=2)
expect = ''
actual += '\n'
with open(filename, "r") as fh:
expect = fh.read()
if actual != expect:
if regenerate_on_error:
with open(filename, "w") as fh:
fh.write(actual)
raise qrtException("replies file error: Expected content of '%s' doesn't match actual content" % filename)
# Process the replies file programmatically here.
# The 'conv' argument contains the whole conversation as a list of
# (command, reply) tuples, where both command and reply are already parsed JSON
# and thus represented by native python types (dict, list, etc ...)
#
# The code below contains a few examples and hints how to use the programatic
# processing. Do not forget to use '--regenerate' flag to update the output files.
#
# Beware that this updates the output file which is used as input for any
# subsequent re-run of the tool which can re-apply the modification.
def modify_replies(conv):
return # remove this to enable modifications
version = None # filled with a dictionary with 'major', 'minor', 'micro' keys
# find version of current qemu for later use
for (cmd, rep) in conv:
if cmd['execute'] == 'query-version':
version = rep['return']['qemu']
break
if version is None:
raise Exception("'query-version' not found in the .replies file")
idx = -1
# Find index of a command, in this case we're looking for the last
# invocation of given command
for i in range(len(conv)):
(cmd, rep) = conv[i]
if cmd['execute'] == 'device-list-properties':
idx = i
if idx == -1:
raise Exception("entry not found")
# Prepare data for inserting a new command
# Command definition and error are instantiated via native python types
cmd = {'execute': 'device-list-properties',
'arguments': {'typename': 'example-device'}}
reply_unsupp = {'error': {'class': 'DeviceNotFound',
'desc': "Device 'example-device' not found"}}
# Real reply data can be also parsed from JSON
reply = json.loads('''
{
"return": [
{
"name": "dummy_prop",
"type": "str"
},
{
"name": "test",
"type": "str"
}
]
}
''')
# insert command into the QMP conversation based on version of qemu
if version['major'] >= 8 and version['minor'] > 0:
conv.insert(idx, (cmd, reply))
else:
conv.insert(idx, (cmd, reply_unsupp))
def process_one(filename, args):
try:
conv = qemu_replies_load(filename)
modify_replies(conv)
qemu_replies_compare_or_replace(filename, conv, args.regenerate)
except qrtException as e:
print("'%s' ... FAIL\n%s" % (filename, e))
return False
print("'%s' ... OK" % filename)
return True
description = '''A Swiss army knife tool for '.replies' files used by 'qemucapabilitiestest'
This tool is used to validate, programmatically update or inspect the
'.*replies' normally stored files under 'tests/qemucapabilitiesdata'.
By default the file(s) passed as positional argument are used. All '.replies'
files in a directory can be processed by specifying '--repliesdir /path/to/dir'
argument.
The default mode is validation which checks the following:
- each command has a reply and both are valid JSON
- numbering of the 'id' field is as expected
- the input file has the expected JSON formatting
The tool can be also used to programmaticaly modify the '.replies' file by
editing the 'modify_replies' method directly in the source, or for
re-formatting and re-numbering the '.replies' file to conform with the required
format. To update the output file the '--regenerate' flag can be used or the
'VIR_TEST_REGENERATE_OUTPUT' environment variable must be set to '1'.
'''
if os.environ.get('VIR_TEST_REGENERATE_OUTPUT', '0') == '1':
default_regenerate = True
else:
default_regenerate = False
parser = argparse.ArgumentParser(formatter_class=argparse.RawDescriptionHelpFormatter,
description=description)
parser.add_argument('--regenerate', action="store_true", default=default_regenerate,
help="regenerate output file if actual output doesn't match")
parser.add_argument('--repliesdir', default='',
help='use all .replies files from the directory')
parser.add_argument('replyfiles', nargs='*',
help='.replies file(s) to process')
args = parser.parse_args()
files = []
if args.replyfiles:
files += args.replyfiles
if args.repliesdir:
files += Path(args.repliesdir).glob('*.replies')
if len(files) == 0:
parser.print_help()
sys.exit(1)
fail = False
for file in files:
if not process_one(str(file), args):
fail = True
if fail:
sys.exit(1)