Merge "logd: move leading_dropped logic into FlushTo()"
diff --git a/adb/test_adb.py b/adb/test_adb.py
index 4b99411..b9f0d54 100755
--- a/adb/test_adb.py
+++ b/adb/test_adb.py
@@ -34,7 +34,6 @@
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 @@
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:]]
+ 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:]]
- @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:]]
+ @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.
+ @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.
- """
+ 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")))
+ 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()
+ 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
+ @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"))
+ 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_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, _):
+ """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)