1import time 2 3import libcxx.test.executor 4 5from libcxx.android import adb 6from lit.util import executeCommand # pylint: disable=import-error 7 8 9class AdbExecutor(libcxx.test.executor.RemoteExecutor): 10 def __init__(self, serial=None): 11 # TODO(danalbert): Should factor out the shared pieces of SSHExecutor 12 # so we don't have this ugly parent constructor... 13 super(AdbExecutor, self).__init__() 14 self.serial = serial 15 self.local_run = executeCommand 16 17 def _remote_temp(self, is_dir): 18 dir_arg = '-d' if is_dir else '' 19 cmd = 'mktemp -q {} /data/local/tmp/libcxx.XXXXXXXXXX'.format(dir_arg) 20 _, temp_path, err, exitCode = self._execute_command_remote([cmd]) 21 temp_path = temp_path.strip() 22 if exitCode != 0: 23 raise RuntimeError(err) 24 return temp_path 25 26 def _copy_in_file(self, src, dst): # pylint: disable=no-self-use 27 adb.push(src, dst) 28 29 def _execute_command_remote(self, cmd, remote_work_dir='.', env=None): 30 adb_cmd = ['adb', 'shell'] 31 if self.serial: 32 adb_cmd.extend(['-s', self.serial]) 33 34 delimiter = 'x' 35 probe_cmd = ' '.join(cmd) + '; echo {}$?'.format(delimiter) 36 37 env_cmd = [] 38 if env is not None: 39 env_cmd = ['env'] + ['%s=%s' % (k, v) for k, v in env.items()] 40 41 remote_cmd = ' '.join(env_cmd + [probe_cmd]) 42 if remote_work_dir != '.': 43 remote_cmd = 'cd {} && {}'.format(remote_work_dir, remote_cmd) 44 45 adb_cmd.append(remote_cmd) 46 47 # Tests will commonly fail with ETXTBSY. Possibly related to this bug: 48 # https://code.google.com/p/android/issues/detail?id=65857. Work around 49 # it by just waiting a second and then retrying. 50 for _ in range(10): 51 out, err, exit_code = self.local_run(adb_cmd) 52 if 'Text file busy' in out: 53 time.sleep(1) 54 else: 55 out, delim, rc_str = out.rpartition(delimiter) 56 if delim == '': 57 continue 58 59 out = out.strip() 60 try: 61 exit_code = int(rc_str) 62 break 63 except ValueError: 64 continue 65 return adb_cmd, out, err, exit_code 66