[Date Prev][Date Next][Thread Prev][Thread Next][Date Index][Thread Index]
[PATCH v2 24/72] scripts/qmp/qemu-ga-client: add mypy type hints
From: |
John Snow |
Subject: |
[PATCH v2 24/72] scripts/qmp/qemu-ga-client: add mypy type hints |
Date: |
Tue, 3 Nov 2020 19:35:14 -0500 |
This script is in slightly rough shape, but it still works. A lot of
care went into its initial development. In good faith, I'm updating it
to the latest Python coding standards. If there is in interest in this
script, though, I'll be asking for a contributor to take care of it
further.
Signed-off-by: John Snow <jsnow@redhat.com>
---
scripts/qmp/qemu-ga-client | 89 +++++++++++++++++++++-----------------
1 file changed, 49 insertions(+), 40 deletions(-)
diff --git a/scripts/qmp/qemu-ga-client b/scripts/qmp/qemu-ga-client
index 59b91c78d884..3e617e7e7abe 100755
--- a/scripts/qmp/qemu-ga-client
+++ b/scripts/qmp/qemu-ga-client
@@ -44,10 +44,18 @@ import errno
import os
import random
import sys
+from typing import (
+ Any,
+ Callable,
+ Dict,
+ Optional,
+ Sequence,
+)
sys.path.append(os.path.join(os.path.dirname(__file__), '..', '..', 'python'))
from qemu import qmp
+from qemu.qmp import SocketAddrT
# This script has not seen many patches or careful attention in quite
@@ -58,18 +66,18 @@ from qemu import qmp
class QemuGuestAgent(qmp.QEMUMonitorProtocol):
- def __getattr__(self, name):
- def wrapper(**kwds):
+ def __getattr__(self, name: str) -> Callable[..., Any]:
+ def wrapper(**kwds: object) -> object:
return self.command('guest-' + name.replace('_', '-'), **kwds)
return wrapper
class QemuGuestAgentClient:
- def __init__(self, address):
+ def __init__(self, address: SocketAddrT):
self.qga = QemuGuestAgent(address)
self.qga.connect(negotiate=False)
- def sync(self, timeout=3):
+ def sync(self, timeout: Optional[float] = 3) -> None:
# Avoid being blocked forever
if not self.ping(timeout):
raise EnvironmentError('Agent seems not alive')
@@ -79,9 +87,9 @@ class QemuGuestAgentClient:
if isinstance(ret, int) and int(ret) == uid:
break
- def __file_read_all(self, handle):
+ def __file_read_all(self, handle: int) -> bytes:
eof = False
- data = ''
+ data = b''
while not eof:
ret = self.qga.file_read(handle=handle, count=1024)
_data = base64.b64decode(ret['buf-b64'])
@@ -89,7 +97,7 @@ class QemuGuestAgentClient:
eof = ret['eof']
return data
- def read(self, path):
+ def read(self, path: str) -> bytes:
handle = self.qga.file_open(path=path)
try:
data = self.__file_read_all(handle)
@@ -97,7 +105,7 @@ class QemuGuestAgentClient:
self.qga.file_close(handle=handle)
return data
- def info(self):
+ def info(self) -> str:
info = self.qga.info()
msgs = []
@@ -113,14 +121,14 @@ class QemuGuestAgentClient:
return '\n'.join(msgs)
@classmethod
- def __gen_ipv4_netmask(cls, prefixlen):
+ def __gen_ipv4_netmask(cls, prefixlen: int) -> str:
mask = int('1' * prefixlen + '0' * (32 - prefixlen), 2)
return '.'.join([str(mask >> 24),
str((mask >> 16) & 0xff),
str((mask >> 8) & 0xff),
str(mask & 0xff)])
- def ifconfig(self):
+ def ifconfig(self) -> str:
nifs = self.qga.network_get_interfaces()
msgs = []
@@ -141,7 +149,7 @@ class QemuGuestAgentClient:
return '\n'.join(msgs)
- def ping(self, timeout):
+ def ping(self, timeout: Optional[float]) -> bool:
self.qga.settimeout(timeout)
try:
self.qga.ping()
@@ -149,37 +157,40 @@ class QemuGuestAgentClient:
return False
return True
- def fsfreeze(self, cmd):
+ def fsfreeze(self, cmd: str) -> object:
if cmd not in ['status', 'freeze', 'thaw']:
raise Exception('Invalid command: ' + cmd)
-
+ # Can be int (freeze, thaw) or GuestFsfreezeStatus (status)
return getattr(self.qga, 'fsfreeze' + '_' + cmd)()
- def fstrim(self, minimum=0):
- return getattr(self.qga, 'fstrim')(minimum=minimum)
+ def fstrim(self, minimum: int) -> Dict[str, object]:
+ # returns GuestFilesystemTrimResponse
+ ret = getattr(self.qga, 'fstrim')(minimum=minimum)
+ assert isinstance(ret, dict)
+ return ret
- def suspend(self, mode):
+ def suspend(self, mode: str) -> None:
if mode not in ['disk', 'ram', 'hybrid']:
raise Exception('Invalid mode: ' + mode)
try:
getattr(self.qga, 'suspend' + '_' + mode)()
# On error exception will raise
- except self.qga.timeout:
+ except TimeoutError:
# On success command will timed out
return
- def shutdown(self, mode='powerdown'):
+ def shutdown(self, mode: str = 'powerdown') -> None:
if mode not in ['powerdown', 'halt', 'reboot']:
raise Exception('Invalid mode: ' + mode)
try:
self.qga.shutdown(mode=mode)
- except self.qga.timeout:
- return
+ except TimeoutError:
+ pass
-def _cmd_cat(client, args):
+def _cmd_cat(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
if len(args) != 1:
print('Invalid argument')
print('Usage: cat <file>')
@@ -187,7 +198,7 @@ def _cmd_cat(client, args):
print(client.read(args[0]))
-def _cmd_fsfreeze(client, args):
+def _cmd_fsfreeze(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
usage = 'Usage: fsfreeze status|freeze|thaw'
if len(args) != 1:
print('Invalid argument')
@@ -201,13 +212,14 @@ def _cmd_fsfreeze(client, args):
ret = client.fsfreeze(cmd)
if cmd == 'status':
print(ret)
- elif cmd == 'freeze':
- print("%d filesystems frozen" % ret)
- else:
- print("%d filesystems thawed" % ret)
+ return
+ assert isinstance(ret, int)
+ verb = 'frozen' if cmd == 'freeze' else 'thawed'
+ print(f"{ret:d} filesystems {verb}")
-def _cmd_fstrim(client, args):
+
+def _cmd_fstrim(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
if len(args) == 0:
minimum = 0
else:
@@ -215,28 +227,25 @@ def _cmd_fstrim(client, args):
print(client.fstrim(minimum))
-def _cmd_ifconfig(client, args):
+def _cmd_ifconfig(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
assert not args
print(client.ifconfig())
-def _cmd_info(client, args):
+def _cmd_info(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
assert not args
print(client.info())
-def _cmd_ping(client, args):
- if len(args) == 0:
- timeout = 3
- else:
- timeout = float(args[0])
+def _cmd_ping(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
+ timeout = 3.0 if len(args) == 0 else float(args[0])
alive = client.ping(timeout)
if not alive:
print("Not responded in %s sec" % args[0])
sys.exit(1)
-def _cmd_suspend(client, args):
+def _cmd_suspend(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
usage = 'Usage: suspend disk|ram|hybrid'
if len(args) != 1:
print('Less argument')
@@ -249,7 +258,7 @@ def _cmd_suspend(client, args):
client.suspend(args[0])
-def _cmd_shutdown(client, args):
+def _cmd_shutdown(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
assert not args
client.shutdown()
@@ -257,12 +266,12 @@ def _cmd_shutdown(client, args):
_cmd_powerdown = _cmd_shutdown
-def _cmd_halt(client, args):
+def _cmd_halt(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
assert not args
client.shutdown('halt')
-def _cmd_reboot(client, args):
+def _cmd_reboot(client: QemuGuestAgentClient, args: Sequence[str]) -> None:
assert not args
client.shutdown('reboot')
@@ -270,7 +279,7 @@ def _cmd_reboot(client, args):
commands = [m.replace('_cmd_', '') for m in dir() if '_cmd_' in m]
-def send_command(address, cmd, args):
+def send_command(address: str, cmd: str, args: Sequence[str]) -> None:
if not os.path.exists(address):
print('%s not found' % address)
sys.exit(1)
@@ -296,7 +305,7 @@ def send_command(address, cmd, args):
globals()['_cmd_' + cmd](client, args)
-def main():
+def main() -> None:
address = os.environ.get('QGA_CLIENT_ADDRESS')
parser = argparse.ArgumentParser()
--
2.26.2
- [PATCH v2 25/72] scripts/qemu-ga-client: move to python/qemu/qmp/qemu_ga_client.py, (continued)
- [PATCH v2 25/72] scripts/qemu-ga-client: move to python/qemu/qmp/qemu_ga_client.py, John Snow, 2020/11/03
- [PATCH v2 17/72] scripts/qemu-ga-client: apply isort rules, John Snow, 2020/11/03
- [PATCH v2 14/72] scripts/qom-fuse: add redirection shim to python/qemu/qmp/qom-fuse.py, John Snow, 2020/11/03
- [PATCH v2 22/72] scripts/qemu-ga-client: apply (most) pylint rules, John Snow, 2020/11/03
- [PATCH v2 31/72] scripts/qmp-shell: fix exception handling, John Snow, 2020/11/03
- [PATCH v2 32/72] scripts/qmp-shell: fix connect method signature, John Snow, 2020/11/03
- [PATCH v2 30/72] scripts/qmp-shell: fix show_banner signature, John Snow, 2020/11/03
- [PATCH v2 35/72] scripts/qmp-shell: Use python3-style super(), John Snow, 2020/11/03
- [PATCH v2 34/72] scripts/qmp-shell: use @classmethod where appropriate, John Snow, 2020/11/03
- [PATCH v2 40/72] scripts/qmp-shell: rename one and two-letter variables, John Snow, 2020/11/03
- [PATCH v2 24/72] scripts/qmp/qemu-ga-client: add mypy type hints,
John Snow <=
- [PATCH v2 33/72] scripts/qmp-shell: remove shadowed variable from _print(), John Snow, 2020/11/03
- [PATCH v2 36/72] scripts/qmp-shell: declare verbose in __init__, John Snow, 2020/11/03
- [PATCH v2 39/72] scripts/qmp-shell: make QMPCompleter returns explicit, John Snow, 2020/11/03
- [PATCH v2 42/72] scripts/qmp-shell: explicitly chain exception context, John Snow, 2020/11/03
- [PATCH v2 37/72] scripts/qmp-shell: use triple-double-quote docstring style, John Snow, 2020/11/03
- [PATCH v2 43/72] scripts/qmp-shell: remove if-raise-else patterns, John Snow, 2020/11/03
- [PATCH v2 41/72] scripts/qmp-shell: fix shell history exception handling, John Snow, 2020/11/03
- [PATCH v2 44/72] scripts/qmp-shell: use isinstance() instead of type(), John Snow, 2020/11/03
- [PATCH v2 45/72] scripts/qmp-shell: use argparse, John Snow, 2020/11/03
- [PATCH v2 46/72] python/qmp: Fix type of SocketAddrT, John Snow, 2020/11/03