Merge changes from topic "cherrypicker-L94700000958095889:N77900001327069473" into tm-qpr-dev (156c1792) · Commits · e / os / android_packages_modules_Bluetooth

android/pandora/mmi2grpc/mmi2grpc/avrcp.py

+166 −0

Original line number	Diff line number	Diff line
		@@ -793,3 +793,169 @@ class AVRCPProxy(ProfileProxy):
		self.mediaplayer.Play()

		return "OK"

		@assert_description
		def _mmi_1016(self, test: str, pts_addr: bytes, **kwargs):
		"""
		Create an AVDTP signaling channel.

		Action: Create an audio or video
		connection with PTS.
		"""
		self.connection = self.host.Connect(address=pts_addr).connection
		if "TG" in test:
		try:
		self.source = self.a2dp.OpenSource(connection=self.connection).source
		except RpcError:
		pass
		else:
		try:
		self.sink = self.a2dp.WaitSink(connection=self.connection).sink
		except RpcError:
		pass

		return "OK"

		@assert_description
		def TSC_AVCTP_mmi_send_AVCT_ConnectReq(self, pts_addr: bytes, **kwargs):
		"""
		Using the Upper Tester, send an AVCT_ConnectReq command to the IUT with
		the following input parameter values:
		* BD_ADDR = BD_ADDRLower_Tester
		* PID = PIDTest_System

		The IUT should then initiate an
		L2CAP_ConnectReq.
		"""

		return "OK"

		@assert_description
		def TSC_AVCTP_mmi_verify_ConnectCfm_CB(self, pts_addr: bytes, **kwargs):
		"""
		Press 'OK' if the following conditions were met :

		1. The IUT returns
		the following AVCT_ConnectReq output parameters to the Upper Tester:
		* Result = 0x0000 (Event successfully registered)

		2. The IUT calls the
		ConnectCfm_CBTest_System function in the Upper Tester with the following
		parameters:
		* BD_ADDR = BD_ADDRLower_Tester
		* Connect Result =
		0x0000 (L2CAP Connect Request successful)
		* Config Result = 0x0000
		(L2CAP Configure successful)
		* Status = L2CAP Connect Request Status
		"""

		return "OK"

		@assert_description
		def TSC_AVCTP_mmi_register_DisconnectCfm_CB(self, pts_addr: bytes, **kwargs):
		"""
		Using the Upper Tester register the function DisconnectCfm_CBTest_System
		for callback on the AVCT_Disconnect_Cfm event by sending an
		AVCT_EventRegistration command to the IUT with the following parameter
		values:
		* Event = AVCT_Disconnect_Cfm
		* Callback =
		DisconnectCfm_CBTest_System
		* PID = PIDTest_System

		Press 'OK' to
		continue once the IUT has responded.
		"""

		return "OK"

		def TSC_AVCTP_mmi_send_AVCT_Disconnect_Req(self, test: str, pts_addr: bytes, **kwargs):
		"""
		Using the Upper Tester send an AVCT_DisconnectReq command to the IUT
		with the following parameter values:
		* BD_ADDR = BD_ADDRLower_Tester
		* PID = PIDTest_System

		The IUT should then initiate an
		L2CAP_DisconnectReq.
		"""
		# Currently disconnect is required in TG role
		if "TG" in test:
		if self.connection is None:
		self.connection = self.host.GetConnection(address=pts_addr).connection
		time.sleep(3)
		self.host.Disconnect(connection=self.connection)
		self.connection = None

		return "OK"

		@assert_description
		def TSC_AVCTP_mmi_verify_DisconnectCfm_CB(self, **kwargs):
		"""
		Press 'OK' if the following conditions were met :

		1. The IUT returns
		the following AVCT_EventRegistration output parameters to the Upper
		Tester:
		* Result = 0x0000 (Event successfully registered)

		2. The IUT
		calls the DisconnectCfm_CBTest_System function in the Upper Tester with
		the following parameter values:
		* BD_ADDR = BD_ADDRLower_Tester
		*
		Disconnect Result = 0x0000 (L2CAP disconnect success)

		3. The IUT
		returns the following AVCT_DisconnectReq output parameter values to the
		Upper Tester:
		* RSP = 0x0000 (Request accepted)
		"""

		return "OK"

		@assert_description
		def TSC_AVCTP_mmi_send_AVCT_SendMessage_TG(self, **kwargs):
		"""
		Upon a call to the call back function MessageInd_CBTest_System, use the
		Upper Tester to send an AVCT_SendMessage command to the IUT with the
		following parameter values:
		* BD_ADDR = BD_ADDRTest_System
		*
		Transaction = TRANSTest_System
		* Type = CRTest_System = 1 (Response
		Message)
		* PID = PIDTest_System
		* Data = ADDRESSdata_buffer
		(Buffer containing DATA[]Upper_Tester)
		* Length =
		LengthOf(DATA[]Upper_Tester) <= MTU – 3bytes
		"""

		return "OK"

		@assert_description
		def TSC_AVCTP_mmi_verify_MessageInd_CB_TG(self, **kwargs):
		"""
		Press 'OK' if the following conditions were met :

		1. The
		MessageInd_CBTest_System function in the Upper Tester is called with the
		following parameters:
		* BD_ADDR = BD_ADDRLower_Tester
		*
		Transaction = TRANSTest_System
		* Type = 0x00 (Command message)
		*
		Data = ADDRESSdata_buffer (Buffer containing DATA[]Lower_Tester)
		*
		Length = LengthOf(DATA[]Lower_Tester)

		2. the IUT returns the following
		AVCT_SendMessage output parameters to the Upper Tester:
		* Result =
		0x0000 (Request accepted)
		"""

		return "OK"

