From 5ece8e43adb4cb4398159c1b9fdb77bf59ed834c Mon Sep 17 00:00:00 2001 From: Josh Gao Date: Tue, 2 Jun 2020 13:14:45 -0700 Subject: adb: don't use parameterized in test_adb.py. The test bots don't have this module installed. Test: treehugger Change-Id: I849a075b784eac296608e45bec7b2fce38f56ac4 --- test_adb.py | 186 ++++++++++++++++++++++++++++++++---------------------------- 1 file changed, 98 insertions(+), 88 deletions(-) diff --git a/test_adb.py b/test_adb.py index 4b994114..b9f0d548 100755 --- a/test_adb.py +++ b/test_adb.py @@ -34,7 +34,6 @@ import time import unittest import warnings from importlib import util -from parameterized import parameterized_class def find_open_port(): # Find an open port. @@ -620,118 +619,129 @@ def zeroconf_register_service(zeroconf_ctx, info): zeroconf_ctx.unregister_service(info) """Should match the service names listed in adb_mdns.h""" -@parameterized_class(('service_name',), [ - ("adb",), - ("adb-tls-connect",), - ("adb-tls-pairing",), -]) -@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available") -class MdnsTest(unittest.TestCase): +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:]] - - @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. - """ + class Base(unittest.TestCase): + @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:]] - 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"))) + @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:]] - try: - yield - finally: - # Perform best-effort disconnection. Discard the output. - subprocess.Popen(["adb", "disconnect", serial], - stdout=subprocess.PIPE, - stderr=subprocess.PIPE).communicate() + @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. + """ - @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 + 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"))) - 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))) + try: + yield + finally: + # Perform best-effort disconnection. Discard the output. + subprocess.Popen(["adb", "disconnect", serial], + stdout=subprocess.PIPE, + stderr=subprocess.PIPE).communicate() - """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_connect(self): - """Ensure that `adb connect` by mdns instance name works (for non-pairing services) - """ - from zeroconf import IPVersion, ServiceInfo + @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: - 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, _): + 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=port) + 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))) - 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_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.""" random.seed(0) -- cgit v1.2.3