Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
@step(args=['name'])
def boot(self, name):
"""Boot the default or a specific boot entry
Args:
name (str): name of the entry to boot"""
if name:
self.console.sendline("boot -v {}".format(name))
else:
self.console.sendline(self.boot_command)
@step()
def on(self):
"""Start the QEMU subprocess, accept the unix socket connection and
afterwards start the emulator using a QMP Command"""
if self.status:
return
self.logger.debug("Starting with: %s", self._cmd)
self._child = subprocess.Popen(
self._cmd, stdin=subprocess.PIPE, stdout=subprocess.PIPE)
# prepare for timeout handing
self._clientsocket, address = self._socket.accept()
self._clientsocket.setblocking(0)
self.logger.debug("new connection from %s", address)
try:
self.qmp = QMPMonitor(self._child.stdout, self._child.stdin)
@step(args=['data'])
def _write(self, data):
return self._clientsocket.send(data)
@step(args=['cmd'], result=True)
def run(self, cmd, timeout=30.0, codec="utf-8", decodeerrors="strict"):
return self._run(cmd, timeout=timeout, codec=codec, decodeerrors=decodeerrors)
@step(args=['cmd'])
def run(self, cmd: str, *, timeout: int = 30): # pylint: disable=unused-argument
return self._run(cmd, timeout=timeout)
@step(args=['cmd'], result=True)
def run(self, cmd, timeout=30):
"""
Runs the specified command on the shell and returns the output.
Args:
cmd (str): command to run on the shell
timeout (int): optional, how long to wait for completion
Returns:
Tuple[List[str],List[str], int]: if successful, None otherwise
"""
return self._run(cmd, timeout=timeout)
@step(args=['cmd'], result=True)
def run(self, cmd, codec="utf-8", decodeerrors="strict", timeout=None): # pylint: disable=unused-argument
return self._run(cmd, codec=codec, decodeerrors=decodeerrors)
@step(result=True)
def get_size(self):
with open(self.storage.path, 'rb') as dst:
dst.seek(0, os.SEEK_END)
size = dst.tell()
return size
@step(title='get', args=['remotefile', 'localfile'])
def _get(self, remotefile: str, localfile: str):
with open(localfile, 'wb') as fh:
buf = self._get_bytes(remotefile)
fh.write(buf)
@step()
def off(self):
self.output.set(False)