apply TRY lint
This commit is contained in:
@@ -4,6 +4,8 @@ import socket
|
||||
from pathlib import Path
|
||||
from time import sleep
|
||||
|
||||
from clan_cli.errors import ClanError
|
||||
|
||||
|
||||
# qga is almost like qmp, but not quite, because:
|
||||
# - server doesn't send initial message
|
||||
@@ -16,9 +18,10 @@ class QgaSession:
|
||||
for _ in range(100):
|
||||
try:
|
||||
self.sock.connect(str(socket_file))
|
||||
return
|
||||
except ConnectionRefusedError:
|
||||
sleep(0.1)
|
||||
else:
|
||||
return
|
||||
self.sock.connect(str(socket_file))
|
||||
|
||||
def get_response(self) -> dict:
|
||||
@@ -59,7 +62,7 @@ class QgaSession:
|
||||
result = self.get_response()
|
||||
if "error" in result and result["error"]["desc"].startswith("PID"):
|
||||
msg = "PID could not be found"
|
||||
raise Exception(msg)
|
||||
raise ClanError(msg)
|
||||
if result["return"]["exited"]:
|
||||
break
|
||||
sleep(0.1)
|
||||
@@ -77,5 +80,5 @@ class QgaSession:
|
||||
)
|
||||
if check and exitcode != 0:
|
||||
msg = f"Command on guest failed\nCommand: {cmd}\nExitcode {exitcode}\nStdout: {stdout}\nStderr: {stderr}"
|
||||
raise Exception(msg)
|
||||
raise ClanError(msg)
|
||||
return exitcode, stdout, stderr
|
||||
|
||||
@@ -16,6 +16,8 @@ import socket
|
||||
import types
|
||||
from typing import Any
|
||||
|
||||
from clan_cli.errors import ClanError
|
||||
|
||||
|
||||
class QMPError(Exception):
|
||||
"""
|
||||
@@ -159,7 +161,6 @@ class QEMUMonitorProtocol:
|
||||
) -> None:
|
||||
# Implement context manager exit function.
|
||||
self.close()
|
||||
return False
|
||||
|
||||
def connect(self, negotiate: bool = True) -> dict[str, Any] | None:
|
||||
"""
|
||||
@@ -211,7 +212,7 @@ class QEMUMonitorProtocol:
|
||||
except OSError as err:
|
||||
if err.errno == errno.EPIPE:
|
||||
return None
|
||||
raise err
|
||||
raise
|
||||
resp = self.__json_read()
|
||||
self.logger.debug("<<< %s", resp)
|
||||
return resp
|
||||
@@ -242,7 +243,7 @@ class QEMUMonitorProtocol:
|
||||
"""
|
||||
ret = self.cmd(cmd, kwds)
|
||||
if "error" in ret:
|
||||
raise Exception(ret["error"]["desc"])
|
||||
raise ClanError(ret["error"]["desc"])
|
||||
return ret["return"]
|
||||
|
||||
def pull_event(self, wait: bool | float = False) -> dict[str, Any] | None:
|
||||
|
||||
Reference in New Issue
Block a user