Merge changes from topic "cherrypicker-L50100000956399295:N18700001295161799" into tm-qpr-dev (69c1f9f9) · Commits · e / os / android_packages_modules_Bluetooth

android/pandora/mmi2grpc/mmi2grpc/init.py

+25 −7

Original line number	Diff line number	Diff line
		@@ -26,10 +26,13 @@ from mmi2grpc.a2dp import A2DPProxy
		from mmi2grpc.avrcp import AVRCPProxy
		from mmi2grpc.gatt import GATTProxy
		from mmi2grpc.hfp import HFPProxy
		from mmi2grpc.hid import HIDProxy
		from mmi2grpc.hogp import HOGPProxy
		from mmi2grpc.sdp import SDPProxy
		from mmi2grpc.sm import SMProxy
		from mmi2grpc._rootcanal import RootCanal
		from mmi2grpc._helpers import format_proxy
		from mmi2grpc._rootcanal import RootCanal

		from pandora_experimental.host_grpc import Host

		@@ -55,31 +58,41 @@ class IUT:
		"""
		self.port = port
		self.test = test
		self.rootcanal = None

		# Profile proxies.
		self._a2dp = None
		self._avrcp = None
		self._gatt = None
		self._hfp = None
		self._hid = None
		self._hogp = None
		self._sdp = None
		self._sm = None
		self._hogp = None

		def __enter__(self):
		"""Resets the IUT when starting a PTS test."""
		self.rootcanal = RootCanal()
		self.rootcanal.reconnect_phone()

		# Note: we don't keep a single gRPC channel instance in the IUT class
		# because reset is allowed to close the gRPC server.
		RootCanal().reconnect_phone()
		with grpc.insecure_channel(f'localhost:{self.port}') as channel:
		self._retry(Host(channel).HardReset)(wait_for_ready=True)

		def __exit__(self, exc_type, exc_value, exc_traceback):
		self.rootcanal.close()
		self.rootcanal = None

		self._a2dp = None
		self._avrcp = None
		self._gatt = None
		self._hfp = None
		self._hid = None
		self._hogp = None
		self._sdp = None
		self._sm = None
		self._hogp = None

		def _retry(self, func):

		@@ -154,6 +167,16 @@ class IUT:
		if not self._hfp:
		self._hfp = HFPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._hfp.interact(test, interaction, description, pts_address)
		# Handles HID MMIs.
		if profile in ('HID'):
		if not self._hid:
		self._hid = HIDProxy(grpc.insecure_channel(f'localhost:{self.port}'), self.rootcanal)
		return self._hid.interact(test, interaction, description, pts_address)
		# Handles HOGP MMIs.
		if profile in ('HOGP'):
		if not self._hogp:
		self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._hogp.interact(test, interaction, description, pts_address)
		# Handles SDP MMIs.
		if profile in ('SDP'):
		if not self._sdp:
		@@ -164,11 +187,6 @@ class IUT:
		if not self._sm:
		self._sm = SMProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._sm.interact(test, interaction, description, pts_address)
		# Handles HOGP MMIs.
		if profile in ('HOGP'):
		if not self._hogp:
		self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._hogp.interact(test, interaction, description, pts_address)

		# Handles unsupported profiles.
		code = format_proxy(profile, interaction, description)

android/pandora/mmi2grpc/mmi2grpc/_rootcanal.py

0 → 100644

+170 −0

Original line number	Diff line number	Diff line
		"""
		Copied from tools/rootcanal/scripts/test_channel.py
		"""

		import socket
		from time import sleep


		class Connection:

		def __init__(self, port):
		self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
		self._socket.connect(("localhost", port))

		def close(self):
		self._socket.close()

		def send(self, data):
		self._socket.sendall(data.encode())

		def receive(self, size):
		return self._socket.recv(size)


		class TestChannel:

		def __init__(self, port):
		self._connection = Connection(port)
		self._closed = False

		def close(self):
		self._connection.close()
		self._closed = True

		def send_command(self, name, args):
		args = [str(arg) for arg in args]
		name_size = len(name)
		args_size = len(args)
		self.lint_command(name, args, name_size, args_size)
		encoded_name = chr(name_size) + name
		encoded_args = chr(args_size) + "".join(chr(len(arg)) + arg for arg in args)
		command = encoded_name + encoded_args
		if self._closed:
		return
		self._connection.send(command)
		if name != "CLOSE_TEST_CHANNEL":
		return self.receive_response().decode()

		def receive_response(self):
		if self._closed:
		return b"Closed"
		size_chars = self._connection.receive(4)
		if not size_chars:
		return b"No response, assuming that the connection is broken"
		response_size = 0
		for i in range(0, len(size_chars) - 1):
		response_size \|= size_chars[i] << (8 * i)
		response = self._connection.receive(response_size)
		return response

		def lint_command(self, name, args, name_size, args_size):
		assert name_size == len(name) and args_size == len(args)
		try:
		name.encode()
		for arg in args:
		arg.encode()
		except UnicodeError:
		print("Unrecognized characters.")
		raise
		if name_size > 255 or args_size > 255:
		raise ValueError # Size must be encodable in one octet.
		for arg in args:
		if len(arg) > 255:
		raise ValueError # Size must be encodable in one octet.


		class RootCanal:

		def __init__(self):
		# port is CONTROL_ROOTCANAL_PORT defined in tradefed
		self.channel = TestChannel(port=6212)
		self.disconnected_dev_phys = None

		# discard initialization messages
		self.channel.receive_response()

		def close(self):
		self.channel.close()

		@staticmethod
		def _parse_device_list(raw):
		# time for some cursed parsing!
		categories = {}
		curr_category = None
		for line in raw.split("\n"):
		line = line.strip()
		if not line:
		continue
		if line[0].isdigit():
		# list entry
		if curr_category is None or ":" not in line:
		raise Exception("Failed to parse rootcanal device list output")
		curr_category.append(line.split(":", 1)[1])
		else:
		if line.endswith(":"):
		line = line[:-1]
		curr_category = []
		categories[line] = curr_category
		return categories

		@staticmethod
		def _parse_phy(raw):
		transport, idxs = raw.split(":")
		idxs = [int(x) for x in idxs.split(",") if x.strip()]
		return transport, idxs

		def reconnect_phone(self):
		raw_devices = None
		try:
		raw_devices = self.channel.send_command("list", [])
		devices = self._parse_device_list(raw_devices)

		for dev_i, name in enumerate(devices["Devices"]):
		# the default transports are always 0 and 1
		classic_phy = 0
		le_phy = 1
		if "beacon" in name:
		target_phys = [le_phy]
		elif "hci_device" in name:
		target_phys = [classic_phy, le_phy]

		for phy in target_phys:
		if dev_i not in self._parse_phy(devices["Phys"][phy])[1]:
		self.channel.send_command("add_device_to_phy", [dev_i, phy])
		except Exception as e:
		print(raw_devices, e)

		def disconnect_phy(self):
		# first, list all devices
		devices = self.channel.send_command("list", [])
		devices = self._parse_device_list(devices)
		dev_phys = []

		for phy_i, phy in enumerate(devices["Phys"]):
		_, idxs = self._parse_phy(phy)

		for dev_i in idxs:
		dev_phys.append((dev_i, phy_i))

		# now, disconnect all pairs
		for dev_i, phy_i in dev_phys:
		self.channel.send_command("del_device_from_phy", [dev_i, phy_i])

		devices = self.channel.send_command("list", [])
		devices = self._parse_device_list(devices)

		self.disconnected_dev_phys = dev_phys

		def reconnect_phy_if_needed(self):
		if self.disconnected_dev_phys is not None:
		for dev_i, phy_i in self.disconnected_dev_phys:
		self.channel.send_command("add_device_to_phy", [dev_i, phy_i])

		self.disconnected_dev_phys = None

		def reconnect_phy(self):
		if self.disconnected_dev_phys is None:
		raise Exception("cannot reconnect_phy before disconnect_phy")

		self.reconnect_phy_if_needed()

android/pandora/mmi2grpc/mmi2grpc/hid.py

0 → 100644

+185 −0

Original line number	Diff line number	Diff line
		from threading import Thread
		from time import sleep
		from mmi2grpc._helpers import assert_description, match_description
		from mmi2grpc._proxy import ProfileProxy

		from pandora_experimental.hid_grpc import HID
		from pandora_experimental.host_grpc import Host
		from pandora_experimental.hid_pb2 import HID_REPORT_TYPE_OUTPUT
		from mmi2grpc._rootcanal import RootCanal


		class HIDProxy(ProfileProxy):

		def __init__(self, channel, rootcanal):
		super().__init__()
		self.hid = HID(channel)
		self.host = Host(channel)
		self.rootcanal = rootcanal
		self.connection = None

		@assert_description
		def TSC_MMI_iut_enable_connection(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then using the Implementation Under Test (IUT) connect to the
		PTS.
		"""

		self.rootcanal.reconnect_phy_if_needed()
		self.connection = self.host.Connect(address=pts_addr).connection

		return "OK"

		@assert_description
		def TSC_MMI_iut_release_connection(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then release the HID connection from the Implementation Under
		Test (IUT) by closing the Interrupt Channel followed by the Control
		Channel.

		Description: This can be done using the anticipated L2CAP
		Disconnection Requests. If the host is unable to perform the connection
		request, the IUT may break the ACL or Baseband Link by going out of
		range.
		"""

		self.host.Disconnect(connection=self.connection)

		return "OK"

		@assert_description
		def TSC_MMI_iut_disable_connection(self, pts_addr: bytes, **kwargs):
		"""
		Disable the connection using the Implementation UnderTest (IUT).

		Note:
		The IUT may either disconnect the Interupt Control Channels or send a
		host initiated virtual cable unplug and wait for the PTS to disconnect
		the channels.
		"""

		self.host.Disconnect(connection=self.connection)
		self.connection = None

		return "OK"

		@assert_description
		def TSC_HID_MMI_iut_accept_connection_ready_confirm(self, **kwargs):
		"""
		Please prepare the IUT to accept connection from PTS and then click OK.
		"""

		self.rootcanal.reconnect_phy_if_needed()

		return "OK"

		@assert_description
		def TSC_MMI_iut_connectable_enter_pw_dev(self, **kwargs):
		"""
		Make the Implementation Under Test (IUT) connectable, then click Ok.
		"""

		self.rootcanal.reconnect_phy_if_needed()

		return "OK"

		@assert_description
		def TSC_HID_MMI_iut_accept_control_channel(self, pts_addr: bytes, **kwargs):
		"""
		Accept the control channel connection from the Implementation Under Test
		(IUT).
		"""

		return "OK"

		@assert_description
		def TSC_MMI_tester_release_connection(self, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in a state which will allow
		the PTS to perform an HID connection release, then click Ok.

		Note: The
		PTS will send an L2CAP disconnect request for the Interrupt channel,
		then the control channel.
		"""

		return "OK"

		@assert_description
		def TSC_MMI_host_iut_prepare_to_receive_pointing_data(self, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in a state to receive and
		verify HID pointing data, then click Ok.
		"""

		return "OK"

		@assert_description
		def TSC_MMI_host_iut_verify_pointing_data(self, **kwargs):
		"""
		Verify that the pointer on the Implementation Under Test (IUT) moved to
		the left (X< 0), then click Ok.
		"""

		# TODO: implement!

		return "OK"

		@assert_description
		def TSC_MMI_host_send_output_report(self, pts_addr: bytes, **kwargs):
		"""
		Send an output report from the HOST.
		"""

		self.hid.SendHostReport(
		address=pts_addr,
		report_type=HID_REPORT_TYPE_OUTPUT,
		report="8", # keyboard enable num-lock
		)

		return "OK"

		@match_description
		def TSC_MMI_verify_output_report(self, **kwargs):
		"""
		Verify that the output report is correct. nnOutput Report =0(?:0\|1)'
		"""

		# TODO: check the report matches the num-lock setting

		return "OK"

		@assert_description
		def TSC_MMI_rf_shield_iut_or_tester(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then perform one of the following actions:

		1. Move the PTS
		and Implementation Under Test (IUT) out of range of each other.
		2. Place
		either the PTS or IUT in an RF sheild box.
		"""

		def disconnect():
		sleep(2)
		self.rootcanal.disconnect_phy()

		Thread(target=disconnect).start()

		return "OK"

		@assert_description
		def TSC_MMI_iut_auto_connection(self, pts_addr: bytes, **kwargs):
		"""
		Click OK, then initiate a HID connection automatically from the IUT to
		the PTS
		"""

		def connect():
		sleep(1)
		self.rootcanal.reconnect_phy_if_needed()
		self.connection = self.host.Connect(address=pts_addr).connection

		Thread(target=connect).start()

		return "OK"

android/pandora/server/Android.bp

+3 −0

Original line number	Diff line number	Diff line
		@@ -134,6 +134,7 @@ genrule {
		"proto/pandora_experimental/avrcp.proto",
		"proto/pandora_experimental/gatt.proto",
		"proto/pandora_experimental/hfp.proto",
		"proto/pandora_experimental/hid.proto",
		"proto/pandora_experimental/host.proto",
		"proto/pandora_experimental/security.proto",
		],
		@@ -146,6 +147,8 @@ genrule {
		"pandora_experimental/gatt_pb2.py",
		"pandora_experimental/hfp_grpc.py",
		"pandora_experimental/hfp_pb2.py",
		"pandora_experimental/hid_grpc.py",
		"pandora_experimental/hid_pb2.py",
		"pandora_experimental/host_grpc.py",
		"pandora_experimental/host_pb2.py",
		"pandora_experimental/security_grpc.py",

android/pandora/server/configs/PtsBotTest.xml

+2 −1

Original line number	Diff line number	Diff line
		@@ -32,10 +32,11 @@
		<option name="profile" value="HFP/AG/HFI" />
		<option name="profile" value="HFP/AG/SLC" />
		<option name="profile" value="HFP/AG/TCA" />
		<option name="profile" value="HID/HOS" />
		<option name="profile" value="HOGP/RH" />
		<option name="profile" value="SDP/SR" />
		<option name="profile" value="SM/CEN/EKS" />
		<option name="profile" value="SM/CEN/JW" />
		<option name="profile" value="SM/CEN/KDU" />
		<option name="profile" value="HOGP/RH" />
		</test>
		</configuration>