1#!/usr/bin/env python3 2# 3# Copyright (C) 2015 The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16# 17"""Tests for the adb program itself. 18 19This differs from things in test_device.py in that there is no API for these 20things. Most of these tests involve specific error messages or the help text. 21""" 22 23import contextlib 24import os 25import random 26import select 27import socket 28import struct 29import subprocess 30import sys 31import threading 32import time 33import unittest 34import warnings 35 36def find_open_port(): 37 # Find an open port. 38 with socket.socket() as s: 39 s.bind(("localhost", 0)) 40 return s.getsockname()[1] 41 42@contextlib.contextmanager 43def fake_adbd(protocol=socket.AF_INET, port=0): 44 """Creates a fake ADB daemon that just replies with a CNXN packet.""" 45 46 serversock = socket.socket(protocol, socket.SOCK_STREAM) 47 serversock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 48 if protocol == socket.AF_INET: 49 serversock.bind(("127.0.0.1", port)) 50 else: 51 serversock.bind(("::1", port)) 52 serversock.listen(1) 53 54 # A pipe that is used to signal the thread that it should terminate. 55 readsock, writesock = socket.socketpair() 56 57 def _adb_packet(command: bytes, arg0: int, arg1: int, data: bytes) -> bytes: 58 bin_command = struct.unpack("I", command)[0] 59 buf = struct.pack("IIIIII", bin_command, arg0, arg1, len(data), 0, 60 bin_command ^ 0xffffffff) 61 buf += data 62 return buf 63 64 def _handle(sock): 65 with contextlib.closing(sock) as serversock: 66 rlist = [readsock, serversock] 67 cnxn_sent = {} 68 while True: 69 read_ready, _, _ = select.select(rlist, [], []) 70 for ready in read_ready: 71 if ready == readsock: 72 # Closure pipe 73 for f in rlist: 74 f.close() 75 return 76 elif ready == serversock: 77 # Server socket 78 conn, _ = ready.accept() 79 rlist.append(conn) 80 else: 81 # Client socket 82 data = ready.recv(1024) 83 if not data or data.startswith(b"OPEN"): 84 if ready in cnxn_sent: 85 del cnxn_sent[ready] 86 ready.shutdown(socket.SHUT_RDWR) 87 ready.close() 88 rlist.remove(ready) 89 continue 90 if ready in cnxn_sent: 91 continue 92 cnxn_sent[ready] = True 93 ready.sendall(_adb_packet(b"CNXN", 0x01000001, 1024 * 1024, 94 b"device::ro.product.name=fakeadb")) 95 96 port = serversock.getsockname()[1] 97 server_thread = threading.Thread(target=_handle, args=(serversock,)) 98 server_thread.start() 99 100 try: 101 yield port, writesock 102 finally: 103 writesock.close() 104 server_thread.join() 105 106 107@contextlib.contextmanager 108def adb_connect(unittest, serial): 109 """Context manager for an ADB connection. 110 111 This automatically disconnects when done with the connection. 112 """ 113 114 output = subprocess.check_output(["adb", "connect", serial]) 115 unittest.assertEqual(output.strip(), 116 "connected to {}".format(serial).encode("utf8")) 117 118 try: 119 yield 120 finally: 121 # Perform best-effort disconnection. Discard the output. 122 subprocess.Popen(["adb", "disconnect", serial], 123 stdout=subprocess.PIPE, 124 stderr=subprocess.PIPE).communicate() 125 126 127@contextlib.contextmanager 128def adb_server(): 129 """Context manager for an ADB server. 130 131 This creates an ADB server and returns the port it's listening on. 132 """ 133 134 port = find_open_port() 135 read_pipe, write_pipe = os.pipe() 136 137 if sys.platform == "win32": 138 import msvcrt 139 write_handle = msvcrt.get_osfhandle(write_pipe) 140 os.set_handle_inheritable(write_handle, True) 141 reply_fd = str(write_handle) 142 else: 143 os.set_inheritable(write_pipe, True) 144 reply_fd = str(write_pipe) 145 146 proc = subprocess.Popen(["adb", "-L", "tcp:localhost:{}".format(port), 147 "fork-server", "server", 148 "--reply-fd", reply_fd], close_fds=False) 149 try: 150 os.close(write_pipe) 151 greeting = os.read(read_pipe, 1024) 152 assert greeting == b"OK\n", repr(greeting) 153 yield port 154 finally: 155 proc.terminate() 156 proc.wait() 157 158 159class CommandlineTest(unittest.TestCase): 160 """Tests for the ADB commandline.""" 161 162 def test_help(self): 163 """Make sure we get _something_ out of help.""" 164 out = subprocess.check_output( 165 ["adb", "help"], stderr=subprocess.STDOUT) 166 self.assertGreater(len(out), 0) 167 168 def test_version(self): 169 """Get a version number out of the output of adb.""" 170 lines = subprocess.check_output(["adb", "version"]).splitlines() 171 version_line = lines[0] 172 self.assertRegex( 173 version_line, rb"^Android Debug Bridge version \d+\.\d+\.\d+$") 174 if len(lines) == 2: 175 # Newer versions of ADB have a second line of output for the 176 # version that includes a specific revision (git SHA). 177 revision_line = lines[1] 178 self.assertRegex( 179 revision_line, rb"^Revision [0-9a-f]{12}-android$") 180 181 def test_tcpip_error_messages(self): 182 """Make sure 'adb tcpip' parsing is sane.""" 183 proc = subprocess.Popen(["adb", "tcpip"], 184 stdout=subprocess.PIPE, 185 stderr=subprocess.STDOUT) 186 out, _ = proc.communicate() 187 self.assertEqual(1, proc.returncode) 188 self.assertIn(b"requires an argument", out) 189 190 proc = subprocess.Popen(["adb", "tcpip", "foo"], 191 stdout=subprocess.PIPE, 192 stderr=subprocess.STDOUT) 193 out, _ = proc.communicate() 194 self.assertEqual(1, proc.returncode) 195 self.assertIn(b"invalid port", out) 196 197 198class ServerTest(unittest.TestCase): 199 """Tests for the ADB server.""" 200 201 @staticmethod 202 def _read_pipe_and_set_event(pipe, event): 203 """Reads a pipe until it is closed, then sets the event.""" 204 pipe.read() 205 event.set() 206 207 def test_handle_inheritance(self): 208 """Test that launch_server() does not inherit handles. 209 210 launch_server() should not let the adb server inherit 211 stdin/stdout/stderr handles, which can cause callers of adb.exe to hang. 212 This test also runs fine on unix even though the impetus is an issue 213 unique to Windows. 214 """ 215 # This test takes 5 seconds to run on Windows: if there is no adb server 216 # running on the the port used below, adb kill-server tries to make a 217 # TCP connection to a closed port and that takes 1 second on Windows; 218 # adb start-server does the same TCP connection which takes another 219 # second, and it waits 3 seconds after starting the server. 220 221 # Start adb client with redirected stdin/stdout/stderr to check if it 222 # passes those redirections to the adb server that it starts. To do 223 # this, run an instance of the adb server on a non-default port so we 224 # don't conflict with a pre-existing adb server that may already be 225 # setup with adb TCP/emulator connections. If there is a pre-existing 226 # adb server, this also tests whether multiple instances of the adb 227 # server conflict on adb.log. 228 229 port = find_open_port() 230 231 try: 232 # We get warnings for unclosed files for the subprocess's pipes, 233 # and it's somewhat cumbersome to close them, so just ignore this. 234 warnings.simplefilter("ignore", ResourceWarning) 235 236 # Run the adb client and have it start the adb server. 237 proc = subprocess.Popen(["adb", "-P", str(port), "start-server"], 238 stdin=subprocess.PIPE, 239 stdout=subprocess.PIPE, 240 stderr=subprocess.PIPE) 241 242 # Start threads that set events when stdout/stderr are closed. 243 stdout_event = threading.Event() 244 stdout_thread = threading.Thread( 245 target=ServerTest._read_pipe_and_set_event, 246 args=(proc.stdout, stdout_event)) 247 stdout_thread.start() 248 249 stderr_event = threading.Event() 250 stderr_thread = threading.Thread( 251 target=ServerTest._read_pipe_and_set_event, 252 args=(proc.stderr, stderr_event)) 253 stderr_thread.start() 254 255 # Wait for the adb client to finish. Once that has occurred, if 256 # stdin/stderr/stdout are still open, it must be open in the adb 257 # server. 258 proc.wait() 259 260 # Try to write to stdin which we expect is closed. If it isn't 261 # closed, we should get an IOError. If we don't get an IOError, 262 # stdin must still be open in the adb server. The adb client is 263 # probably letting the adb server inherit stdin which would be 264 # wrong. 265 with self.assertRaises(IOError): 266 proc.stdin.write(b"x") 267 proc.stdin.flush() 268 269 # Wait a few seconds for stdout/stderr to be closed (in the success 270 # case, this won't wait at all). If there is a timeout, that means 271 # stdout/stderr were not closed and and they must be open in the adb 272 # server, suggesting that the adb client is letting the adb server 273 # inherit stdout/stderr which would be wrong. 274 self.assertTrue(stdout_event.wait(5), "adb stdout not closed") 275 self.assertTrue(stderr_event.wait(5), "adb stderr not closed") 276 stdout_thread.join() 277 stderr_thread.join() 278 finally: 279 # If we started a server, kill it. 280 subprocess.check_output(["adb", "-P", str(port), "kill-server"], 281 stderr=subprocess.STDOUT) 282 283 @unittest.skipUnless( 284 os.name == "posix", 285 "adb doesn't yet support IPv6 on Windows", 286 ) 287 def test_starts_on_ipv6_localhost(self): 288 """ 289 Tests that the server can start up on ::1 and that it's accessible 290 """ 291 292 server_port = find_open_port() 293 try: 294 subprocess.check_output( 295 ["adb", "-L", "tcp:[::1]:{}".format(server_port), "server"], 296 stderr=subprocess.STDOUT, 297 ) 298 with fake_adbd() as (port, _): 299 with adb_connect(self, serial="localhost:{}".format(port)): 300 pass 301 finally: 302 # If we started a server, kill it. 303 subprocess.check_output( 304 ["adb", "-P", str(server_port), "kill-server"], 305 stderr=subprocess.STDOUT, 306 ) 307 308 309 310 311class EmulatorTest(unittest.TestCase): 312 """Tests for the emulator connection.""" 313 314 def _reset_socket_on_close(self, sock): 315 """Use SO_LINGER to cause TCP RST segment to be sent on socket close.""" 316 # The linger structure is two shorts on Windows, but two ints on Unix. 317 linger_format = "hh" if os.name == "nt" else "ii" 318 l_onoff = 1 319 l_linger = 0 320 321 sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 322 struct.pack(linger_format, l_onoff, l_linger)) 323 # Verify that we set the linger structure properly by retrieving it. 324 linger = sock.getsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 16) 325 self.assertEqual((l_onoff, l_linger), 326 struct.unpack_from(linger_format, linger)) 327 328 def test_emu_kill(self): 329 """Ensure that adb emu kill works. 330 331 Bug: https://code.google.com/p/android/issues/detail?id=21021 332 """ 333 with contextlib.closing( 334 socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as listener: 335 # Use SO_REUSEADDR so subsequent runs of the test can grab the port 336 # even if it is in TIME_WAIT. 337 listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 338 listener.bind(("127.0.0.1", 0)) 339 listener.listen(4) 340 port = listener.getsockname()[1] 341 342 # Now that listening has started, start adb emu kill, telling it to 343 # connect to our mock emulator. 344 proc = subprocess.Popen( 345 ["adb", "-s", "emulator-" + str(port), "emu", "kill"], 346 stderr=subprocess.STDOUT) 347 348 accepted_connection, addr = listener.accept() 349 with contextlib.closing(accepted_connection) as conn: 350 # If WSAECONNABORTED (10053) is raised by any socket calls, 351 # then adb probably isn't reading the data that we sent it. 352 conn.sendall(("Android Console: type 'help' for a list " 353 "of commands\r\n").encode("utf8")) 354 conn.sendall(b"OK\r\n") 355 356 with contextlib.closing(conn.makefile()) as connf: 357 line = connf.readline() 358 if line.startswith("auth"): 359 # Ignore the first auth line. 360 line = connf.readline() 361 self.assertEqual("kill\n", line) 362 self.assertEqual("quit\n", connf.readline()) 363 364 conn.sendall(b"OK: killing emulator, bye bye\r\n") 365 366 # Use SO_LINGER to send TCP RST segment to test whether adb 367 # ignores WSAECONNRESET on Windows. This happens with the 368 # real emulator because it just calls exit() without closing 369 # the socket or calling shutdown(SD_SEND). At process 370 # termination, Windows sends a TCP RST segment for every 371 # open socket that shutdown(SD_SEND) wasn't used on. 372 self._reset_socket_on_close(conn) 373 374 # Wait for adb to finish, so we can check return code. 375 proc.communicate() 376 377 # If this fails, adb probably isn't ignoring WSAECONNRESET when 378 # reading the response from the adb emu kill command (on Windows). 379 self.assertEqual(0, proc.returncode) 380 381 def test_emulator_connect(self): 382 """Ensure that the emulator can connect. 383 384 Bug: http://b/78991667 385 """ 386 with adb_server() as server_port: 387 with fake_adbd() as (port, _): 388 serial = "emulator-{}".format(port - 1) 389 # Ensure that the emulator is not there. 390 try: 391 subprocess.check_output(["adb", "-P", str(server_port), 392 "-s", serial, "get-state"], 393 stderr=subprocess.STDOUT) 394 self.fail("Device should not be available") 395 except subprocess.CalledProcessError as err: 396 self.assertEqual( 397 err.output.strip(), 398 "error: device '{}' not found".format(serial).encode("utf8")) 399 400 # Let the ADB server know that the emulator has started. 401 with contextlib.closing( 402 socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: 403 sock.connect(("localhost", server_port)) 404 command = "host:emulator:{}".format(port).encode("utf8") 405 sock.sendall(b"%04x%s" % (len(command), command)) 406 407 # Ensure the emulator is there. 408 subprocess.check_call(["adb", "-P", str(server_port), 409 "-s", serial, "wait-for-device"]) 410 output = subprocess.check_output(["adb", "-P", str(server_port), 411 "-s", serial, "get-state"]) 412 self.assertEqual(output.strip(), b"device") 413 414 415class ConnectionTest(unittest.TestCase): 416 """Tests for adb connect.""" 417 418 def test_connect_ipv4_ipv6(self): 419 """Ensure that `adb connect localhost:1234` will try both IPv4 and IPv6. 420 421 Bug: http://b/30313466 422 """ 423 for protocol in (socket.AF_INET, socket.AF_INET6): 424 try: 425 with fake_adbd(protocol=protocol) as (port, _): 426 serial = "localhost:{}".format(port) 427 with adb_connect(self, serial): 428 pass 429 except socket.error: 430 print("IPv6 not available, skipping") 431 continue 432 433 def test_already_connected(self): 434 """Ensure that an already-connected device stays connected.""" 435 436 with fake_adbd() as (port, _): 437 serial = "localhost:{}".format(port) 438 with adb_connect(self, serial): 439 # b/31250450: this always returns 0 but probably shouldn't. 440 output = subprocess.check_output(["adb", "connect", serial]) 441 self.assertEqual( 442 output.strip(), 443 "already connected to {}".format(serial).encode("utf8")) 444 445 @unittest.skip("Currently failing b/123247844") 446 def test_reconnect(self): 447 """Ensure that a disconnected device reconnects.""" 448 449 with fake_adbd() as (port, _): 450 serial = "localhost:{}".format(port) 451 with adb_connect(self, serial): 452 # Wait a bit to give adb some time to connect. 453 time.sleep(0.25) 454 455 output = subprocess.check_output(["adb", "-s", serial, 456 "get-state"]) 457 self.assertEqual(output.strip(), b"device") 458 459 # This will fail. 460 proc = subprocess.Popen(["adb", "-s", serial, "shell", "true"], 461 stdout=subprocess.PIPE, 462 stderr=subprocess.STDOUT) 463 output, _ = proc.communicate() 464 self.assertEqual(output.strip(), b"error: closed") 465 466 subprocess.check_call(["adb", "-s", serial, "wait-for-device"]) 467 468 output = subprocess.check_output(["adb", "-s", serial, 469 "get-state"]) 470 self.assertEqual(output.strip(), b"device") 471 472 # Once we explicitly kick a device, it won't attempt to 473 # reconnect. 474 output = subprocess.check_output(["adb", "disconnect", serial]) 475 self.assertEqual( 476 output.strip(), 477 "disconnected {}".format(serial).encode("utf8")) 478 try: 479 subprocess.check_output(["adb", "-s", serial, "get-state"], 480 stderr=subprocess.STDOUT) 481 self.fail("Device should not be available") 482 except subprocess.CalledProcessError as err: 483 self.assertEqual( 484 err.output.strip(), 485 "error: device '{}' not found".format(serial).encode("utf8")) 486 487 488class DisconnectionTest(unittest.TestCase): 489 """Tests for adb disconnect.""" 490 491 def test_disconnect(self): 492 """Ensure that `adb disconnect` takes effect immediately.""" 493 494 def _devices(port): 495 output = subprocess.check_output(["adb", "-P", str(port), "devices"]) 496 return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] 497 498 with adb_server() as server_port: 499 with fake_adbd() as (port, sock): 500 device_name = "localhost:{}".format(port) 501 output = subprocess.check_output(["adb", "-P", str(server_port), 502 "connect", device_name]) 503 self.assertEqual(output.strip(), 504 "connected to {}".format(device_name).encode("utf8")) 505 506 507 self.assertEqual(_devices(server_port), [[device_name, "device"]]) 508 509 # Send a deliberately malformed packet to make the device go offline. 510 packet = struct.pack("IIIIII", 0, 0, 0, 0, 0, 0) 511 sock.sendall(packet) 512 513 # Wait a bit. 514 time.sleep(0.1) 515 516 self.assertEqual(_devices(server_port), [[device_name, "offline"]]) 517 518 # Disconnect the device. 519 output = subprocess.check_output(["adb", "-P", str(server_port), 520 "disconnect", device_name]) 521 522 # Wait a bit. 523 time.sleep(0.1) 524 525 self.assertEqual(_devices(server_port), []) 526 527 528@unittest.skipUnless(sys.platform == "win32", "requires Windows") 529class PowerTest(unittest.TestCase): 530 def test_resume_usb_kick(self): 531 """Resuming from sleep/hibernate should kick USB devices.""" 532 try: 533 usb_serial = subprocess.check_output(["adb", "-d", "get-serialno"]).strip() 534 except subprocess.CalledProcessError: 535 # If there are multiple USB devices, we don't have a way to check whether the selected 536 # device is USB. 537 raise unittest.SkipTest('requires single USB device') 538 539 try: 540 serial = subprocess.check_output(["adb", "get-serialno"]).strip() 541 except subprocess.CalledProcessError: 542 # Did you forget to select a device with $ANDROID_SERIAL? 543 raise unittest.SkipTest('requires $ANDROID_SERIAL set to a USB device') 544 545 # Test only works with USB devices because adb _power_notification_thread does not kick 546 # non-USB devices on resume event. 547 if serial != usb_serial: 548 raise unittest.SkipTest('requires USB device') 549 550 # Run an adb shell command in the background that takes a while to complete. 551 proc = subprocess.Popen(['adb', 'shell', 'sleep', '5']) 552 553 # Wait for startup of adb server's _power_notification_thread. 554 time.sleep(0.1) 555 556 # Simulate resuming from sleep/hibernation by sending Windows message. 557 import ctypes 558 from ctypes import wintypes 559 HWND_BROADCAST = 0xffff 560 WM_POWERBROADCAST = 0x218 561 PBT_APMRESUMEAUTOMATIC = 0x12 562 563 PostMessageW = ctypes.windll.user32.PostMessageW 564 PostMessageW.restype = wintypes.BOOL 565 PostMessageW.argtypes = (wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM) 566 result = PostMessageW(HWND_BROADCAST, WM_POWERBROADCAST, PBT_APMRESUMEAUTOMATIC, 0) 567 if not result: 568 raise ctypes.WinError() 569 570 # Wait for connection to adb shell to be broken by _power_notification_thread detecting the 571 # Windows message. 572 start = time.time() 573 proc.wait() 574 end = time.time() 575 576 # If the power event was detected, the adb shell command should be broken very quickly. 577 self.assertLess(end - start, 2) 578 579 580def main(): 581 """Main entrypoint.""" 582 random.seed(0) 583 unittest.main(verbosity=3) 584 585 586if __name__ == "__main__": 587 main() 588