Merge "[Pandora] First HFP tests involving calls" (287965ab) · Commits · e / os / android_packages_modules_Bluetooth

android/pandora/mmi2grpc/mmi2grpc/init.py

+50 −33

Original line number	Diff line number	Diff line
		@@ -63,6 +63,7 @@ class IUT:
		self.pandora_server_port = int(args[0]) if len(args) > 0 else PANDORA_SERVER_PORT
		self.rootcanal_control_port = int(args[1]) if len(args) > 1 else ROOTCANAL_CONTROL_PORT
		self.modem_simulator_port = int(args[2]) if len(args) > 2 else MODEM_SIMULATOR_PORT

		self.test = test
		self.rootcanal = None
		self.modem = None
		@@ -89,7 +90,7 @@ class IUT:

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

		def __exit__(self, exc_type, exc_value, exc_traceback):
		@@ -123,7 +124,7 @@ class IUT:
		if tries >= MAX_RETRIES:
		raise
		else:
		print(f'Retry {func.__name__}: {tries}/{MAX_RETRIES}')
		print(f"Retry {func.__name__}: {tries}/{MAX_RETRIES}")
		time.sleep(1)

		return wrapper
		@@ -137,7 +138,7 @@ class IUT:
		mut_address = None

		def read_local_address():
		with grpc.insecure_channel(f'localhost:{self.pandora_server_port}') as channel:
		with grpc.insecure_channel(f"localhost:{self.pandora_server_port}") as channel:
		nonlocal mut_address
		mut_address = self._retry(Host(channel).ReadLocalAddress)(wait_for_ready=True).address

		@@ -150,8 +151,16 @@ class IUT:
		else:
		return mut_address

		def interact(self, pts_address: bytes, profile: str, test: str, interaction: str, description: str, style: str,
		**kwargs) -> str:
		def interact(
		self,
		pts_address: bytes,
		profile: str,
		test: str,
		interaction: str,
		description: str,
		style: str,
		**kwargs,
		) -> str:
		"""Routes MMI calls to corresponding profile proxy.

		Args:
		@@ -162,70 +171,78 @@ class IUT:
		description: MMI description.
		style: MMI popup style, unused for now.
		"""
		print(f'{profile} mmi: {interaction}', file=sys.stderr)
		print(f"{profile} mmi: {interaction}", file=sys.stderr)

		# Handles A2DP and AVDTP MMIs.
		if profile in ('A2DP', 'AVDTP'):
		if profile in ("A2DP", "AVDTP"):
		if not self._a2dp:
		self._a2dp = A2DPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._a2dp = A2DPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._a2dp.interact(test, interaction, description, pts_address)
		# Handles AVRCP and AVCTP MMIs.
		if profile in ('AVRCP', 'AVCTP'):
		if profile in ("AVRCP", "AVCTP"):
		if not self._avrcp:
		self._avrcp = AVRCPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._avrcp = AVRCPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._avrcp.interact(test, interaction, description, pts_address)
		# Handles GATT MMIs.
		if profile in ('GATT'):
		if profile in ("GATT"):
		if not self._gatt:
		self._gatt = GATTProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._gatt = GATTProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._gatt.interact(test, interaction, description, pts_address)
		# Handles GAP MMIs.
		if profile in ('GAP'):
		if profile in ("GAP"):
		if not self._gap:
		self._gap = GAPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._gap = GAPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._gap.interact(test, interaction, description, pts_address)
		# Handles HFP MMIs.
		if profile in ('HFP'):
		if profile in ("HFP"):
		if not self._hfp:
		self._hfp = HFPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._hfp = HFPProxy(
		test,
		grpc.insecure_channel(f"localhost:{self.pandora_server_port}"),
		self.rootcanal,
		self.modem,
		)
		return self._hfp.interact(test, interaction, description, pts_address)
		# Handles HID MMIs.
		if profile in ('HID'):
		if profile in ("HID"):
		if not self._hid:
		self._hid = HIDProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'), self.rootcanal)
		self._hid = HIDProxy(
		grpc.insecure_channel(f"localhost:{self.pandora_server_port}"),
		self.rootcanal,
		)
		return self._hid.interact(test, interaction, description, pts_address)
		# Handles HOGP MMIs.
		if profile in ('HOGP'):
		if profile in ("HOGP"):
		if not self._hogp:
		self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._hogp = HOGPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._hogp.interact(test, interaction, description, pts_address)
		# Instantiates L2CAP proxy and reroutes corresponding MMIs to it.
		if profile in ('L2CAP'):
		if profile in ("L2CAP"):
		if not self._l2cap:
		self._l2cap = L2CAPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._l2cap = L2CAPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._l2cap.interact(test, interaction, description, pts_address)
		# Handles RFCOMM MMIs.
		if profile in ('RFCOMM'):
		if profile in ("RFCOMM"):
		if not self._rfcomm:
		self._rfcomm = RFCOMMProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._rfcomm = RFCOMMProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._rfcomm.interact(test, interaction, description, pts_address)
		# Handles SDP MMIs.
		if profile in ('SDP'):
		if profile in ("SDP"):
		if not self._sdp:
		self._sdp = SDPProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._sdp = SDPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._sdp.interact(test, interaction, description, pts_address)
		# Handles SM MMIs.
		if profile in ('SM'):
		if profile in ("SM"):
		if not self._sm:
		self._sm = SMProxy(grpc.insecure_channel(f'localhost:{self.pandora_server_port}'))
		self._sm = SMProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"))
		return self._sm.interact(test, interaction, description, pts_address)

		# Handles unsupported profiles.
		code = format_proxy(profile, interaction, description)
		error_msg = (f'Missing {profile} proxy and mmi: {interaction}\n'
		f'Create a {profile.lower()}.py in mmi2grpc/:\n\n{code}\n'
		f'Then, instantiate the corresponding proxy in __init__.py\n'
		f'Finally, create a {profile.lower()}.proto in proto/pandora/'
		f'and generate the corresponding interface.')
		error_msg = (f"Missing {profile} proxy and mmi: {interaction}\n"
		f"Create a {profile.lower()}.py in mmi2grpc/:\n\n{code}\n"
		f"Then, instantiate the corresponding proxy in __init__.py\n"
		f"Finally, create a {profile.lower()}.proto in proto/pandora/"
		f"and generate the corresponding interface.")

		assert False, error_msg

