Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
# command failed (https://code.google.com/p/android/issues/detail?id=3254).
# Homogenise this behaviour by running the command then echoing the exit
# code of the executed command itself.
command = r'({}); echo "\n$?"'.format(command)
parts = ['adb']
if adb_server is not None:
parts += ['-H', adb_server]
if device is not None:
parts += ['-s', device]
parts += ['shell',
command if not as_root else su_cmd.format(quote(command))]
logger.debug(' '.join(quote(part) for part in parts))
try:
raw_output, error = check_output(parts, timeout, shell=False)
except subprocess.CalledProcessError as e:
raise TargetStableError(str(e))
if raw_output:
try:
output, exit_code, _ = raw_output.replace('\r\n', '\n').replace('\r', '\n').rsplit('\n', 2)
except ValueError:
exit_code, _ = raw_output.replace('\r\n', '\n').replace('\r', '\n').rsplit('\n', 1)
output = ''
else: # raw_output is empty
exit_code = '969696' # just because
output = ''
if check_exit_code:
exit_code = exit_code.strip()
re_search = AM_START_ERROR.findall(output)
def view(self, binfile):
check_output('{} {}'.format(self.kernelshark, binfile), shell=True)
def execute(self, command, timeout=None, check_exit_code=True,
as_root=False, strip_colors=True, will_succeed=False):
self.logger.debug(command)
if as_root and not self.connected_as_root:
if self.unrooted:
raise TargetStableError('unrooted')
password = self._get_password()
command = 'echo {} | sudo -S -- sh -c '.format(quote(password)) + quote(command)
ignore = None if check_exit_code else 'all'
try:
return check_output(command, shell=True, timeout=timeout, ignore=ignore)[0]
except subprocess.CalledProcessError as e:
message = 'Got exit code {}\nfrom: {}\nOUTPUT: {}'.format(
e.returncode, command, e.output)
if will_succeed:
raise TargetTransientError(message)
else:
raise TargetStableError(message)
port_string = '-P {}'.format(quote(str(self.port))) if (self.port and self.port != 22) else ''
keyfile_string = '-i {}'.format(quote(self.keyfile)) if self.keyfile else ''
options = " ".join(["-o {}={}".format(key, val)
for key, val in self.options.items()])
paths = ' '.join(map(quote, paths))
command = '{} {} -r {} {} {}'.format(scp,
options,
keyfile_string,
port_string,
paths)
command_redacted = command
logger.debug(command)
if self.password:
command, command_redacted = _give_password(self.password, command)
try:
check_output(command, timeout=timeout, shell=True)
except subprocess.CalledProcessError as e:
raise_from(HostError("Failed to copy file with '{}'. Output:\n{}".format(
command_redacted, e.output)), None)
except TimeoutError as e:
raise TimeoutError(command_redacted, e.output)
def fastboot_command(command, timeout=None, device=None):
_check_env()
target = '-s {}'.format(quote(device)) if device else ''
full_command = 'fastboot {} {}'.format(target, command)
logger.debug(full_command)
output, _ = check_output(full_command, timeout, shell=True)
return output
def adb_command(device, command, timeout=None, adb_server=None):
full_command = get_adb_command(device, command, adb_server)
logger.debug(full_command)
output, _ = check_output(full_command, timeout, shell=True)
return output
tries = 0
output = None
while tries <= attempts:
tries += 1
if device:
if "." in device: # Connect is required only for ADB-over-IP
# ADB does not automatically remove a network device from it's
# devices list when the connection is broken by the remote, so the
# adb connection may have gone "stale", resulting in adb blocking
# indefinitely when making calls to the device. To avoid this,
# always disconnect first.
adb_disconnect(device, adb_server)
adb_cmd = get_adb_command(None, 'connect', adb_server)
command = '{} {}'.format(adb_cmd, quote(device))
logger.debug(command)
output, _ = check_output(command, shell=True, timeout=timeout)
if _ping(device, adb_server):
break
time.sleep(10)
else: # did not connect to the device
message = 'Could not connect to {}'.format(device or 'a device')
if output:
message += '; got: "{}"'.format(output)
raise HostError(message)