OpenE2K/qemu-e2k
13
4
Fork 0
Code Issues 4 Pull Requests Projects 2 Releases Activity

scripts/qemu-ga-client: add mypy type hints

Browse Source 
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>
Message-id: 20210604155532.1499282-9-jsnow@redhat.com
Signed-off-by: John Snow <jsnow@redhat.com>
This commit is contained in:
John Snow 2021-06-04 11:55:29 -04:00
parent 1f6399393b
commit ca683d4a2f
1 changed files with 49 additions and 40 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
scripts/qmp/qemu-ga-client
View File

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