Merge "adb: don't use parameterized in test_adb.py."
This commit is contained in:
Treehugger Robot
Gerrit Code Review
commit
648266ab1c
1 changed files with 97 additions and 87 deletions
184
adb/test_adb.py
184
adb/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:]]
|
||||
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)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue