Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _StartEmulator(self, timer,
net_type, new_process_group, window_scale,
with_audio, with_boot_anim):
"""Start emulator or user mode android."""
if not self.emulator_adb_port:
self.emulator_adb_port = portpicker.PickUnusedPort()
if not self.emulator_telnet_port:
self.emulator_telnet_port = portpicker.PickUnusedPort()
if not self.device_serial:
self.device_serial = 'localhost:%s' % self.emulator_adb_port
emulator_binary = os.path.abspath(
self.android_platform.GetEmulator(
self._metadata_pb.emulator_architecture,
self._metadata_pb.emulator_type))
pipe_dir = self._TempDir('pipe_trav')
exec_dir = self._SessionImagesDir()
self._emulator_env = self._MakeEmulatorEnv(os.environ)
if (self._metadata_pb.emulator_type in
[emulator_meta_data_pb2.EmulatorMetaDataPb.QEMU,
def ConnectDevice(self):
"""Connects the device to the adb server.
Returns:
True on Success, False otherwise.
"""
assert self._CanConnect()
if FLAGS.skip_connect_device:
return True
connect_args = [self.android_platform.real_adb,
'connect',
'localhost:%s' % self.emulator_adb_port]
logging.info('Connecting adb server to device: %s', connect_args)
connect_task = None
if not self.adb_server_port:
self.adb_server_port = portpicker.PickUnusedPort()
elif self.adb_server_port < 0 or self.adb_server_port > 65535:
logging.warn('Invalid adb server port %d, skip connecting',
self.adb_server_port)
return
try:
logging.info('Starting: %s', connect_args)
connect_task = common.SpawnAndWaitWithRetry(
connect_args,
proc_output=True,
exec_env=self._AdbEnv(),
timeout_seconds=ADB_SHORT_TIMEOUT_SECONDS,
retries=5)
logging.info('Done: %s', connect_args)
except common.SpawnError:
return False
def _AdbEnv(self):
"""Prepare environment for running adb."""
env = {}
if not self.adb_server_port:
self.adb_server_port = portpicker.PickUnusedPort()
if not self.device_serial:
self.device_serial = 'localhost:%s' % self.emulator_adb_port
env['ANDROID_ADB_SERVER_PORT'] = str(self.adb_server_port)
env['ANDROID_ADB'] = self.android_platform.real_adb
env['ANDROID_SERIAL'] = self.device_serial
env['ADB_LIBUSB'] = '0'
if self._emulator_env and 'HOME' in self._emulator_env:
env['HOME'] = self._emulator_env['HOME']
test_output_dir = os.environ.get('TEST_UNDECLARED_OUTPUTS_DIR')
if test_output_dir:
env['TMPDIR'] = test_output_dir
return env
def _StartEmulator(self, timer,
net_type, new_process_group, window_scale,
with_audio, with_boot_anim):
"""Start emulator or user mode android."""
if not self.emulator_adb_port:
self.emulator_adb_port = portpicker.PickUnusedPort()
if not self.emulator_telnet_port:
self.emulator_telnet_port = portpicker.PickUnusedPort()
if not self.device_serial:
self.device_serial = 'localhost:%s' % self.emulator_adb_port
emulator_binary = os.path.abspath(
self.android_platform.GetEmulator(
self._metadata_pb.emulator_architecture,
self._metadata_pb.emulator_type))
pipe_dir = self._TempDir('pipe_trav')
exec_dir = self._SessionImagesDir()
self._emulator_env = self._MakeEmulatorEnv(os.environ)
if (self._metadata_pb.emulator_type in
[emulator_meta_data_pb2.EmulatorMetaDataPb.QEMU,
emulator_meta_data_pb2.EmulatorMetaDataPb.QEMU2]):
self._PrepareQemuArgs(emulator_binary, net_type, window_scale,
self._emulator_cmd = emulator_cmd
self._http = httplib2.Http()
self.__running = False
self._tmp_dir = tempfile.mkdtemp(dir=working_directory)
self._project_directory = os.path.join(self._tmp_dir, self._project_id)
p = subprocess.Popen([emulator_cmd,
'create',
'--project_id=%s' % self._project_id,
self._project_directory])
if p.wait() != 0:
raise IOError('could not create project in directory: %s'
% self._project_directory)
# Start the emulator and wait for it to start responding to requests.
port = portpicker.PickUnusedPort()
self._host = 'http://localhost:%d' % port
cmd = [self._emulator_cmd, 'start', '--port=%d' % port]
cmd.extend(_DEFAULT_EMULATOR_OPTIONS)
if start_options:
cmd.extend(start_options)
cmd.append(self._project_directory)
subprocess.Popen(cmd)
if not self._WaitForStartup(deadline):
raise IOError('emulator did not respond within %ds' % deadline)
endpoint = '%s/v1/projects/%s' % (self._host, self._project_id)
self.__datastore = connection.Datastore(project_endpoint=endpoint)
self.__running = True
logging.warning(
'The default encoding of your local Python interpreter is set to %r '
'while App Engine\'s production environment uses %r; as a result '
'your code may behave differently when deployed.',
sys.getdefaultencoding(), constants.PROD_DEFAULT_ENCODING)
if options.port == 0:
logging.warn('DEFAULT_VERSION_HOSTNAME will not be set correctly with '
'--port=0')
_setup_environ(configuration.app_id)
# grpc_proxy is only needed for python2 because remote_api_stub.py is
# imported in local python runtime sandbox. For more details, see
# grpc_proxy_util.py.
grpc_proxy_port = portpicker.PickUnusedPort()
self._dispatcher = dispatcher.Dispatcher(
configuration, options.host, options.port, options.auth_domain,
constants.LOG_LEVEL_TO_RUNTIME_CONSTANT[options.log_level],
self._create_php_config(options),
self._create_python_config(options, grpc_proxy_port),
self._create_java_config(options),
self._create_go_config(options),
self._create_custom_config(options),
self._create_cloud_sql_config(options),
self._create_vm_config(options),
self._create_module_to_setting(options.max_module_instances,