adb: don't use parameterized in test_adb.py.

The test bots don't have this module installed.

Test: treehugger
Change-Id: I849a075b784eac296608e45bec7b2fce38f56ac4

1 files 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)