android/pandora/mmi2grpc/mmi2grpc/hfp.py

+306 −15

Original line number	Diff line number	Diff line
		@@ -18,7 +18,7 @@ from mmi2grpc._proxy import ProfileProxy

		from pandora_experimental.hfp_grpc import HFP
		from pandora_experimental.host_grpc import Host
		from pandora_experimental.host_pb2 import ConnectabilityMode
		from pandora_experimental.host_pb2 import ConnectabilityMode, DiscoverabilityMode
		from pandora_experimental.security_grpc import Security, SecurityStorage
		from pandora_experimental.hfp_pb2 import AudioPath

		@@ -96,6 +96,9 @@ class HFPProxy(ProfileProxy):
		if not self.connection:
		self.connection = self.host.Connect(address=pts_addr).connection

		if "HFP/HF" in test:
		self.hfp.EnableSlcAsHandsfree(connection=self.connection)
		else:
		self.hfp.EnableSlc(connection=self.connection)

		threading.Thread(target=enable_slc).start()
		@@ -137,7 +140,7 @@ class HFPProxy(ProfileProxy):
		return "OK"

		@assert_description
		def TSC_iut_disable_slc(self, pts_addr: bytes, **kwargs):
		def TSC_iut_disable_slc(self, test: str, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then disable the service level connection using the
		Implementation Under Test (IUT).
		@@ -147,6 +150,9 @@ class HFPProxy(ProfileProxy):

		def disable_slc():
		time.sleep(2)
		if "HFP/HF" in test:
		self.hfp.DisableSlcAsHandsfree(connection=self.connection)
		else:
		self.hfp.DisableSlc(connection=self.connection)

		threading.Thread(target=disable_slc).start()
		@@ -225,15 +231,20 @@ class HFPProxy(ProfileProxy):
		return "OK"

		@assert_description
		def TSC_iut_disable_audio(self, **kwargs):
		def TSC_iut_disable_audio(self, test: str, pts_addr: bytes, **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.
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def disable_audio():
		time.sleep(2)
		if "HFP/HF" in test:
		self.hfp.DisconnectToAudioAsHandsfree(connection=self.connection)
		else:
		self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_SPEAKERS)

		threading.Thread(target=disable_audio).start()
		@@ -249,14 +260,19 @@ class HFPProxy(ProfileProxy):
		return "OK"

		@assert_description
		def TSC_iut_enable_audio(self, **kwargs):
		def TSC_iut_enable_audio(self, test: str, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then initiate an audio connection (SCO) from the
		Implementation Under Test (IUT) to the PTS.
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def enable_audio():
		time.sleep(2)
		if "HFP/HF" in test:
		self.hfp.ConnectToAudioAsHandsfree(connection=self.connection)
		else:
		self.hfp.SetAudioPath(audio_path=AudioPath.AUDIO_PATH_HANDSFREE)

		threading.Thread(target=enable_audio).start()
		@@ -579,12 +595,13 @@ class HFPProxy(ProfileProxy):
		return "OK"

		@assert_description
		def TSC_prepare_iut_for_vra(self, pts_addr: bytes, **kwargs):
		def TSC_prepare_iut_for_vra(self, pts_addr: bytes, test: str, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in a state which will allow a
		request from the PTS to activate voice recognition, then click Ok.
		"""

		if "HFP/HF" not in test:
		self.hfp.SetVoiceRecognition(
		enabled=True,
		connection=self.host.GetConnection(address=pts_addr).connection,
		@@ -592,6 +609,15 @@ class HFPProxy(ProfileProxy):

		return "OK"

		@assert_description
		def TSC_prepare_iut_for_vrd(self, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in a state which will allow a
		voice recognition deactivation from PTS, then click Ok.
		"""

		return "OK"

		@assert_description
		def TSC_ag_iut_clear_call_history(self, **kwargs):
		"""
		@@ -604,20 +630,285 @@ class HFPProxy(ProfileProxy):
		return "OK"

		@assert_description
		def TSC_reject_call(self, **kwargs):
		def TSC_reject_call(self, test: str, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then reject the incoming call using the Implemention Under
		Test (IUT).
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def reject_call():
		time.sleep(2)
		if "HFP/HF" in test:
		self.hfp.DeclineCallAsHandsfree(connection=self.connection)
		else:
		self.hfp.DeclineCall()

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

		return "OK"

		@assert_description
		def TSC_hf_iut_answer_call(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then answer the incoming call using the Implementation Under
		Test (IUT).
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def answer_call():
		time.sleep(2)
		self.hfp.AnswerCallAsHandsfree(connection=self.connection)

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

		return "OK"

		@assert_description
		def TSC_iut_disable_audio_poweroff_ok(self, **kwargs):
		"""
		Click Ok, then close the audio connection (SCO) by one of the following
		ways:

		1. Close the service level connection (SLC)
		2. Powering off the
		Implementation Under Test (IUT)
		"""

		self.host.Reset()

		return "OK"

		@assert_description
		def TSC_verify_inband_ring(self, **kwargs):
		"""
		Verify that the in-band ringtone is audible, then click Ok.
		"""

		return "OK"

		@assert_description
		def TSC_verify_inband_ring_muting(self, **kwargs):
		"""
		Verify that the in-band ringtone is not audible , then click Ok.
		"""

		return "OK"

		@assert_description
		def TSC_hf_iut_disable_call(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then end the call process from the Implementation Under Test
		(IUT).
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def disable_call():
		time.sleep(2)
		self.hfp.EndCallAsHandsfree(connection=self.connection)

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

		return "OK"

		@assert_description
		def TSC_mute_inband_ring_iut(self, **kwargs):
		"""
		Mute the in-band ringtone on the Implementation Under Test (IUT) and
		then click OK.
		"""

		return "OK"

		@assert_description
		def TSC_verify_iut_alerting(self, **kwargs):
		"""
		Verify that the Implementation Under Test (IUT) is generating a local
		alert, then click Ok.
		"""

		return "OK"

		@assert_description
		def TSC_verify_iut_not_alerting(self, **kwargs):
		"""
		Verify that the Implementation Under Test (IUT) is not generating a
		local alert.
		"""

		return "OK"

		@assert_description
		def TSC_hf_iut_enable_call_number(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then place an outgoing call from the Implementation Under Test
		(IUT) using an enterted phone number.
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def disable_call():
		time.sleep(2)
		self.hfp.MakeCallAsHandsfree(connection=self.connection, number="42")

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

		return "OK"

		@assert_description
		def TSC_hf_iut_enable_call_memory(self, **kwargs):
		"""
		Click Ok, then place an outgoing call from the Implementation Under Test
		(IUT) by entering the memory index. For further clarification please
		see the HFP 1.5 Specification.
		"""

		return "OK"

		@assert_description
		def TSC_hf_iut_call_swap_then_disable_held_alternative(self, pts_addr: bytes, **kwargs):
		"""
		Using the Implementation Under Test (IUT), perform one of the following
		two actions:

		1. Click OK, make the held/waiting call active, disabling
		the active call.
		2. Click OK, make the held/waiting call active, placing
		the active call on hold.
		"""

		self.connection = self.host.GetConnection(address=pts_addr).connection

		def call_swap_then_disable_held_alternative():
		time.sleep(2)
		self.hfp.CallTransferAsHandsfree(connection=self.connection)

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

		return "OK"

		@assert_description
		def TSC_iut_make_discoverable(self, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in discoverable mode, then
		click Ok.
		"""

		self.host.SetDiscoverabilityMode(mode=DiscoverabilityMode.DISCOVERABLE_GENERAL)

		return "OK"

		@assert_description
		def TSC_iut_accept_connection(self, **kwargs):
		"""
		Click Ok, then accept the pairing and connection requests on the
		Implementation Under Test (IUT), if prompted.
		"""

		return "OK"

		@assert_description
		def TSC_voice_recognition_enable_iut(self, pts_addr: bytes, **kwargs):
		"""
		Using the Implementation Under Test (IUT), activate voice recognition.
		"""

		self.hfp.SetVoiceRecognitionAsHandsfree(
		enabled=True,
		connection=self.host.GetConnection(address=pts_addr).connection,
		)

		return "OK"

		@assert_description
		def TSC_voice_recognition_disable_iut(self, pts_addr: bytes, **kwargs):
		"""
		Using the Implementation Under Test (IUT), deactivate voice recognition.
		"""

		self.hfp.SetVoiceRecognitionAsHandsfree(
		enabled=False,
		connection=self.host.GetConnection(address=pts_addr).connection,
		)

		return "OK"

		@match_description
		def TSC_dtmf_send(self, pts_addr: bytes, dtmf: str, **kwargs):
		r"""
		Send the DTMF code, then click Ok. (?P<dtmf>.*)
		"""

		self.hfp.SendDtmfFromHandsfree(
		connection=self.host.GetConnection(address=pts_addr).connection,
		code=dtmf[0].encode("ascii")[0],
		)

		return "OK"

		@assert_description
		def TSC_verify_hf_iut_reports_held_and_active_call(self, **kwargs):
		"""
		Verify that the Implementation Under Test (IUT) interprets both held and
		active call signals, then click Ok. If applicable, verify that the
		information is correctly displayed on the IUT, then click Ok.
		"""

		return "OK"

		def TSC_rf_shield_iut_or_pts(self, **kwargs):
		"""
		Click Ok, then move the PTS and the Implementation Under Test (IUT) out
		of range of each other by performing one of the following IUT specific
		actions:

		1. Hands Free (HF) IUT - Place the IUT in the RF shield box or
		physically take out of range from the PTS.

		2. Audio Gateway (AG) IUT-
		Physically take the IUT out range. Do not place in the RF shield box as
		it will interfere with the cellular network.

		Note: The PTS can also be
		placed in the RF shield box if necessary.
		"""

		def shield_iut_or_pts():
		time.sleep(2)
		self.rootcanal.disconnect_phy()

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

		return "OK"

		@assert_description
		def TSC_rf_shield_open(self, **kwargs):
		"""
		Click Ok, then remove the Implementation Under Test (IUT) and/or the PTS
		from the RF shield. If the out of range method was used, bring the IUT
		and PTS back within range.
		"""

		def shield_open():
		time.sleep(2)
		self.rootcanal.reconnect_phy_if_needed()

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

		return "OK"

		@match_description
		def TSC_verify_speaker_volume(self, volume: str, **kwargs):
		r"""
		Verify that the Hands Free \(HF\) speaker volume is displayed correctly on
		the Implementation Under Test \(IUT\).(?P<volume>[0-9]*)
		"""

		return "OK"

		def _auto_confirm_requests(self, times=None):

		def task():

android/pandora/mmi2grpc/mmi2grpc/l2cap.py

+33 −5

Original line number	Diff line number	Diff line
		@@ -16,6 +16,7 @@ from typing import Optional
		class L2CAPProxy(ProfileProxy):
		test_status_map = {} # record test status and pass them between MMI
		LE_DATA_PACKET_LARGE = "data: LE_DATA_PACKET_LARGE"
		LE_DATA_PACKET1 = "data: LE_PACKET1"
		connection: Optional[Connection] = None

		def __init__(self, channel):
		@@ -101,6 +102,10 @@ class L2CAPProxy(ProfileProxy):
		)
		# not strictly necessary, but can save time on waiting connection
		tests_to_open_bluetooth_server_socket = [
		"L2CAP/COS/CFC/BV-01-C",
		"L2CAP/COS/CFC/BV-02-C",
		"L2CAP/COS/CFC/BV-03-C",
		"L2CAP/COS/CFC/BV-04-C",
		"L2CAP/LE/CFC/BV-03-C",
		"L2CAP/LE/CFC/BV-05-C",
		"L2CAP/LE/CFC/BV-06-C",
		@@ -138,7 +143,7 @@ class L2CAPProxy(ProfileProxy):
		return "OK"

		@match_description
		def MMI_UPPER_TESTER_CONFIRM_LE_DATA(self, sent_data: str, **kwargs):
		def MMI_UPPER_TESTER_CONFIRM_LE_DATA(self, sent_data: str, test: str, **kwargs):
		"""
		Did the Upper Tester send the data (?P<sent_data>[0-9A-F]*) to to the
		PTS\? Click Yes if it matched, otherwise click No.
		@@ -146,10 +151,13 @@ class L2CAPProxy(ProfileProxy):
		Description: The Implementation Under Test
		\(IUT\) send data is receive correctly in the PTS.
		"""
		hex_LE_DATA_PACKET_LARGE = self.LE_DATA_PACKET_LARGE.encode("utf-8").hex().upper()
		if sent_data != hex_LE_DATA_PACKET_LARGE:
		print(f"data not match, sent_data:{sent_data} and {hex_LE_DATA_PACKET_LARGE}", file=sys.stderr)
		raise Exception(f"data not match, sent_data:{sent_data} and {hex_LE_DATA_PACKET_LARGE}")
		if test == 'L2CAP/COS/CFC/BV-02-C':
		hex_LE_DATA_PACKET = self.LE_DATA_PACKET1.encode("utf-8").hex().upper()
		else:
		hex_LE_DATA_PACKET = self.LE_DATA_PACKET_LARGE.encode("utf-8").hex().upper()
		if sent_data != hex_LE_DATA_PACKET:
		print(f"data not match, sent_data:{sent_data} and {hex_LE_DATA_PACKET}", file=sys.stderr)
		raise Exception(f"data not match, sent_data:{sent_data} and {hex_LE_DATA_PACKET}")
		return "OK"

		@assert_description
		@@ -424,3 +432,23 @@ class L2CAPProxy(ProfileProxy):
		"""

		return "OK"

		@assert_description
		def MMI_UPPER_TESTER_SEND_LE_DATA_PACKET1(self, **kwargs):
		"""
		Upper Tester command IUT to send a non-segmented LE data packet to the
		PTS with any values.
		Description : The Implementation Under Test(IUT)
		should send none segmantation LE frame of LE data to the PTS.
		"""
		self.l2cap.SendData(connection=self.connection, data=bytes(self.LE_DATA_PACKET1, "utf-8"))
		return "OK"

		@assert_description
		def MMI_IUT_SEND_L2CAP_DATA(self, **kwargs):
		"""
		Using the Implementation Under Test(IUT), send L2CAP_Data over the
		assigned channel with correct DCID to the PTS.
		"""

		return "OK"

android/pandora/server/configs/PtsBotTest.xml

+2 −0

Original line number	Diff line number	Diff line
		@@ -38,9 +38,11 @@
		<option name="profile" value="GAP" />
		<option name="profile" value="GATT" />
		<option name="profile" value="HFP/AG" />
		<option name="profile" value="HFP/HF" />
		<option name="profile" value="HID/HOS" />
		<option name="profile" value="HOGP" />
		<option name="profile" value="L2CAP/COS" />
		<option name="profile" value="L2CAP/EXF" />
		<option name="profile" value="L2CAP/LE" />
		<option name="profile" value="MAP" />
		<option name="profile" value="OPP" />

android/pandora/server/configs/PtsBotTestMts.xml

+2 −0

Original line number	Diff line number	Diff line
		@@ -38,9 +38,11 @@
		<option name="profile" value="GAP" />
		<option name="profile" value="GATT" />
		<option name="profile" value="HFP/AG" />
		<option name="profile" value="HFP/HF" />
		<option name="profile" value="HID/HOS" />
		<option name="profile" value="HOGP" />
		<option name="profile" value="L2CAP/COS" />
		<option name="profile" value="L2CAP/EXF" />
		<option name="profile" value="L2CAP/LE" />
		<option name="profile" value="MAP" />
		<option name="profile" value="OPP" />