ACPR/android_system_core
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Merge "adb: don't use parameterized in test_adb.py." am: 648266ab1c

Browse source 
Original change: https://android-review.googlesource.com/c/platform/system/core/+/1322179

Change-Id: Icca5eb952bc158013efdb6851d8d6f7a1c794cba
This commit is contained in:
Treehugger Robot 2020-06-02 21:49:15 +00:00 • committed by Automerger Merge Worker
commit 1d73552164
1 changed files with 97 additions and 87 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

184
adb/test_adb.py
View file

@ -34,7 +34,6 @@ import time
import unittest import unittest
import warnings import warnings
from importlib import util from importlib import util
from parameterized import parameterized_class
def find_open_port(): def find_open_port():
# Find an open port. # Find an open port.
@ -620,118 +619,129 @@ def zeroconf_register_service(zeroconf_ctx, info):
zeroconf_ctx.unregister_service(info) zeroconf_ctx.unregister_service(info)
"""Should match the service names listed in adb_mdns.h""" """Should match the service names listed in adb_mdns.h"""
@parameterized_class(('service_name',), [ class MdnsTest:
("adb",),
("adb-tls-connect",),
("adb-tls-pairing",),
])
@unittest.skipIf(not is_adb_mdns_available(), "mdns feature not available")
class MdnsTest(unittest.TestCase):
"""Tests for adb mdns.""" """Tests for adb mdns."""
@staticmethod class Base(unittest.TestCase):
def _mdns_services(port): @staticmethod
output = subprocess.check_output(["adb", "-P", str(port), "mdns", "services"]) def _mdns_services(port):
return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] output = subprocess.check_output(["adb", "-P", str(port), "mdns", "services"])
return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]]
@staticmethod @staticmethod
def _devices(port): def _devices(port):
output = subprocess.check_output(["adb", "-P", str(port), "devices"]) output = subprocess.check_output(["adb", "-P", str(port), "devices"])
return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]] return [x.split("\t") for x in output.decode("utf8").strip().splitlines()[1:]]
@contextlib.contextmanager @contextlib.contextmanager
def _adb_mdns_connect(self, server_port, mdns_instance, serial, should_connect): def _adb_mdns_connect(self, server_port, mdns_instance, serial, should_connect):
"""Context manager for an ADB connection. """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]) output = subprocess.check_output(["adb", "-P", str(server_port), "connect", mdns_instance])
if should_connect: if should_connect:
self.assertEqual(output.strip(), "connected to {}".format(serial).encode("utf8")) self.assertEqual(output.strip(), "connected to {}".format(serial).encode("utf8"))
else: else:
self.assertTrue(output.startswith("failed to resolve host: '{}'" self.assertTrue(output.startswith("failed to resolve host: '{}'"
.format(mdns_instance).encode("utf8"))) .format(mdns_instance).encode("utf8")))
try: try:
yield yield
finally: finally:
# Perform best-effort disconnection. Discard the output. # Perform best-effort disconnection. Discard the output.
subprocess.Popen(["adb", "disconnect", serial], subprocess.Popen(["adb", "disconnect", serial],
stdout=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE).communicate() stderr=subprocess.PIPE).communicate()
@unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed") @unittest.skipIf(not is_zeroconf_installed(), "zeroconf library not installed")
def test_mdns_services_register_unregister(self): def test_mdns_services_register_unregister(self):
"""Ensure that `adb mdns services` correctly adds and removes a service """Ensure that `adb mdns services` correctly adds and removes a service
""" """
from zeroconf import IPVersion, ServiceInfo from zeroconf import IPVersion, ServiceInfo
with adb_server() as server_port: with adb_server() as server_port:
output = subprocess.check_output(["adb", "-P", str(server_port), output = subprocess.check_output(["adb", "-P", str(server_port),
"mdns", "services"]).strip() "mdns", "services"]).strip()
self.assertTrue(output.startswith(b"List of discovered mdns services")) self.assertTrue(output.startswith(b"List of discovered mdns services"))
"""TODO(joshuaduong): Add ipv6 tests once we have it working in adb""" """TODO(joshuaduong): Add ipv6 tests once we have it working in adb"""
"""Register/Unregister a service""" """Register/Unregister a service"""
with zeroconf_context(IPVersion.V4Only) as zc: with zeroconf_context(IPVersion.V4Only) as zc:
serv_instance = "my_fake_test_service" serv_instance = "my_fake_test_service"
serv_type = "_" + self.service_name + "._tcp." serv_type = "_" + self.service_name + "._tcp."
serv_ipaddr = socket.inet_aton("1.2.3.4") serv_ipaddr = socket.inet_aton("1.2.3.4")
serv_port = 12345 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, _):
service_info = ServiceInfo( service_info = ServiceInfo(
serv_type + "local.", serv_type + "local.",
name=serv_instance + "." + serv_type + "local.", name=serv_instance + "." + serv_type + "local.",
addresses=[serv_ipaddr], addresses=[serv_ipaddr],
port=port) port=serv_port)
with zeroconf_register_service(zc, service_info) as info: with zeroconf_register_service(zc, service_info) as info:
"""Give adb some time to register the service""" """Give adb some time to register the service"""
time.sleep(1) time.sleep(1)
self.assertTrue(any((serv_instance in line and serv_type in line) self.assertTrue(any((serv_instance in line and serv_type in line)
for line in MdnsTest._mdns_services(server_port))) 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""" """Give adb some time to unregister the service"""
time.sleep(1) time.sleep(1)
self.assertFalse(any((serv_instance in line and serv_type in line) self.assertFalse(any((serv_instance in line and serv_type in line)
for line in MdnsTest._mdns_services(server_port))) 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(): def main():
"""Main entrypoint.""" """Main entrypoint."""
random.seed(0) random.seed(0)