diff options
Diffstat (limited to 'adb/test_adb.py')
| -rwxr-xr-x | adb/test_adb.py | 225 |
1 files changed, 225 insertions, 0 deletions
diff --git a/adb/test_adb.py b/adb/test_adb.py index c872fb0f72..a32d8757d9 100755 --- a/adb/test_adb.py +++ b/adb/test_adb.py @@ -25,6 +25,7 @@ import os import random import select import socket +import string import struct import subprocess import sys @@ -32,6 +33,7 @@ import threading import time import unittest import warnings +from importlib import util def find_open_port(): # Find an open port. @@ -576,6 +578,229 @@ class PowerTest(unittest.TestCase): # If the power event was detected, the adb shell command should be broken very quickly. self.assertLess(end - start, 2) +"""Use 'adb mdns check' to see if mdns discovery is available.""" +def is_adb_mdns_available(): + with adb_server() as server_port: + output = subprocess.check_output(["adb", "-P", str(server_port), + "mdns", "check"]).strip() + return output.startswith(b"mdns daemon version") + +"""Check if we have zeroconf python library installed""" +def is_zeroconf_installed(): + zeroconf_spec = util.find_spec("zeroconf") + return zeroconf_spec is not None + +@contextlib.contextmanager +def zeroconf_context(ipversion): + from zeroconf import Zeroconf + """Context manager for a zeroconf instance + + This creates a zeroconf instance and returns it. + """ + + try: + zeroconf = Zeroconf(ip_version=ipversion) + yield zeroconf + finally: + zeroconf.close() + +@contextlib.contextmanager +def zeroconf_register_service(zeroconf_ctx, info): + """Context manager for a zeroconf service + + Registers a service and unregisters it on cleanup. Returns the ServiceInfo + supplied. + """ + + try: + zeroconf_ctx.register_service(info) + yield info + finally: + zeroconf_ctx.unregister_service(info) + +@contextlib.contextmanager +def zeroconf_register_services(zeroconf_ctx, infos): + """Context manager for multiple zeroconf services + + Registers all services given and unregisters all on cleanup. Returns the ServiceInfo + list supplied. + """ + + try: + for info in infos: + zeroconf_ctx.register_service(info) + yield infos + finally: + for info in infos: + zeroconf_ctx.unregister_service(info) + +"""Should match the service names listed in adb_mdns.h""" +class MdnsTest: + """Tests for adb mdns.""" + @staticmethod + def _mdns_services(port): + output = subprocess.check_output(["adb", "-P", str(port), "mdns", "services"]) + return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] + + @staticmethod + def _devices(port): + output = subprocess.check_output(["adb", "-P", str(port), "devices"]) + return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] + + + class Base(unittest.TestCase): + @contextlib.contextmanager + def _adb_mdns_connect(self, server_port, mdns_instance, serial, should_connect): + """Context manager for an ADB connection. + + This automatically disconnects when done with the connection. + """ + + output = subprocess.check_output(["adb", "-P", str(server_port), "connect", mdns_instance]) + if should_connect: + self.assertEqual(output.strip(), "connected to {}".format(serial).encode("utf8")) + else: + self.assertTrue(output.startswith("failed to resolve host: '{}'" + .format(mdns_instance).encode("utf8"))) + + try: + yield + finally: + # Perform best-effort disconnection. Discard the output. + subprocess.Popen(["adb", "disconnect", serial], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE).communicate() + + + @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") + def test_mdns_services_register_unregister(self): + """Ensure that `adb mdns services` correctly adds and removes a service + """ + from zeroconf import IPVersion, ServiceInfo + + with adb_server() as server_port: + output = subprocess.check_output(["adb", "-P", str(server_port), + "mdns", "services"]).strip() + self.assertTrue(output.startswith(b"List of discovered mdns services")) + + """TODO(joshuaduong): Add ipv6 tests once we have it working in adb""" + """Register/Unregister a service""" + with zeroconf_context(IPVersion.V4Only) as zc: + serv_instance = "my_fake_test_service" + serv_type = "_" + self.service_name + "._tcp." + serv_ipaddr = socket.inet_aton("1.2.3.4") + serv_port = 12345 + service_info = ServiceInfo( + serv_type + "local.", + name=serv_instance + "." + serv_type + "local.", + addresses=[serv_ipaddr], + port=serv_port) + with zeroconf_register_service(zc, service_info) as info: + """Give adb some time to register the service""" + time.sleep(1) + self.assertTrue(any((serv_instance in line and serv_type in line) + for line in MdnsTest._mdns_services(server_port))) + + """Give adb some time to unregister the service""" + time.sleep(1) + self.assertFalse(any((serv_instance in line and serv_type in line) + for line in MdnsTest._mdns_services(server_port))) + + @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") + def test_mdns_services_register_unregister_multiple(self): + """Ensure that `adb mdns services` correctly adds and removes multiple services + """ + from zeroconf import IPVersion, ServiceInfo + + with adb_server() as server_port: + output = subprocess.check_output(["adb", "-P", str(server_port), + "mdns", "services"]).strip() + self.assertTrue(output.startswith(b"List of discovered mdns services")) + + """TODO(joshuaduong): Add ipv6 tests once we have it working in adb""" + """Register/Unregister a service""" + with zeroconf_context(IPVersion.V4Only) as zc: + srvs = { + 'mdns_name': ["testservice0", "testservice1", "testservice2"], + 'mdns_type': "_" + self.service_name + "._tcp.", + 'ipaddr': [ + socket.inet_aton("192.168.0.1"), + socket.inet_aton("10.0.0.255"), + socket.inet_aton("172.16.1.100")], + 'port': [10000, 20000, 65535]} + srv_infos = [] + for i in range(len(srvs['mdns_name'])): + srv_infos.append(ServiceInfo( + srvs['mdns_type'] + "local.", + name=srvs['mdns_name'][i] + "." + srvs['mdns_type'] + "local.", + addresses=[srvs['ipaddr'][i]], + port=srvs['port'][i])) + + """ Register all devices, then unregister""" + with zeroconf_register_services(zc, srv_infos) as infos: + """Give adb some time to register the service""" + time.sleep(1) + for i in range(len(srvs['mdns_name'])): + self.assertTrue(any((srvs['mdns_name'][i] in line and srvs['mdns_type'] in line) + for line in MdnsTest._mdns_services(server_port))) + + """Give adb some time to unregister the service""" + time.sleep(1) + for i in range(len(srvs['mdns_name'])): + self.assertFalse(any((srvs['mdns_name'][i] in line and srvs['mdns_type'] in line) + for line in MdnsTest._mdns_services(server_port))) + + @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") + def test_mdns_connect(self): + """Ensure that `adb connect` by mdns instance name works (for non-pairing services) + """ + from zeroconf import IPVersion, ServiceInfo + + with adb_server() as server_port: + with zeroconf_context(IPVersion.V4Only) as zc: + serv_instance = "fakeadbd-" + ''.join( + random.choice(string.ascii_letters) for i in range(4)) + serv_type = "_" + self.service_name + "._tcp." + serv_ipaddr = socket.inet_aton("127.0.0.1") + should_connect = self.service_name != "adb-tls-pairing" + with fake_adbd() as (port, _): + service_info = ServiceInfo( + serv_type + "local.", + name=serv_instance + "." + serv_type + "local.", + addresses=[serv_ipaddr], + port=port) + with zeroconf_register_service(zc, service_info) as info: + """Give adb some time to register the service""" + time.sleep(1) + self.assertTrue(any((serv_instance in line and serv_type in line) + for line in MdnsTest._mdns_services(server_port))) + full_name = '.'.join([serv_instance, serv_type]) + with self._adb_mdns_connect(server_port, serv_instance, full_name, + should_connect): + if should_connect: + self.assertEqual(MdnsTest._devices(server_port), + [[full_name, "device"]]) + + """Give adb some time to unregister the service""" + time.sleep(1) + self.assertFalse(any((serv_instance in line and serv_type in line) + for line in MdnsTest._mdns_services(server_port))) + + +@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") +class MdnsTestAdb(MdnsTest.Base): + service_name = "adb" + + +@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") +class MdnsTestAdbTlsConnect(MdnsTest.Base): + service_name = "adb-tls-connect" + + +@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") +class MdnsTestAdbTlsPairing(MdnsTest.Base): + service_name = "adb-tls-pairing" + def main(): """Main entrypoint.""" |