Lines Matching refs:self
47 def wrapper(self, *args): argument
48 was_root = self.device.shell(['id', '-un'])[0].strip() == 'root'
50 self.device.unroot()
51 self.device.wait()
54 func(self, *args)
57 self.device.root()
58 self.device.wait()
64 def setUp(self) -> None: argument
65 self.device = adb.get_device()
69 def test_smoke(self): argument
74 self.assertEqual(abb.returncode == 0, cmd.returncode == 0)
75 self.assertEqual(abb.stdout, cmd.stdout)
76 self.assertEqual(abb.stderr, cmd.stderr)
79 def _test_no_rebind(self, description, direction_list, direction, argument
82 self.assertEqual('', msg.strip(),
88 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
91 with self.assertRaises(subprocess.CalledProcessError):
94 self.assertFalse(re.search(r'tcp:5566.+tcp:6677', msg))
95 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
100 self.assertFalse(re.search(r'tcp:5566.+tcp:6655', msg))
101 self.assertTrue(re.search(r'tcp:5566.+tcp:6677', msg))
105 self.assertEqual('', msg.strip())
107 def test_forward_no_rebind(self): argument
108 self._test_no_rebind('forward', self.device.forward_list,
109 self.device.forward, self.device.forward_no_rebind,
110 self.device.forward_remove_all)
112 def test_reverse_no_rebind(self): argument
113 self._test_no_rebind('reverse', self.device.reverse_list,
114 self.device.reverse, self.device.reverse_no_rebind,
115 self.device.reverse_remove_all)
117 def test_forward(self): argument
118 msg = self.device.forward_list()
119 self.assertEqual('', msg.strip(),
121 self.device.forward('tcp:5566', 'tcp:6655')
122 msg = self.device.forward_list()
123 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
124 self.device.forward('tcp:7788', 'tcp:8877')
125 msg = self.device.forward_list()
126 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
127 self.assertTrue(re.search(r'tcp:7788.+tcp:8877', msg))
128 self.device.forward_remove('tcp:5566')
129 msg = self.device.forward_list()
130 self.assertFalse(re.search(r'tcp:5566.+tcp:6655', msg))
131 self.assertTrue(re.search(r'tcp:7788.+tcp:8877', msg))
132 self.device.forward_remove_all()
133 msg = self.device.forward_list()
134 self.assertEqual('', msg.strip())
136 def test_forward_old_protocol(self): argument
137 serialno = subprocess.check_output(self.device.adb_cmd + ['get-serialno']).strip()
139 msg = self.device.forward_list()
140 self.assertEqual('', msg.strip(),
148 msg = self.device.forward_list()
149 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
151 self.device.forward_remove_all()
152 msg = self.device.forward_list()
153 self.assertEqual('', msg.strip())
155 def test_forward_tcp_port_0(self): argument
156 self.assertEqual('', self.device.forward_list().strip(),
162 port = self.device.forward('tcp:0', 'tcp:8888').strip()
166 self.assertTrue(re.search(r'tcp:{}.+tcp:8888'.format(port),
167 self.device.forward_list()))
169 self.device.forward_remove_all()
171 def test_reverse(self): argument
172 msg = self.device.reverse_list()
173 self.assertEqual('', msg.strip(),
175 self.device.reverse('tcp:5566', 'tcp:6655')
176 msg = self.device.reverse_list()
177 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
178 self.device.reverse('tcp:7788', 'tcp:8877')
179 msg = self.device.reverse_list()
180 self.assertTrue(re.search(r'tcp:5566.+tcp:6655', msg))
181 self.assertTrue(re.search(r'tcp:7788.+tcp:8877', msg))
182 self.device.reverse_remove('tcp:5566')
183 msg = self.device.reverse_list()
184 self.assertFalse(re.search(r'tcp:5566.+tcp:6655', msg))
185 self.assertTrue(re.search(r'tcp:7788.+tcp:8877', msg))
186 self.device.reverse_remove_all()
187 msg = self.device.reverse_list()
188 self.assertEqual('', msg.strip())
190 def test_reverse_tcp_port_0(self): argument
191 self.assertEqual('', self.device.reverse_list().strip(),
197 port = self.device.reverse('tcp:0', 'tcp:8888').strip()
201 self.assertTrue(re.search(r'tcp:{}.+tcp:8888'.format(port),
202 self.device.reverse_list()))
204 self.device.reverse_remove_all()
206 def test_forward_reverse_echo(self): argument
217 self.device.forward(forward_spec, forward_spec)
220 self.device.reverse(forward_spec, reverse_spec)
251 self.assertEqual(data, server.makefile().read().encode("utf8"))
254 self.device.reverse_remove(forward_spec)
256 self.device.forward_remove(forward_spec)
260 def _interactive_shell(self, shell_args, input): argument
273 if not self.device.has_shell_protocol():
277 self.device.adb_cmd + ['shell'] + shell_args,
286 def test_cat(self): argument
288 out = self.device.shell(['cat', '/proc/uptime'])[0].strip()
290 self.assertEqual(len(elements), 2)
293 self.assertGreater(float(uptime), 0.0)
294 self.assertGreater(float(idle), 0.0)
296 def test_throws_on_failure(self): argument
297 self.assertRaises(adb.ShellError, self.device.shell, ['false'])
299 def test_output_not_stripped(self): argument
300 out = self.device.shell(['echo', 'foo'])[0]
301 self.assertEqual(out, 'foo' + self.device.linesep)
303 def test_shell_command_length(self): argument
305 if self.device.has_shell_protocol():
306 rc, out, err = self.device.shell_nocheck(['echo', 'x' * 16384])
307 self.assertEqual(rc, 0)
308 self.assertTrue(out == ('x' * 16384 + '\n'))
310 def test_shell_nocheck_failure(self): argument
311 rc, out, _ = self.device.shell_nocheck(['false'])
312 self.assertNotEqual(rc, 0)
313 self.assertEqual(out, '')
315 def test_shell_nocheck_output_not_stripped(self): argument
316 rc, out, _ = self.device.shell_nocheck(['echo', 'foo'])
317 self.assertEqual(rc, 0)
318 self.assertEqual(out, 'foo' + self.device.linesep)
320 def test_can_distinguish_tricky_results(self): argument
324 rc, out, _ = self.device.shell_nocheck(['echo', '-n', '1'])
325 self.assertEqual(rc, 0)
326 self.assertEqual(out, '1')
328 def test_line_endings(self): argument
333 output = self.device.shell(['uname'])[0]
334 self.assertEqual(output, 'Linux' + self.device.linesep)
336 def test_pty_logic(self): argument
353 test_cmd = self.device.adb_cmd + ['shell'] + args + ['[ -t 0 ]']
368 self.assertEqual((False, False), check_pty(['-T']))
371 if self.device.has_shell_protocol() and os.isatty(sys.stdin.fileno()):
374 self.assertEqual((False, False), check_pty([]))
377 self.assertEqual((True, False), check_pty(['-t']))
380 self.assertEqual((True, True), check_pty(['-t', '-t']))
383 self.assertEqual((True, True), check_pty(['-tt']))
387 self.assertEqual((True, True), check_pty(['-ttt']))
390 self.assertEqual((True, True), check_pty(['-ttx']))
393 self.assertEqual((True, True), check_pty(['-Ttt']))
396 self.assertEqual((False, False), check_pty(['-ttT']))
398 def test_shell_protocol(self): argument
406 if not self.device.has_shell_protocol():
410 result = self.device.shell_nocheck(
412 self.assertEqual(17, result[0])
413 self.assertEqual('foo' + self.device.linesep, result[1])
414 self.assertEqual('bar' + self.device.linesep, result[2])
416 self.assertEqual(17, self._interactive_shell([], b'exit 17'))
419 result = self.device.shell_nocheck(
421 self.assertEqual(0, result[0])
422 self.assertEqual('foo{0}bar{0}'.format(self.device.linesep), result[1])
423 self.assertEqual('', result[2])
425 self.assertEqual(0, self._interactive_shell(['-x'], b'exit 17'))
427 def test_non_interactive_sigint(self): argument
436 if not self.device.has_shell_protocol():
441 self.device.adb_cmd + shlex.split('shell echo $$; sleep 60'),
445 self.assertIsNone(sleep_proc.returncode, 'subprocess terminated early')
449 self.device.shell(proc_query)
455 while self.device.shell_nocheck(proc_query)[0] != 1:
456 self.assertFalse(time.time() > end_time,
459 def test_non_interactive_stdin(self): argument
461 if not self.device.has_shell_protocol():
472 proc = subprocess.Popen(self.device.adb_cmd + ['shell', 'cat'],
477 self.assertEqual(input.splitlines(), stdout.splitlines())
478 self.assertEqual(b'', stderr)
480 def test_sighup(self): argument
485 self.device.shell_nocheck(["echo", ">", log_path])
496 with self.device.shell_popen([script], kill_atexit=False,
500 self.assertEqual(b"Waiting\n", process.stdout.readline())
508 stdout, _ = self.device.shell(["cat", log_path])
509 self.assertEqual(stdout.strip(), "SIGHUP")
513 def disabled_test_exit_stress(self): argument
534 self.assertTrue(success)
536 def disabled_test_parallel(self): argument
568 def test_shell_escaping(self): argument
576 result = self.device.shell(
579 self.assertEqual(['', 'world'], result)
581 result = self.device.shell(
583 self.assertEqual(['hello', 'world'], result)
586 result = self.device.shell(shlex.split("'true && echo t'"))[0].strip()
587 self.assertEqual('t', result)
588 result = self.device.shell(
590 self.assertEqual('t', result)
593 result = self.device.shell(shlex.split('FOO=a BAR=b echo t'))[0].strip()
594 self.assertEqual('t', result)
595 result = self.device.shell(
597 self.assertEqual('123Linux', result)
599 def test_install_argument_escaping(self): argument
609 output = self.device.install(tf.name.decode("utf8"))
613 self.assertIn(file_suffix, output)
618 def _test_root(self): argument
619 message = self.device.root()
622 self.device.wait()
623 self.assertEqual('root', self.device.shell(['id', '-un'])[0].strip())
625 def _test_unroot(self): argument
626 self.device.unroot()
627 self.device.wait()
628 self.assertEqual('shell', self.device.shell(['id', '-un'])[0].strip())
630 def test_root_unroot(self): argument
632 if self.device.get_prop('ro.debuggable') != '1':
635 original_user = self.device.shell(['id', '-un'])[0].strip()
638 self._test_unroot()
639 self._test_root()
641 self._test_root()
642 self._test_unroot()
645 self.device.root()
647 self.device.unroot()
648 self.device.wait()
652 def test_tcpip_failure_raises(self): argument
657 self.assertRaises(
658 subprocess.CalledProcessError, self.device.tcpip, '')
659 self.assertRaises(
660 subprocess.CalledProcessError, self.device.tcpip, 'foo')
664 def test_get_prop(self): argument
665 self.assertEqual(self.device.get_prop('init.svc.adbd'), 'running')
667 def test_set_prop(self): argument
670 self.device.shell(['setprop', prop_name, '""'])
673 self.device.set_prop(prop_name, str(val))
674 self.assertEqual(
675 self.device.shell(['getprop', prop_name])[0].strip(), str(val))
685 def __init__(self, handle, checksum): argument
686 self.handle = handle
687 self.checksum = checksum
688 self.full_path = handle.name
689 self.base_name = os.path.basename(self.full_path)
693 def __init__(self, checksum, full_path): argument
694 self.checksum = checksum
695 self.full_path = full_path
696 self.base_name = posixpath.basename(self.full_path)
743 def setUp(self): argument
745 self.previous_env = os.environ.get("ADB_COMPRESSION")
746 os.environ["ADB_COMPRESSION"] = self.compression
748 def tearDown(self): argument
749 if self.previous_env is None:
752 os.environ["ADB_COMPRESSION"] = self.previous_env
754 def _verify_remote(self, checksum, remote_path): argument
755 dev_md5, _ = self.device.shell(['md5sum', remote_path])[0].split()
756 self.assertEqual(checksum, dev_md5)
758 def _verify_local(self, checksum, local_path): argument
761 self.assertEqual(host_md5, checksum)
763 def test_push(self): argument
771 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
772 self.device.push(local=tmp.name, remote=self.DEVICE_TEMP_FILE)
774 self._verify_remote(compute_md5(rand_str), self.DEVICE_TEMP_FILE)
775 self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
779 def test_push_dir(self): argument
781 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
782 self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
792 self.device.push(host_dir, self.DEVICE_TEMP_DIR)
795 remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
798 self._verify_remote(temp_file.checksum, remote_path)
799 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
804 def disabled_test_push_empty(self): argument
806 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
807 self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
819 self.device.push(empty_dir_path, self.DEVICE_TEMP_DIR)
821 remote_path = os.path.join(self.DEVICE_TEMP_DIR, "empty")
823 rc, _, _ = self.device.shell_nocheck(test_empty_cmd)
825 self.assertEqual(rc, 0)
826 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
832 def test_push_symlink(self): argument
850 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
851 self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
852 self.device.push(symlink_path, self.DEVICE_TEMP_DIR)
853 rc, out, _ = self.device.shell_nocheck(
854 ['cat', posixpath.join(self.DEVICE_TEMP_DIR, 'symlink')])
855 self.assertEqual(0, rc)
856 self.assertEqual(out.strip(), 'foo')
861 def test_multiple_push(self): argument
867 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
868 self.device.shell(['mkdir', self.DEVICE_TEMP_DIR])
883 self.device._simple_call(['push'] + paths + [self.DEVICE_TEMP_DIR])
886 remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
888 self._verify_remote(temp_file.checksum, remote_path)
891 remote_path = posixpath.join(self.DEVICE_TEMP_DIR,
895 self._verify_remote(temp_file.checksum, remote_path)
898 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
904 def test_push_error_reporting(self): argument
913 self.device.push(local=tmp_file.name, remote='/system/')
914 self.fail('push should not have succeeded')
918 self.assertTrue(b'Permission denied' in output or
922 def test_push_directory_creation(self): argument
930 remote_path = self.DEVICE_TEMP_DIR + '/test_push_directory_creation'
931 self.device.shell(['rm', '-rf', remote_path])
934 self.device.push(local=tmp_file.name, remote=remote_path)
936 def disabled_test_push_multiple_slash_root(self): argument
946 remote_path = '/' + self.DEVICE_TEMP_DIR + '/test_push_multiple_slash_root'
947 self.device.shell(['rm', '-rf', remote_path])
948 self.device.push(local=tmp_file.name, remote=remote_path)
950 def _test_pull(self, remote_file, checksum): argument
953 self.device.pull(remote=remote_file, local=tmp_write.name)
957 self.assertEqual(checksum, host_md5)
961 def test_pull_error_reporting(self): argument
962 self.device.shell(['touch', self.DEVICE_TEMP_FILE])
963 self.device.shell(['chmod', 'a-rwx', self.DEVICE_TEMP_FILE])
966 output = self.device.pull(remote=self.DEVICE_TEMP_FILE, local='x')
970 self.assertIn(b'Permission denied', output)
972 self.device.shell(['rm', '-f', self.DEVICE_TEMP_FILE])
974 def test_pull(self): argument
977 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_FILE])
979 'of={}'.format(self.DEVICE_TEMP_FILE), 'bs=1024',
981 self.device.shell(cmd)
982 dev_md5, _ = self.device.shell(['md5sum', self.DEVICE_TEMP_FILE])[0].split()
983 self._test_pull(self.DEVICE_TEMP_FILE, dev_md5)
984 self.device.shell_nocheck(['rm', self.DEVICE_TEMP_FILE])
986 def test_pull_dir(self): argument
991 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
992 self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
996 self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
998 self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
1002 host_dir, posixpath.basename(self.DEVICE_TEMP_DIR),
1004 self._verify_local(temp_file.checksum, host_path)
1006 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1011 def test_pull_dir_symlink(self): argument
1026 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1027 self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
1031 self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
1033 self.device.pull(remote=self.DEVICE_TEMP_DIR, local=symlink)
1037 real_dir, posixpath.basename(self.DEVICE_TEMP_DIR),
1039 self._verify_local(temp_file.checksum, host_path)
1041 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1046 def test_pull_dir_symlink_collision(self): argument
1054 tmp_dirname = os.path.basename(self.DEVICE_TEMP_DIR)
1059 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1060 self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
1064 self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
1066 self.device.pull(remote=self.DEVICE_TEMP_DIR, local=host_dir)
1070 self._verify_local(temp_file.checksum, host_path)
1072 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1077 def test_pull_dir_nonexistent(self): argument
1083 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1084 self.device.shell(['mkdir', '-p', self.DEVICE_TEMP_DIR])
1088 self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=32)
1090 self.device.pull(remote=self.DEVICE_TEMP_DIR, local=dest_dir)
1094 self._verify_local(temp_file.checksum, host_path)
1096 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1102 def disabled_test_pull_symlink_dir(self): argument
1107 remote_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'contents')
1108 remote_links = posixpath.join(self.DEVICE_TEMP_DIR, 'links')
1109 remote_symlink = posixpath.join(self.DEVICE_TEMP_DIR, 'symlink')
1111 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1112 self.device.shell(['mkdir', '-p', remote_dir, remote_links])
1113 self.device.shell(['ln', '-s', remote_links, remote_symlink])
1117 self.device, in_dir=remote_dir, num_files=32)
1120 self.device.shell(
1124 self.device.pull(remote=remote_symlink, local=host_dir)
1129 self._verify_local(temp_file.checksum, host_path)
1131 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1136 def test_pull_empty(self): argument
1141 remote_empty_path = posixpath.join(self.DEVICE_TEMP_DIR, 'empty')
1142 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1143 self.device.shell(['mkdir', '-p', remote_empty_path])
1145 self.device.pull(remote=remote_empty_path, local=host_dir)
1146 self.assertTrue(os.path.isdir(os.path.join(host_dir, 'empty')))
1151 def test_multiple_pull(self): argument
1157 subdir = posixpath.join(self.DEVICE_TEMP_DIR, 'subdir')
1158 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1159 self.device.shell(['mkdir', '-p', subdir])
1163 self.device, in_dir=self.DEVICE_TEMP_DIR, num_files=4)
1166 self.device, in_dir=subdir, num_files=4, prefix='subdir_')
1170 self.device._simple_call(['pull'] + paths + [host_dir])
1174 self._verify_local(temp_file.checksum, local_path)
1180 self._verify_local(subdir_temp_file.checksum, local_path)
1182 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1187 def verify_sync(self, device, temp_files, device_dir): argument
1194 self.assertEqual(temp_file.checksum, dev_md5)
1196 def test_sync(self): argument
1203 full_dir_path = base_dir + self.DEVICE_TEMP_DIR
1212 device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1222 self.verify_sync(device, temp_files, self.DEVICE_TEMP_DIR)
1229 def test_push_sync(self): argument
1236 device_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'sync_src_dst')
1244 self.verify_sync(device, temp_files, device_dir)
1246 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1251 def test_push_sync_multiple(self): argument
1258 device_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'sync_src_dst')
1268 self.verify_sync(device, temp_files, device_dir)
1270 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1276 def test_push_dry_run_nonexistent_file(self): argument
1281 device_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'push_dry_run')
1284 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1285 self.device.shell(['mkdir', '-p', device_dir])
1293 self.device._simple_call(['push', '-n', host_file, device_file])
1294 rc, _, _ = self.device.shell_nocheck(['[', '-e', device_file, ']'])
1295 self.assertNotEqual(0, rc)
1297 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1302 def test_push_dry_run_existent_file(self): argument
1308 device_dir = posixpath.join(self.DEVICE_TEMP_DIR, 'push_dry_run')
1311 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1312 self.device.shell(['mkdir', '-p', device_dir])
1313 self.device.shell(['echo', 'foo', '>', device_file])
1320 self.device._simple_call(['push', '-n', host_file, device_file])
1321 stdout, stderr = self.device.shell(['cat', device_file])
1322 self.assertEqual(stdout.strip(), "foo")
1324 self.device.shell(['rm', '-rf', self.DEVICE_TEMP_DIR])
1329 def test_unicode_paths(self): argument
1333 self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
1339 self.device.push(tf.name, remote_path)
1341 self.assertFalse(os.path.exists(tf.name))
1344 output = self.device.shell(
1346 self.assertEqual(remote_path, output)
1349 self.device.pull(remote_path, tf.name)
1350 self.assertTrue(os.path.exists(tf.name))
1352 self.device.shell(['rm', '-f', '/data/local/tmp/adb-test-*'])
1372 def _get_device_state(self, serialno): argument
1373 output = subprocess.check_output(self.device.adb_cmd + ['devices'])
1380 def disabled_test_killed_when_pushing_a_large_file(self): argument
1388 serialno = subprocess.check_output(self.device.adb_cmd + ['get-serialno']).strip()
1395 subproc = subprocess.Popen(self.device.adb_cmd + ['push', file_path, '/data/local/tmp'])
1398 subprocess.check_call(self.device.adb_cmd + ['kill-server'])
1409 self.assertEqual(self._get_device_state(serialno), 'device')
1411 def disabled_test_killed_when_pulling_a_large_file(self): argument
1418 serialno = subprocess.check_output(self.device.adb_cmd + ['get-serialno']).strip()
1422 self.device.shell(['dd', 'if=/dev/zero', 'of=/data/local/tmp/tmp_large_file',
1425 subproc = subprocess.Popen(self.device.adb_cmd +
1429 subprocess.check_call(self.device.adb_cmd + ['kill-server'])
1439 self.assertEqual(self._get_device_state(serialno), 'device')
1442 def test_packet_size_regression(self): argument
1455 rc, stdout, _ = self.device.shell_nocheck(cmd)
1457 self.assertEqual(0, rc)
1460 self.assertEqual(length, len(stdout) - 4)
1461 self.assertEqual(stdout, "\0" * length + "foo\n")
1463 def test_zero_packet(self): argument
1470 local_port = int(self.device.forward("tcp:0", "tcp:12345"))
1475 rc, stdout, stderr = self.device.shell_nocheck(cmd)
1486 self.assertEqual(size, bytesWritten)
1488 self.assertEqual(b"foo\n", readBytes)
1492 self.device.forward_remove("tcp:{}".format(local_port))
1496 def test_socket_flush(self): argument
1519 self.assertEqual(b"OKAY", response)
1525 self.assertEqual(b"OKAY", response)
1548 self.assertEqual(1024 * 1024 + len("foo\n"), len(received))
1553 def test_framebuffer(self): argument
1555 output = subprocess.check_output(self.device.adb_cmd + ["raw", "framebuffer:"])
1556 self.assertFalse(len(output) == 0)
1606 def __init__(self, *args, **kwds): argument
1607 super(CONSOLE_SCREEN_BUFFER_INFOEX, self).__init__(
1609 self.cbSize = ctypes.sizeof(self)
1737 def test_unicode_output(self): argument
1752 process = self.device.shell_popen(['echo', '"' + unicode_string + '"'])
1756 self.assertEqual(unicode_string, console_output)
1764 def test_devices(self): argument
1767 self.assertEqual(len(lines), 3)
1769 self.assertTrue(self.serial in line)
1770 self.assertFalse("{" in line)
1771 self.assertFalse("}" in line)
1772 self.assertTrue("device" in line)
1773 self.assertFalse("product" in line)
1774 self.assertFalse("transport" in line)
1776 def test_devices_l(self): argument
1779 self.assertEqual(len(lines), 3)
1781 self.assertTrue(self.serial in line)
1782 self.assertFalse("{" in line)
1783 self.assertFalse("}" in line)
1784 self.assertTrue("device" in line)
1785 self.assertTrue("product" in line)
1786 self.assertTrue("transport" in line)
1788 def test_track_devices(self): argument
1793 self.assertFalse("{" in output)
1794 self.assertFalse("}" in output)
1795 self.assertTrue(self.serial in output)
1796 self.assertTrue("device" in output)
1797 self.assertFalse("product" in output)
1798 self.assertFalse("transport" in output)
1801 def test_track_devices_l(self): argument
1806 self.assertFalse("{" in output)
1807 self.assertFalse("}" in output)
1808 self.assertTrue(self.serial in output)
1809 self.assertTrue("device" in output)
1810 self.assertTrue("product" in output)
1811 self.assertTrue("transport" in output)
1814 def test_track_devices_proto_text(self): argument
1819 self.assertTrue("{" in output)
1820 self.assertTrue("}" in output)
1821 self.assertTrue(self.serial in output)
1822 self.assertTrue("device" in output)
1823 self.assertTrue("product" in output)
1824 self.assertTrue("connection_type" in output)
1827 def test_track_devices_proto_binary(self): argument
1837 self.assertTrue(device.serial == self.serial)
1838 self.assertFalse(device.bus_address == "")
1839 self.assertFalse(device.product == "")
1840 self.assertFalse(device.model == "")
1841 self.assertFalse(device.device == "")
1842 self.assertTrue(device.negotiated_speed == int(device.negotiated_speed))
1843 self.assertTrue(device.max_speed == int(device.max_speed))
1844 self.assertTrue(device.transport_id == int(device.transport_id))
1852 def test_track_app_appinfo(self): argument
1872 self.assertTrue(app.debuggable)
1873 self.assertTrue("adb.test.app1" in app.package_names)
1874 self.assertTrue("adb.test.app2" in app.package_names)
1878 self.assertTrue(app.debuggable)
1879 self.assertTrue("adb.test.app1" in app.package_names)
1881 self.assertTrue(foundAdbAppDefProc)
1882 self.assertTrue(foundAdbAppOwnProc)