Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""Waits for Android framework to broadcast ACTION_BOOT_COMPLETED.
This function times out after 15 minutes.
Args:
timeout: float, the number of seconds to wait before timing out.
If not specified, no timeout takes effect.
"""
timeout_start = time.time()
self.adb.wait_for_device(timeout=timeout)
while time.time() < timeout_start + timeout:
try:
if self.is_boot_completed():
return
except (adb.AdbError, adb.AdbTimeoutError):
# adb shell calls may fail during certain period of booting
# process, which is normal. Ignoring these errors.
pass
time.sleep(5)
raise DeviceError(self, 'Booting process timed out')
def _get_persist_command(self):
"""Check availability and return path of command if available."""
for command in [_SETSID_COMMAND, _NOHUP_COMMAND]:
try:
if command in self._adb.shell(['which',
command]).decode('utf-8'):
return command
except adb.AdbError:
continue
self.log.warning(
'No %s and %s commands available to launch instrument '
'persistently, tests that depend on UiAutomator and '
'at the same time performs USB disconnection may fail',
_SETSID_COMMAND, _NOHUP_COMMAND)
return ''
logpersist_warning = ('%s encountered an error enabling persistent'
' logs, logs may not get saved.')
# Android L and older versions do not have logpersist installed,
# so check that the logpersist scripts exists before trying to use
# them.
if not self._ad.adb.has_shell_command('logpersist.start'):
logging.warning(logpersist_warning, self)
return
try:
# Disable adb log spam filter for rootable devices. Have to stop
# and clear settings first because 'start' doesn't support --clear
# option before Android N.
self._ad.adb.shell('logpersist.stop --clear')
self._ad.adb.shell('logpersist.start')
except adb.AdbError:
logging.warning(logpersist_warning, self)
break
finally:
# Note, communicate will not contain any buffered output.
(unexpected_out, err) = proc.communicate()
if unexpected_out:
out = '[unexpected stdout] %s' % unexpected_out
for line in unexpected_out.splitlines():
handler(line)
ret = proc.returncode
logging.debug('cmd: %s, stdout: %s, stderr: %s, ret: %s',
utils.cli_cmd_to_string(args), out, err, ret)
if ret == 0:
return err
else:
raise AdbError(cmd=args, stdout=out, stderr=err, ret_code=ret)
(ret, out, err) = utils.run_command(args,
shell=shell,
timeout=timeout)
except psutil.TimeoutExpired:
raise AdbTimeoutError(cmd=args,
timeout=timeout,
serial=self.serial)
if stderr:
stderr.write(err)
logging.debug('cmd: %s, stdout: %s, stderr: %s, ret: %s',
utils.cli_cmd_to_string(args), out, err, ret)
if ret == 0:
return out
else:
raise AdbError(cmd=args,
stdout=out,
stderr=err,
ret_code=ret,
serial=self.serial)