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