Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def execute(self, command, timeout=None, check_exit_code=False,
as_root=False, strip_colors=True, will_succeed=False):
try:
return adb_shell(self.device, command, timeout, check_exit_code,
as_root, adb_server=self.adb_server, su_cmd=self.su_cmd)
except TargetStableError as e:
if will_succeed:
raise TargetTransientError(e)
else:
raise
self.logger.debug('Waiting for the shell prompt.')
tty.expect(target.shell_prompt)
self.logger.debug('Waiting for IP address...')
wait_start_time = time.time()
try:
while True:
tty.sendline('ip addr list eth0')
time.sleep(1)
try:
tty.expect(r'inet ([1-9]\d*.\d+.\d+.\d+)', timeout=10)
return tty.match.group(1).decode('utf-8')
except pexpect.TIMEOUT:
pass # We have our own timeout -- see below.
if (time.time() - wait_start_time) > self.ready_timeout:
raise TargetTransientError('Could not acquire IP address.')
finally:
tty.sendline('exit') # exit shell created by "su" call at the start
if int(exit_code):
message = ('Got exit code {}\nfrom target command: {}\n'
'OUTPUT: {}\nSTDERR: {}\n')
raise TargetStableError(message.format(exit_code, command, output, error))
elif re_search:
message = 'Could not start activity; got the following:\n{}'
raise TargetStableError(message.format(re_search[0]))
else: # not all digits
if re_search:
message = 'Could not start activity; got the following:\n{}'
raise TargetStableError(message.format(re_search[0]))
else:
message = 'adb has returned early; did not get an exit code. '\
'Was kill-server invoked?\nOUTPUT:\n-----\n{}\n'\
'-----\nSTDERR:\n-----\n{}\n-----'
raise TargetTransientError(message.format(raw_output, error))
return output + error
def adb_disconnect(device, adb_server=None):
_check_env()
if not device:
return
if ":" in device and device in adb_list_devices(adb_server):
adb_cmd = get_adb_command(None, 'disconnect', adb_server)
command = "{} {}".format(adb_cmd, device)
logger.debug(command)
retval = subprocess.call(command, stdout=open(os.devnull, 'wb'), shell=True)
if retval:
raise TargetTransientError('"{}" returned {}'.format(command, retval))
def _check_ready(self):
"""
Check if the gem5 platform is ready
"""
if not self.ready:
raise TargetTransientError('Gem5 is not ready to interact yet')
port=None,
timeout=10,
original_prompt=None):
_check_env()
start_time = time.time()
while True:
conn = TelnetPxssh(original_prompt=original_prompt)
try:
conn.login(host, username, password, port=port, login_timeout=timeout)
break
except EOF:
timeout -= time.time() - start_time
if timeout <= 0:
message = 'Could not connect to {}; is the host name correct?'
raise TargetTransientError(message.format(host))
time.sleep(5)
conn.setwinsize(500, 200)
conn.sendline('')
conn.prompt()
conn.setecho(False)
return conn