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 string 29import struct 30import subprocess 31import sys 32import threading 33import time 34import unittest 35import warnings 36from importlib import util 37 38def find_open_port(): 39 # Find an open port. 40 with socket.socket() as s: 41 s.bind(("localhost", 0)) 42 return s.getsockname()[1] 43 44@contextlib.contextmanager 45def fake_adbd(protocol=socket.AF_INET, port=0): 46 """Creates a fake ADB daemon that just replies with a CNXN packet.""" 47 48 serversock = socket.socket(protocol, socket.SOCK_STREAM) 49 serversock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 50 if protocol == socket.AF_INET: 51 serversock.bind(("127.0.0.1", port)) 52 else: 53 serversock.bind(("::1", port)) 54 serversock.listen(1) 55 56 # A pipe that is used to signal the thread that it should terminate. 57 readsock, writesock = socket.socketpair() 58 59 def _adb_packet(command: bytes, arg0: int, arg1: int, data: bytes) -> bytes: 60 bin_command = struct.unpack("I", command)[0] 61 buf = struct.pack("IIIIII", bin_command, arg0, arg1, len(data), 0, 62 bin_command ^ 0xffffffff) 63 buf += data 64 return buf 65 66 def _handle(sock): 67 with contextlib.closing(sock) as serversock: 68 rlist = [readsock, serversock] 69 cnxn_sent = {} 70 while True: 71 read_ready, _, _ = select.select(rlist, [], []) 72 for ready in read_ready: 73 if ready == readsock: 74 # Closure pipe 75 for f in rlist: 76 f.close() 77 return 78 elif ready == serversock: 79 # Server socket 80 conn, _ = ready.accept() 81 rlist.append(conn) 82 else: 83 # Client socket 84 data = ready.recv(1024) 85 if not data or data.startswith(b"OPEN"): 86 if ready in cnxn_sent: 87 del cnxn_sent[ready] 88 ready.shutdown(socket.SHUT_RDWR) 89 ready.close() 90 rlist.remove(ready) 91 continue 92 if ready in cnxn_sent: 93 continue 94 cnxn_sent[ready] = True 95 ready.sendall(_adb_packet(b"CNXN", 0x01000001, 1024 * 1024, 96 b"device::ro.product.name=fakeadb")) 97 98 port = serversock.getsockname()[1] 99 server_thread = threading.Thread(target=_handle, args=(serversock,)) 100 server_thread.start() 101 102 try: 103 yield port, writesock 104 finally: 105 writesock.close() 106 server_thread.join() 107 108 109@contextlib.contextmanager 110def adb_connect(unittest, serial): 111 """Context manager for an ADB connection. 112 113 This automatically disconnects when done with the connection. 114 """ 115 116 output = subprocess.check_output(["adb", "connect", serial]) 117 unittest.assertEqual(output.strip(), 118 "connected to {}".format(serial).encode("utf8")) 119 120 try: 121 yield 122 finally: 123 # Perform best-effort disconnection. Discard the output. 124 subprocess.Popen(["adb", "disconnect", serial], 125 stdout=subprocess.PIPE, 126 stderr=subprocess.PIPE).communicate() 127 128 129@contextlib.contextmanager 130def adb_server(): 131 """Context manager for an ADB server. 132 133 This creates an ADB server and returns the port it's listening on. 134 """ 135 136 port = find_open_port() 137 read_pipe, write_pipe = os.pipe() 138 139 if sys.platform == "win32": 140 import msvcrt 141 write_handle = msvcrt.get_osfhandle(write_pipe) 142 os.set_handle_inheritable(write_handle, True) 143 reply_fd = str(write_handle) 144 else: 145 os.set_inheritable(write_pipe, True) 146 reply_fd = str(write_pipe) 147 148 proc = subprocess.Popen(["adb", "-L", "tcp:localhost:{}".format(port), 149 "fork-server", "server", 150 "--reply-fd", reply_fd], close_fds=False) 151 try: 152 os.close(write_pipe) 153 greeting = os.read(read_pipe, 1024) 154 assert greeting == b"OK\n", repr(greeting) 155 yield port 156 finally: 157 proc.terminate() 158 proc.wait() 159 160 161class CommandlineTest(unittest.TestCase): 162 """Tests for the ADB commandline.""" 163 164 def test_help(self): 165 """Make sure we get _something_ out of help.""" 166 out = subprocess.check_output( 167 ["adb", "help"], stderr=subprocess.STDOUT) 168 self.assertGreater(len(out), 0) 169 170 def test_version(self): 171 """Get a version number out of the output of adb.""" 172 lines = subprocess.check_output(["adb", "version"]).splitlines() 173 version_line = lines[0] 174 self.assertRegex( 175 version_line, rb"^Android Debug Bridge version \d+\.\d+\.\d+$") 176 if len(lines) == 2: 177 # Newer versions of ADB have a second line of output for the 178 # version that includes a specific revision (git SHA). 179 revision_line = lines[1] 180 self.assertRegex( 181 revision_line, rb"^Revision [0-9a-f]{12}-android$") 182 183 def test_tcpip_error_messages(self): 184 """Make sure 'adb tcpip' parsing is sane.""" 185 proc = subprocess.Popen(["adb", "tcpip"], 186 stdout=subprocess.PIPE, 187 stderr=subprocess.STDOUT) 188 out, _ = proc.communicate() 189 self.assertEqual(1, proc.returncode) 190 self.assertIn(b"requires an argument", out) 191 192 proc = subprocess.Popen(["adb", "tcpip", "foo"], 193 stdout=subprocess.PIPE, 194 stderr=subprocess.STDOUT) 195 out, _ = proc.communicate() 196 self.assertEqual(1, proc.returncode) 197 self.assertIn(b"invalid port", out) 198 199 200class ServerTest(unittest.TestCase): 201 """Tests for the ADB server.""" 202 203 @staticmethod 204 def _read_pipe_and_set_event(pipe, event): 205 """Reads a pipe until it is closed, then sets the event.""" 206 pipe.read() 207 event.set() 208 209 def test_handle_inheritance(self): 210 """Test that launch_server() does not inherit handles. 211 212 launch_server() should not let the adb server inherit 213 stdin/stdout/stderr handles, which can cause callers of adb.exe to hang. 214 This test also runs fine on unix even though the impetus is an issue 215 unique to Windows. 216 """ 217 # This test takes 5 seconds to run on Windows: if there is no adb server 218 # running on the the port used below, adb kill-server tries to make a 219 # TCP connection to a closed port and that takes 1 second on Windows; 220 # adb start-server does the same TCP connection which takes another 221 # second, and it waits 3 seconds after starting the server. 222 223 # Start adb client with redirected stdin/stdout/stderr to check if it 224 # passes those redirections to the adb server that it starts. To do 225 # this, run an instance of the adb server on a non-default port so we 226 # don't conflict with a pre-existing adb server that may already be 227 # setup with adb TCP/emulator connections. If there is a pre-existing 228 # adb server, this also tests whether multiple instances of the adb 229 # server conflict on adb.log. 230 231 port = find_open_port() 232 233 try: 234 # We get warnings for unclosed files for the subprocess's pipes, 235 # and it's somewhat cumbersome to close them, so just ignore this. 236 warnings.simplefilter("ignore", ResourceWarning) 237 238 # Run the adb client and have it start the adb server. 239 proc = subprocess.Popen(["adb", "-P", str(port), "start-server"], 240 stdin=subprocess.PIPE, 241 stdout=subprocess.PIPE, 242 stderr=subprocess.PIPE) 243 244 # Start threads that set events when stdout/stderr are closed. 245 stdout_event = threading.Event() 246 stdout_thread = threading.Thread( 247 target=ServerTest._read_pipe_and_set_event, 248 args=(proc.stdout, stdout_event)) 249 stdout_thread.start() 250 251 stderr_event = threading.Event() 252 stderr_thread = threading.Thread( 253 target=ServerTest._read_pipe_and_set_event, 254 args=(proc.stderr, stderr_event)) 255 stderr_thread.start() 256 257 # Wait for the adb client to finish. Once that has occurred, if 258 # stdin/stderr/stdout are still open, it must be open in the adb 259 # server. 260 proc.wait() 261 262 # Try to write to stdin which we expect is closed. If it isn't 263 # closed, we should get an IOError. If we don't get an IOError, 264 # stdin must still be open in the adb server. The adb client is 265 # probably letting the adb server inherit stdin which would be 266 # wrong. 267 with self.assertRaises(IOError): 268 proc.stdin.write(b"x") 269 proc.stdin.flush() 270 271 # Wait a few seconds for stdout/stderr to be closed (in the success 272 # case, this won't wait at all). If there is a timeout, that means 273 # stdout/stderr were not closed and and they must be open in the adb 274 # server, suggesting that the adb client is letting the adb server 275 # inherit stdout/stderr which would be wrong. 276 self.assertTrue(stdout_event.wait(5), "adb stdout not closed") 277 self.assertTrue(stderr_event.wait(5), "adb stderr not closed") 278 stdout_thread.join() 279 stderr_thread.join() 280 finally: 281 # If we started a server, kill it. 282 subprocess.check_output(["adb", "-P", str(port), "kill-server"], 283 stderr=subprocess.STDOUT) 284 285 @unittest.skipUnless( 286 os.name == "posix", 287 "adb doesn't yet support IPv6 on Windows", 288 ) 289 def test_starts_on_ipv6_localhost(self): 290 """ 291 Tests that the server can start up on ::1 and that it's accessible 292 """ 293 294 server_port = find_open_port() 295 try: 296 subprocess.check_output( 297 ["adb", "-L", "tcp:[::1]:{}".format(server_port), "server"], 298 stderr=subprocess.STDOUT, 299 ) 300 with fake_adbd() as (port, _): 301 with adb_connect(self, serial="localhost:{}".format(port)): 302 pass 303 finally: 304 # If we started a server, kill it. 305 subprocess.check_output( 306 ["adb", "-P", str(server_port), "kill-server"], 307 stderr=subprocess.STDOUT, 308 ) 309 310 311 312 313class EmulatorTest(unittest.TestCase): 314 """Tests for the emulator connection.""" 315 316 def _reset_socket_on_close(self, sock): 317 """Use SO_LINGER to cause TCP RST segment to be sent on socket close.""" 318 # The linger structure is two shorts on Windows, but two ints on Unix. 319 linger_format = "hh" if os.name == "nt" else "ii" 320 l_onoff = 1 321 l_linger = 0 322 323 sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 324 struct.pack(linger_format, l_onoff, l_linger)) 325 # Verify that we set the linger structure properly by retrieving it. 326 linger = sock.getsockopt(socket.SOL_SOCKET, socket.SO_LINGER, 16) 327 self.assertEqual((l_onoff, l_linger), 328 struct.unpack_from(linger_format, linger)) 329 330 def test_emu_kill(self): 331 """Ensure that adb emu kill works. 332 333 Bug: https://code.google.com/p/android/issues/detail?id=21021 334 """ 335 with contextlib.closing( 336 socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as listener: 337 # Use SO_REUSEADDR so subsequent runs of the test can grab the port 338 # even if it is in TIME_WAIT. 339 listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) 340 listener.bind(("127.0.0.1", 0)) 341 listener.listen(4) 342 port = listener.getsockname()[1] 343 344 # Now that listening has started, start adb emu kill, telling it to 345 # connect to our mock emulator. 346 proc = subprocess.Popen( 347 ["adb", "-s", "emulator-" + str(port), "emu", "kill"], 348 stderr=subprocess.STDOUT) 349 350 accepted_connection, addr = listener.accept() 351 with contextlib.closing(accepted_connection) as conn: 352 # If WSAECONNABORTED (10053) is raised by any socket calls, 353 # then adb probably isn't reading the data that we sent it. 354 conn.sendall(("Android Console: type 'help' for a list " 355 "of commands\r\n").encode("utf8")) 356 conn.sendall(b"OK\r\n") 357 358 with contextlib.closing(conn.makefile()) as connf: 359 line = connf.readline() 360 if line.startswith("auth"): 361 # Ignore the first auth line. 362 line = connf.readline() 363 self.assertEqual("kill\n", line) 364 self.assertEqual("quit\n", connf.readline()) 365 366 conn.sendall(b"OK: killing emulator, bye bye\r\n") 367 368 # Use SO_LINGER to send TCP RST segment to test whether adb 369 # ignores WSAECONNRESET on Windows. This happens with the 370 # real emulator because it just calls exit() without closing 371 # the socket or calling shutdown(SD_SEND). At process 372 # termination, Windows sends a TCP RST segment for every 373 # open socket that shutdown(SD_SEND) wasn't used on. 374 self._reset_socket_on_close(conn) 375 376 # Wait for adb to finish, so we can check return code. 377 proc.communicate() 378 379 # If this fails, adb probably isn't ignoring WSAECONNRESET when 380 # reading the response from the adb emu kill command (on Windows). 381 self.assertEqual(0, proc.returncode) 382 383 def test_emulator_connect(self): 384 """Ensure that the emulator can connect. 385 386 Bug: http://b/78991667 387 """ 388 with adb_server() as server_port: 389 with fake_adbd() as (port, _): 390 serial = "emulator-{}".format(port - 1) 391 # Ensure that the emulator is not there. 392 try: 393 subprocess.check_output(["adb", "-P", str(server_port), 394 "-s", serial, "get-state"], 395 stderr=subprocess.STDOUT) 396 self.fail("Device should not be available") 397 except subprocess.CalledProcessError as err: 398 self.assertEqual( 399 err.output.strip(), 400 "error: device '{}' not found".format(serial).encode("utf8")) 401 402 # Let the ADB server know that the emulator has started. 403 with contextlib.closing( 404 socket.socket(socket.AF_INET, socket.SOCK_STREAM)) as sock: 405 sock.connect(("localhost", server_port)) 406 command = "host:emulator:{}".format(port).encode("utf8") 407 sock.sendall(b"%04x%s" % (len(command), command)) 408 409 # Ensure the emulator is there. 410 subprocess.check_call(["adb", "-P", str(server_port), 411 "-s", serial, "wait-for-device"]) 412 output = subprocess.check_output(["adb", "-P", str(server_port), 413 "-s", serial, "get-state"]) 414 self.assertEqual(output.strip(), b"device") 415 416 417class ConnectionTest(unittest.TestCase): 418 """Tests for adb connect.""" 419 420 def test_connect_ipv4_ipv6(self): 421 """Ensure that `adb connect localhost:1234` will try both IPv4 and IPv6. 422 423 Bug: http://b/30313466 424 """ 425 for protocol in (socket.AF_INET, socket.AF_INET6): 426 try: 427 with fake_adbd(protocol=protocol) as (port, _): 428 serial = "localhost:{}".format(port) 429 with adb_connect(self, serial): 430 pass 431 except socket.error: 432 print("IPv6 not available, skipping") 433 continue 434 435 def test_already_connected(self): 436 """Ensure that an already-connected device stays connected.""" 437 438 with fake_adbd() as (port, _): 439 serial = "localhost:{}".format(port) 440 with adb_connect(self, serial): 441 # b/31250450: this always returns 0 but probably shouldn't. 442 output = subprocess.check_output(["adb", "connect", serial]) 443 self.assertEqual( 444 output.strip(), 445 "already connected to {}".format(serial).encode("utf8")) 446 447 @unittest.skip("Currently failing b/123247844") 448 def test_reconnect(self): 449 """Ensure that a disconnected device reconnects.""" 450 451 with fake_adbd() as (port, _): 452 serial = "localhost:{}".format(port) 453 with adb_connect(self, serial): 454 # Wait a bit to give adb some time to connect. 455 time.sleep(0.25) 456 457 output = subprocess.check_output(["adb", "-s", serial, 458 "get-state"]) 459 self.assertEqual(output.strip(), b"device") 460 461 # This will fail. 462 proc = subprocess.Popen(["adb", "-s", serial, "shell", "true"], 463 stdout=subprocess.PIPE, 464 stderr=subprocess.STDOUT) 465 output, _ = proc.communicate() 466 self.assertEqual(output.strip(), b"error: closed") 467 468 subprocess.check_call(["adb", "-s", serial, "wait-for-device"]) 469 470 output = subprocess.check_output(["adb", "-s", serial, 471 "get-state"]) 472 self.assertEqual(output.strip(), b"device") 473 474 # Once we explicitly kick a device, it won't attempt to 475 # reconnect. 476 output = subprocess.check_output(["adb", "disconnect", serial]) 477 self.assertEqual( 478 output.strip(), 479 "disconnected {}".format(serial).encode("utf8")) 480 try: 481 subprocess.check_output(["adb", "-s", serial, "get-state"], 482 stderr=subprocess.STDOUT) 483 self.fail("Device should not be available") 484 except subprocess.CalledProcessError as err: 485 self.assertEqual( 486 err.output.strip(), 487 "error: device '{}' not found".format(serial).encode("utf8")) 488 489 490class DisconnectionTest(unittest.TestCase): 491 """Tests for adb disconnect.""" 492 493 def test_disconnect(self): 494 """Ensure that `adb disconnect` takes effect immediately.""" 495 496 def _devices(port): 497 output = subprocess.check_output(["adb", "-P", str(port), "devices"]) 498 return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] 499 500 with adb_server() as server_port: 501 with fake_adbd() as (port, sock): 502 device_name = "localhost:{}".format(port) 503 output = subprocess.check_output(["adb", "-P", str(server_port), 504 "connect", device_name]) 505 self.assertEqual(output.strip(), 506 "connected to {}".format(device_name).encode("utf8")) 507 508 509 self.assertEqual(_devices(server_port), [[device_name, "device"]]) 510 511 # Send a deliberately malformed packet to make the device go offline. 512 packet = struct.pack("IIIIII", 0, 0, 0, 0, 0, 0) 513 sock.sendall(packet) 514 515 # Wait a bit. 516 time.sleep(0.1) 517 518 self.assertEqual(_devices(server_port), [[device_name, "offline"]]) 519 520 # Disconnect the device. 521 output = subprocess.check_output(["adb", "-P", str(server_port), 522 "disconnect", device_name]) 523 524 # Wait a bit. 525 time.sleep(0.1) 526 527 self.assertEqual(_devices(server_port), []) 528 529 530@unittest.skipUnless(sys.platform == "win32", "requires Windows") 531class PowerTest(unittest.TestCase): 532 def test_resume_usb_kick(self): 533 """Resuming from sleep/hibernate should kick USB devices.""" 534 try: 535 usb_serial = subprocess.check_output(["adb", "-d", "get-serialno"]).strip() 536 except subprocess.CalledProcessError: 537 # If there are multiple USB devices, we don't have a way to check whether the selected 538 # device is USB. 539 raise unittest.SkipTest('requires single USB device') 540 541 try: 542 serial = subprocess.check_output(["adb", "get-serialno"]).strip() 543 except subprocess.CalledProcessError: 544 # Did you forget to select a device with $ANDROID_SERIAL? 545 raise unittest.SkipTest('requires $ANDROID_SERIAL set to a USB device') 546 547 # Test only works with USB devices because adb _power_notification_thread does not kick 548 # non-USB devices on resume event. 549 if serial != usb_serial: 550 raise unittest.SkipTest('requires USB device') 551 552 # Run an adb shell command in the background that takes a while to complete. 553 proc = subprocess.Popen(['adb', 'shell', 'sleep', '5']) 554 555 # Wait for startup of adb server's _power_notification_thread. 556 time.sleep(0.1) 557 558 # Simulate resuming from sleep/hibernation by sending Windows message. 559 import ctypes 560 from ctypes import wintypes 561 HWND_BROADCAST = 0xffff 562 WM_POWERBROADCAST = 0x218 563 PBT_APMRESUMEAUTOMATIC = 0x12 564 565 PostMessageW = ctypes.windll.user32.PostMessageW 566 PostMessageW.restype = wintypes.BOOL 567 PostMessageW.argtypes = (wintypes.HWND, wintypes.UINT, wintypes.WPARAM, wintypes.LPARAM) 568 result = PostMessageW(HWND_BROADCAST, WM_POWERBROADCAST, PBT_APMRESUMEAUTOMATIC, 0) 569 if not result: 570 raise ctypes.WinError() 571 572 # Wait for connection to adb shell to be broken by _power_notification_thread detecting the 573 # Windows message. 574 start = time.time() 575 proc.wait() 576 end = time.time() 577 578 # If the power event was detected, the adb shell command should be broken very quickly. 579 self.assertLess(end - start, 2) 580 581"""Use 'adb mdns check' to see if mdns discovery is available.""" 582def is_adb_mdns_available(): 583 with adb_server() as server_port: 584 output = subprocess.check_output(["adb", "-P", str(server_port), 585 "mdns", "check"]).strip() 586 return output.startswith(b"mdns daemon version") 587 588"""Check if we have zeroconf python library installed""" 589def is_zeroconf_installed(): 590 zeroconf_spec = util.find_spec("zeroconf") 591 return zeroconf_spec is not None 592 593@contextlib.contextmanager 594def zeroconf_context(ipversion): 595 from zeroconf import Zeroconf 596 """Context manager for a zeroconf instance 597 598 This creates a zeroconf instance and returns it. 599 """ 600 601 try: 602 zeroconf = Zeroconf(ip_version=ipversion) 603 yield zeroconf 604 finally: 605 zeroconf.close() 606 607@contextlib.contextmanager 608def zeroconf_register_service(zeroconf_ctx, info): 609 """Context manager for a zeroconf service 610 611 Registers a service and unregisters it on cleanup. Returns the ServiceInfo 612 supplied. 613 """ 614 615 try: 616 zeroconf_ctx.register_service(info) 617 yield info 618 finally: 619 zeroconf_ctx.unregister_service(info) 620 621@contextlib.contextmanager 622def zeroconf_register_services(zeroconf_ctx, infos): 623 """Context manager for multiple zeroconf services 624 625 Registers all services given and unregisters all on cleanup. Returns the ServiceInfo 626 list supplied. 627 """ 628 629 try: 630 for info in infos: 631 zeroconf_ctx.register_service(info) 632 yield infos 633 finally: 634 for info in infos: 635 zeroconf_ctx.unregister_service(info) 636 637"""Should match the service names listed in adb_mdns.h""" 638class MdnsTest: 639 """Tests for adb mdns.""" 640 @staticmethod 641 def _mdns_services(port): 642 output = subprocess.check_output(["adb", "-P", str(port), "mdns", "services"]) 643 return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] 644 645 @staticmethod 646 def _devices(port): 647 output = subprocess.check_output(["adb", "-P", str(port), "devices"]) 648 return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] 649 650 651 class Base(unittest.TestCase): 652 @contextlib.contextmanager 653 def _adb_mdns_connect(self, server_port, mdns_instance, serial, should_connect): 654 """Context manager for an ADB connection. 655 656 This automatically disconnects when done with the connection. 657 """ 658 659 output = subprocess.check_output(["adb", "-P", str(server_port), "connect", mdns_instance]) 660 if should_connect: 661 self.assertEqual(output.strip(), "connected to {}".format(serial).encode("utf8")) 662 else: 663 self.assertTrue(output.startswith("failed to resolve host: '{}'" 664 .format(mdns_instance).encode("utf8"))) 665 666 try: 667 yield 668 finally: 669 # Perform best-effort disconnection. Discard the output. 670 subprocess.Popen(["adb", "disconnect", serial], 671 stdout=subprocess.PIPE, 672 stderr=subprocess.PIPE).communicate() 673 674 675 @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") 676 def test_mdns_services_register_unregister(self): 677 """Ensure that `adb mdns services` correctly adds and removes a service 678 """ 679 from zeroconf import IPVersion, ServiceInfo 680 681 with adb_server() as server_port: 682 output = subprocess.check_output(["adb", "-P", str(server_port), 683 "mdns", "services"]).strip() 684 self.assertTrue(output.startswith(b"List of discovered mdns services")) 685 686 """TODO(joshuaduong): Add ipv6 tests once we have it working in adb""" 687 """Register/Unregister a service""" 688 with zeroconf_context(IPVersion.V4Only) as zc: 689 serv_instance = "my_fake_test_service" 690 serv_type = "_" + self.service_name + "._tcp." 691 serv_ipaddr = socket.inet_aton("1.2.3.4") 692 serv_port = 12345 693 service_info = ServiceInfo( 694 serv_type + "local.", 695 name=serv_instance + "." + serv_type + "local.", 696 addresses=[serv_ipaddr], 697 port=serv_port) 698 with zeroconf_register_service(zc, service_info) as info: 699 """Give adb some time to register the service""" 700 time.sleep(1) 701 self.assertTrue(any((serv_instance in line and serv_type in line) 702 for line in MdnsTest._mdns_services(server_port))) 703 704 """Give adb some time to unregister the service""" 705 time.sleep(1) 706 self.assertFalse(any((serv_instance in line and serv_type in line) 707 for line in MdnsTest._mdns_services(server_port))) 708 709 @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") 710 def test_mdns_services_register_unregister_multiple(self): 711 """Ensure that `adb mdns services` correctly adds and removes multiple services 712 """ 713 from zeroconf import IPVersion, ServiceInfo 714 715 with adb_server() as server_port: 716 output = subprocess.check_output(["adb", "-P", str(server_port), 717 "mdns", "services"]).strip() 718 self.assertTrue(output.startswith(b"List of discovered mdns services")) 719 720 """TODO(joshuaduong): Add ipv6 tests once we have it working in adb""" 721 """Register/Unregister a service""" 722 with zeroconf_context(IPVersion.V4Only) as zc: 723 srvs = { 724 'mdns_name': ["testservice0", "testservice1", "testservice2"], 725 'mdns_type': "_" + self.service_name + "._tcp.", 726 'ipaddr': [ 727 socket.inet_aton("192.168.0.1"), 728 socket.inet_aton("10.0.0.255"), 729 socket.inet_aton("172.16.1.100")], 730 'port': [10000, 20000, 65535]} 731 srv_infos = [] 732 for i in range(len(srvs['mdns_name'])): 733 srv_infos.append(ServiceInfo( 734 srvs['mdns_type'] + "local.", 735 name=srvs['mdns_name'][i] + "." + srvs['mdns_type'] + "local.", 736 addresses=[srvs['ipaddr'][i]], 737 port=srvs['port'][i])) 738 739 """ Register all devices, then unregister""" 740 with zeroconf_register_services(zc, srv_infos) as infos: 741 """Give adb some time to register the service""" 742 time.sleep(1) 743 for i in range(len(srvs['mdns_name'])): 744 self.assertTrue(any((srvs['mdns_name'][i] in line and srvs['mdns_type'] in line) 745 for line in MdnsTest._mdns_services(server_port))) 746 747 """Give adb some time to unregister the service""" 748 time.sleep(1) 749 for i in range(len(srvs['mdns_name'])): 750 self.assertFalse(any((srvs['mdns_name'][i] in line and srvs['mdns_type'] in line) 751 for line in MdnsTest._mdns_services(server_port))) 752 753 @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") 754 def test_mdns_connect(self): 755 """Ensure that `adb connect` by mdns instance name works (for non-pairing services) 756 """ 757 from zeroconf import IPVersion, ServiceInfo 758 759 with adb_server() as server_port: 760 with zeroconf_context(IPVersion.V4Only) as zc: 761 serv_instance = "fakeadbd-" + ''.join( 762 random.choice(string.ascii_letters) for i in range(4)) 763 serv_type = "_" + self.service_name + "._tcp." 764 serv_ipaddr = socket.inet_aton("127.0.0.1") 765 should_connect = self.service_name != "adb-tls-pairing" 766 with fake_adbd() as (port, _): 767 service_info = ServiceInfo( 768 serv_type + "local.", 769 name=serv_instance + "." + serv_type + "local.", 770 addresses=[serv_ipaddr], 771 port=port) 772 with zeroconf_register_service(zc, service_info) as info: 773 """Give adb some time to register the service""" 774 time.sleep(1) 775 self.assertTrue(any((serv_instance in line and serv_type in line) 776 for line in MdnsTest._mdns_services(server_port))) 777 full_name = '.'.join([serv_instance, serv_type]) 778 with self._adb_mdns_connect(server_port, serv_instance, full_name, 779 should_connect): 780 if should_connect: 781 self.assertEqual(MdnsTest._devices(server_port), 782 [[full_name, "device"]]) 783 784 """Give adb some time to unregister the service""" 785 time.sleep(1) 786 self.assertFalse(any((serv_instance in line and serv_type in line) 787 for line in MdnsTest._mdns_services(server_port))) 788 789 790@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") 791class MdnsTestAdb(MdnsTest.Base): 792 service_name = "adb" 793 794 795@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") 796class MdnsTestAdbTlsConnect(MdnsTest.Base): 797 service_name = "adb-tls-connect" 798 799 800@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") 801class MdnsTestAdbTlsPairing(MdnsTest.Base): 802 service_name = "adb-tls-pairing" 803 804 805def main(): 806 """Main entrypoint.""" 807 random.seed(0) 808 unittest.main(verbosity=3) 809 810 811if __name__ == "__main__": 812 main() 813