Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _run(self, cmd: str, *, timeout: int = 30, codec: str = "utf-8", decodeerrors: str = "strict"): # pylint: disable=unused-argument,line-too-long
"""
If Uboot is in Command-Line mode: Run command cmd and return it's
output.
Arguments:
cmd - Command to run
"""
# TODO: use codec, decodeerrors
# Check if Uboot is in command line mode
if self._status != 1:
return None
marker = gen_marker()
# Create multi-part command like we would do for a normal uboot.
# but since this simple uboot does not have an echo-command we will
# handle it's error message as an echo-output.
# additionally we are not able to get the command's return code and
# will always return 0.
cmp_command = "echo{marker}; {cmd}; echo{marker}".format(marker=marker, cmd=cmd)
self.console.sendline(cmp_command)
_, before, _, _ = self.console.expect(self.prompt, timeout=timeout)
data = self.re_vt100.sub(
'', before.decode('utf-8'), count=1000000
).replace("\r", "").split("\n")
data = data[1:]
data = data[data.index("Unknown command 'echo{}' - try 'help'".format(marker)) +1 :]
def _run(self, cmd: str, *, timeout: int = 30, codec: str = "utf-8", decodeerrors: str = "strict"): # pylint: disable=unused-argument,line-too-long
"""
Runs the specified command on the shell and returns the output.
Args:
cmd (str): command to run on the shell
timeout (int): optional, timeout in seconds
Returns:
Tuple[List[str],List[str], int]: if successful, None otherwise
"""
# FIXME: use codec, decodeerrors
marker = gen_marker()
# hide marker from expect
hidden_marker = '"{}""{}"'.format(marker[:4], marker[4:])
cmp_command = '''echo -o /cmd {cmd}; echo {marker}; sh /cmd; echo {marker} $?;'''.format(
cmd=shlex.quote(cmd), marker=hidden_marker)
if self._status == 1:
self.console.sendline(cmp_command)
_, _, match, _ = self.console.expect(
r'{marker}(.*){marker}\s+(\d+)\s+.*{prompt}'.format(
marker=marker, prompt=self.prompt),
timeout=timeout)
# Remove VT100 Codes and split by newline
data = self.re_vt100.sub('', match.group(1).decode('utf-8')).split('\r\n')[1:-1]
self.logger.debug("Received Data: %s", data)
# Get exit code
exitcode = int(match.group(2))
return (data, [], exitcode)
def _run(self, cmd: str, *, timeout: int = 30, codec: str = "utf-8", decodeerrors: str = "strict"): # pylint: disable=unused-argument,line-too-long
# TODO: use codec, decodeerrors
# TODO: Shell Escaping for the U-Boot Shell
marker = gen_marker()
cmp_command = """echo '{}''{}'; {}; echo "$?"; echo '{}''{}';""".format(
marker[:4],
marker[4:],
cmd,
marker[:4],
marker[4:],
)
if self._status == 1:
self.console.sendline(cmp_command)
_, before, _, _ = self.console.expect(self.prompt, timeout=timeout)
# Remove VT100 Codes and split by newline
data = self.re_vt100.sub(
'', before.decode('utf-8'), count=1000000
).replace("\r", "").split("\n")
self.logger.debug("Received Data: %s", data)
# Remove first element, the invoked cmd
def _start_xmodem_transfer(self, cmd):
"""
Start transfer command and synchronize until start of XMODEM stream.
We don't use _run() here because it expects a prompt, but we want to
read from the console directly into our XMODEM instance instead.
"""
marker = gen_marker()
marked_cmd = "echo '{}''{}'; {}".format(marker[:4], marker[4:], cmd)
self.console.sendline(marked_cmd)
self.console.expect(marker, timeout=30)
def _check_prompt(self):
"""
Internal function to check if we have a valid prompt.
It sets the internal _status to 1 or 0 based on the prompt detection.
"""
marker = gen_marker()
# hide marker from expect
hidden_marker = '"{}""{}"'.format(marker[:4], marker[4:])
self.console.sendline("echo {}".format(hidden_marker))
try:
self.console.expect("{}".format(marker), timeout=2)
self.console.expect(self.prompt, timeout=1)
self._status = 1
except TIMEOUT:
self._status = 0
raise
def _check_prompt(self):
"""
Internal function to check if we have a valid prompt
"""
marker = gen_marker()
# hide marker from expect
self.console.sendline("echo '{}''{}'".format(marker[:4], marker[4:]))
try:
self.console.expect(
r"{marker}\s+{prompt}".format(marker=marker, prompt=self.prompt),
timeout=5
)
self._status = 1
except TIMEOUT:
self._status = 0
raise
def _check_prompt(self):
"""
Internal function to check if we have a valid prompt.
It sets the internal _status to 1 or 0 based on the prompt detection.
"""
marker = gen_marker()
# hide marker from expect
self.console.sendline("echo '{}''{}'".format(marker[:4], marker[4:]))
try:
self.console.expect("{}".format(marker), timeout=2)
self.console.expect(self.prompt, timeout=1)
self._status = 1
except TIMEOUT:
self._status = 0
raise