android/pandora/mmi2grpc/mmi2grpc/hfp.py

+103 −4

Original line number	Diff line number	Diff line
		@@ -19,6 +19,7 @@ from mmi2grpc._proxy import ProfileProxy
		from pandora_experimental.hfp_grpc import HFP
		from pandora_experimental.host_grpc import Host
		from pandora_experimental.security_grpc import Security
		from pandora_experimental.hfp_pb2 import AudioPath

		import sys
		import threading
		@@ -30,17 +31,23 @@ WAIT_DELAY_BEFORE_CONNECTION = 2
		# The tests needs the MMI to accept pairing confirmation request.
		NEEDS_WAIT_CONNECTION_BEFORE_TEST = {'HFP/AG/WBS/BV-01-I', 'HFP/AG/SLC/BV-05-I'}

		IXIT_PHONE_NUMBER = 42


		class HFPProxy(ProfileProxy):

		def __init__(self, channel):
		def __init__(self, test, channel, rootcanal, modem):
		super().__init__(channel)
		self.hfp = HFP(channel)
		self.host = Host(channel)
		self.security = Security(channel)
		self.rootcanal = rootcanal
		self.modem = modem

		self.connection = None

		self._auto_confirm_requests()

		def asyncWaitConnection(self, pts_addr, delay=WAIT_DELAY_BEFORE_CONNECTION):
		"""
		Send a WaitConnection in a grpc callback
		@@ -97,7 +104,12 @@ class HFPProxy(ProfileProxy):
		Implementation Under Test (IUT).
		"""

		def connect():
		time.sleep(2)
		self.connection = self.host.Connect(address=pts_addr).connection

		threading.Thread(target=connect).start()

		return "OK"

		@assert_description
		@@ -118,11 +130,13 @@ class HFPProxy(ProfileProxy):
		Implementation Under Test (IUT).
		"""

		def go():
		self.connection = self.host.GetConnection(address=pts_addr).connection

		def disable_slc():
		time.sleep(2)
		self.hfp.DisableSlc(connection=self.connection)

		threading.Thread(target=go).start()
		threading.Thread(target=disable_slc).start()

		return "OK"

		@@ -147,3 +161,88 @@ class HFPProxy(ProfileProxy):
		self.hfp.SetBatteryLevel(connection=self.connection, battery_percentage=42)

		return "OK"

		@assert_description
		def TSC_ag_iut_enable_call(self, **kwargs):
		"""
		Click Ok, then place a call from an external line to the Implementation
		Under Test (IUT). Do not answer the call unless prompted to do so.
		"""

		def enable_call():
		time.sleep(2)
		self.modem.call(IXIT_PHONE_NUMBER)

		threading.Thread(target=enable_call).start()

		return "OK"

		@assert_description
		def TSC_verify_audio(self, **kwargs):
		"""
		Verify the presence of an audio connection, then click Ok.
		"""

		# TODO

		return "OK"

		@assert_description
		def TSC_ag_iut_disable_call_external(self, **kwargs):
		"""
		Click Ok, then end the call using the external terminal.
		"""

		def disable_call_external():
		time.sleep(2)
		self.hfp.DeclineCall()

		threading.Thread(target=disable_call_external).start()

		return "OK"

		@assert_description
		def TSC_iut_enable_audio_using_codec(self, **kwargs):
		"""
		Click OK, then initiate an audio connection using the Codec Connection
		Setup procedure.
		"""

		return "OK"

		@assert_description
		def TSC_iut_disable_audio(self, **kwargs):
		"""
		Click Ok, then close the audio connection (SCO) between the
		Implementation Under Test (IUT) and the PTS. Do not close the serivice
		level connection (SLC) or power-off the IUT.
		"""

		def disable_audio():
		time.sleep(2)
		self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_SPEAKERS)

		threading.Thread(target=disable_audio).start()

		return "OK"

		@assert_description
		def TSC_verify_no_audio(self, **kwargs):
		"""
		Verify the absence of an audio connection (SCO), then click Ok.
		"""

		return "OK"

		def _auto_confirm_requests(self, times=None):

		def task():
		cnt = 0
		pairing_events = self.security.OnPairing()
		for event in pairing_events:
		if event.WhichOneof('method') in {"just_works", "numeric_comparison"}:
		if times is None or cnt < times:
		cnt += 1
		pairing_events.send(event=event, confirm=True)

		threading.Thread(target=task).start()

android/pandora/server/configs/pts_bot_tests_config.json

+10 −7

Original line number	Diff line number	Diff line
		@@ -234,6 +234,11 @@
		"GATT/SR/UNS/BI-01-C",
		"GATT/SR/UNS/BI-02-C",
		"HFP/AG/DIS/BV-01-I",
		"HFP/AG/ACC/BV-08-I",
		"HFP/AG/ACC/BV-09-I",
		"HFP/AG/ACR/BV-01-I",
		"HFP/AG/ACR/BV-02-I",
		"HFP/AG/ACS/BI-14-I",
		"HFP/AG/HFI/BV-02-I",
		"HFP/AG/PSI/BV-03-C",
		"HFP/AG/SLC/BV-09-I",
		@@ -581,10 +586,8 @@
		"HFP/AG/ACS/BV-04-I",
		"HFP/AG/ACS/BV-08-I",
		"HFP/AG/ACS/BV-11-I",
		"HFP/AG/ACS/BI-14-I",
		"HFP/AG/ACC/BI-13-I",
		"HFP/AG/ACS/BV-16-I",
		"HFP/AG/ACR/BV-01-I",
		"HFP/AG/ACR/BV-02-I",
		"HFP/AG/CLI/BV-01-I",
		"HFP/AG/ICA/BV-04-I",
		"HFP/AG/ICA/BV-07-I",
		@@ -619,12 +622,9 @@
		"HFP/AG/SLC/BV-05-I",
		"HFP/AG/SLC/BV-06-I",
		"HFP/AG/SLC/BV-07-I",
		"HFP/AG/ACC/BV-08-I",
		"HFP/AG/ACC/BV-09-I",
		"HFP/AG/ACC/BV-10-I",
		"HFP/AG/ACC/BV-11-I",
		"HFP/AG/ACC/BI-12-I",
		"HFP/AG/ACC/BI-13-I",
		"HFP/AG/ACC/BI-14-I",
		"HFP/AG/ACC/BV-15-I",
		"HFP/AG/SDP/BV-01-I",
		@@ -1672,7 +1672,10 @@
		"GATT": {},
		"GAVDP": {},
		"HCI": {},
		"HFP": {},
		"HFP": {
		"TSPX_phone_number": "42",
		"TSPX_second_phone_number": "42"
		},
		"HID": {},
		"HOGP": {},
		"HSP": {},