@@ -88,6 +88,7 @@ class QMPCompleter(list):
else:
state -= 1
+
class QMPShellError(Exception):
pass
@@ -105,6 +106,7 @@ class FuzzyJSON(ast.NodeTransformer):
node.id = 'None'
return node
+
# TODO: QMPShell's interface is a bit ugly (eg. _fill_completion() and
# _execute_cmd()). Let's design a better one.
class QMPShell(qmp.QEMUMonitorProtocol):
@@ -131,8 +133,8 @@ class QMPShell(qmp.QEMUMonitorProtocol):
readline.set_history_length(1024)
readline.set_completer(self._completer.complete)
readline.parse_and_bind("tab: complete")
- # XXX: default delimiters conflict with some command names (eg. query-),
- # clearing everything as it doesn't seem to matter
+ # NB: default delimiters conflict with some command names
+ # (eg. query-), clearing everything as it doesn't seem to matter
readline.set_completer_delims('')
try:
readline.read_history_file(self._histfile)
@@ -180,7 +182,9 @@ class QMPShell(qmp.QEMUMonitorProtocol):
for arg in tokens:
(key, sep, val) = arg.partition('=')
if sep != '=':
- raise QMPShellError("Expected a key=value pair, got '%s'" % arg)
+ raise QMPShellError(
+ f"Expected a key=value pair, got '{arg!s}'"
+ )
value = self.__parse_value(val)
optpath = key.split('.')
@@ -189,14 +193,16 @@ class QMPShell(qmp.QEMUMonitorProtocol):
curpath.append(p)
d = parent.get(p, {})
if type(d) is not dict:
- raise QMPShellError('Cannot use "%s" as both leaf and non-leaf key' % '.'.join(curpath))
+ msg = 'Cannot use "{:s}" as both leaf and non-leaf key'
+ raise QMPShellError(msg.format('.'.join(curpath)))
parent[p] = d
parent = d
if optpath[-1] in parent:
if type(parent[optpath[-1]]) is dict:
- raise QMPShellError('Cannot use "%s" as both leaf and non-leaf key' % '.'.join(curpath))
+ msg = 'Cannot use "{:s}" as both leaf and non-leaf key'
+ raise QMPShellError(msg.format('.'.join(curpath)))
else:
- raise QMPShellError('Cannot set "%s" multiple times' % key)
+ raise QMPShellError(f'Cannot set "{key}" multiple times')
parent[optpath[-1]] = value
def __build_cmd(self, cmdline):
@@ -206,7 +212,8 @@ class QMPShell(qmp.QEMUMonitorProtocol):
< command-name > [ arg-name1=arg1 ] ... [ arg-nameN=argN ]
"""
- cmdargs = re.findall(r'''(?:[^\s"']|"(?:\\.|[^"])*"|'(?:\\.|[^'])*')+''', cmdline)
+ argument_regex = r'''(?:[^\s"']|"(?:\\.|[^"])*"|'(?:\\.|[^'])*')+'''
+ cmdargs = re.findall(argument_regex, cmdline)
# Transactional CLI entry/exit:
if cmdargs[0] == 'transaction(':
@@ -215,9 +222,12 @@ class QMPShell(qmp.QEMUMonitorProtocol):
elif cmdargs[0] == ')' and self._transmode:
self._transmode = False
if len(cmdargs) > 1:
- raise QMPShellError("Unexpected input after close of Transaction sub-shell")
- qmpcmd = { 'execute': 'transaction',
- 'arguments': { 'actions': self._actions } }
+ msg = 'Unexpected input after close of Transaction sub-shell'
+ raise QMPShellError(msg)
+ qmpcmd = {
+ 'execute': 'transaction',
+ 'arguments': {'actions': self._actions}
+ }
self._actions = list()
return qmpcmd
@@ -228,7 +238,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
# Parse and then cache this Transactional Action
if self._transmode:
finalize = False
- action = { 'type': cmdargs[0], 'data': {} }
+ action = {'type': cmdargs[0], 'data': {}}
if cmdargs[-1] == ')':
cmdargs.pop(-1)
finalize = True
@@ -237,7 +247,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
return self.__build_cmd(')') if finalize else None
# Standard command: parse and return it to be executed.
- qmpcmd = { 'execute': cmdargs[0], 'arguments': {} }
+ qmpcmd = {'execute': cmdargs[0], 'arguments': {}}
self.__cli_expr(cmdargs[1:], qmpcmd['arguments'])
return qmpcmd
@@ -278,7 +288,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
print('Connected')
return
version = self._greeting['QMP']['version']['qemu']
- print('Connected to QEMU %d.%d.%d\n' % (version['major'],version['minor'],version['micro']))
+ print("Connected to QEMU {major}.{minor}.{micro}\n".format(**version))
def get_prompt(self):
if self._transmode:
@@ -307,6 +317,7 @@ class QMPShell(qmp.QEMUMonitorProtocol):
def set_verbosity(self, verbose):
self._verbose = verbose
+
class HMPShell(QMPShell):
def __init__(self, address):
QMPShell.__init__(self, address)
@@ -315,7 +326,7 @@ class HMPShell(QMPShell):
def __cmd_completion(self):
for cmd in self.__cmd_passthrough('help')['return'].split('\r\n'):
if cmd and cmd[0] != '[' and cmd[0] != '\t':
- name = cmd.split()[0] # drop help text
+ name = cmd.split()[0] # drop help text
if name == 'info':
continue
if name.find('|') != -1:
@@ -327,7 +338,7 @@ class HMPShell(QMPShell):
else:
name = opt[0]
self._completer.append(name)
- self._completer.append('help ' + name) # help completion
+ self._completer.append('help ' + name) # help completion
def __info_completion(self):
for cmd in self.__cmd_passthrough('info')['return'].split('\r\n'):
@@ -343,17 +354,21 @@ class HMPShell(QMPShell):
self.__info_completion()
self.__other_completion()
- def __cmd_passthrough(self, cmdline, cpu_index = 0):
- return self.cmd_obj({ 'execute': 'human-monitor-command', 'arguments':
- { 'command-line': cmdline,
- 'cpu-index': cpu_index } })
+ def __cmd_passthrough(self, cmdline, cpu_index=0):
+ return self.cmd_obj({
+ 'execute': 'human-monitor-command',
+ 'arguments': {
+ 'command-line': cmdline,
+ 'cpu-index': cpu_index
+ }
+ })
def _execute_cmd(self, cmdline):
if cmdline.split()[0] == "cpu":
# trap the cpu command, it requires special setting
try:
idx = int(cmdline.split()[1])
- if not 'return' in self.__cmd_passthrough('info version', idx):
+ if 'return' not in self.__cmd_passthrough('info version', idx):
print('bad CPU index')
return True
self.__cpu_index = idx
@@ -377,20 +392,26 @@ class HMPShell(QMPShell):
def show_banner(self):
QMPShell.show_banner(self, msg='Welcome to the HMP shell!')
+
def die(msg):
sys.stderr.write('ERROR: %s\n' % msg)
sys.exit(1)
+
def fail_cmdline(option=None):
if option:
sys.stderr.write('ERROR: bad command-line option \'%s\'\n' % option)
- sys.stderr.write('qmp-shell [ -v ] [ -p ] [ -H ] [ -N ] < UNIX socket path> | < TCP address:port >\n')
+ sys.stderr.write(
+ 'qmp-shell [ -v ] [ -p ] [ -H ] [ -N ] '
+ '< UNIX socket path> | < TCP address:port >\n'
+ )
sys.stderr.write(' -v Verbose (echo command sent and received)\n')
sys.stderr.write(' -p Pretty-print JSON\n')
sys.stderr.write(' -H Use HMP interface\n')
sys.stderr.write(' -N Skip negotiate (for qemu-ga)\n')
sys.exit(1)
+
def main():
addr = ''
qemu = None
@@ -440,5 +461,6 @@ def main():
pass
qemu.close()
+
if __name__ == '__main__':
main()