HFP/HF PTS-Bot tests (da44b75e) · Commits · e / os / android_packages_modules_Bluetooth

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/server/configs/PtsBotTest.xml

+1 −0

Original line number	Diff line number	Diff line
		@@ -38,6 +38,7 @@
		<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" />

android/pandora/server/configs/PtsBotTestMts.xml

+1 −0

Original line number	Diff line number	Diff line
		@@ -38,6 +38,7 @@
		<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" />

android/pandora/server/configs/pts_bot_tests_config.json

+18 −0

Original line number	Diff line number	Diff line
		@@ -737,6 +737,21 @@
		"HFP/AG/HFI/BI-03-I",
		"HFP/AG/OCN/BV-01-I",
		"HFP/AG/SLC/BV-04-C",
		"HFP/HF/OCM/BV-01-I",
		"HFP/HF/OCM/BV-02-I",
		"HFP/HF/OCL/BV-01-I",
		"HFP/HF/OCL/BV-02-I",
		"HFP/HF/TWC/BV-02-I",
		"HFP/HF/TWC/BV-03-I",
		"HFP/HF/ENO/BV-01-I",
		"HFP/HF/VRD/BV-01-I",
		"HFP/HF/NUM/BV-01-I",
		"HFP/HF/NUM/BI-01-I",
		"HFP/HF/ACC/BV-01-I",
		"HFP/HF/ACC/BV-02-I",
		"HFP/HF/ECC/BV-01-I",
		"HFP/HF/ECC/BV-02-I",
		"HFP/HF/ECS/BV-01-I",
		"HID/HOS/HCR/BV-01-I",
		"L2CAP/COS/CED/BI-02-C",
		"L2CAP/COS/CED/BV-01-C",
		@@ -1411,8 +1426,11 @@
		"TSPC_HFP_3_14": true,
		"TSPC_HFP_3_15": true,
		"TSPC_HFP_3_17": true,
		"TSPC_HFP_3_18a": true,
		"TSPC_HFP_3_18c": true,
		"TSPC_HFP_3_20": true,
		"TSPC_HFP_3_21a": true,
		"TSPC_HFP_3_21b": true,
		"TSPC_HFP_3_23": true,
		"TSPC_HFP_3_24": true,
		"TSPC_HFP_3_26": true,

android/pandora/server/src/com/android/pandora/HfpHandsfree.kt

0 → 100644

+190 −0

Original line number	Diff line number	Diff line
		/*
		* Copyright (C) 2022 The Android Open Source Project
		*
		* Licensed under the Apache License, Version 2.0 (the "License");
		* you may not use this file except in compliance with the License.
		* You may obtain a copy of the License at
		*
		* http://www.apache.org/licenses/LICENSE-2.0
		*
		* Unless required by applicable law or agreed to in writing, software
		* distributed under the License is distributed on an "AS IS" BASIS,
		* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
		* See the License for the specific language governing permissions and
		* limitations under the License.
		*/

		package com.android.pandora

		import android.annotation.SuppressLint
		import android.bluetooth.BluetoothDevice
		import android.bluetooth.BluetoothHeadset
		import android.bluetooth.BluetoothHeadsetClient
		import android.bluetooth.BluetoothManager
		import android.bluetooth.BluetoothProfile
		import android.content.Context
		import android.content.Intent
		import android.content.IntentFilter
		import android.net.Uri
		import android.os.Bundle
		import android.os.IBinder
		import android.provider.CallLog
		import android.telecom.Call
		import android.telecom.CallAudioState
		import android.telecom.InCallService
		import android.telecom.TelecomManager
		import android.telecom.VideoProfile
		import com.google.protobuf.Empty
		import io.grpc.stub.StreamObserver
		import kotlinx.coroutines.CoroutineScope
		import kotlinx.coroutines.Dispatchers
		import kotlinx.coroutines.cancel
		import kotlinx.coroutines.flow.Flow
		import kotlinx.coroutines.flow.SharingStarted
		import kotlinx.coroutines.flow.shareIn
		import pandora.HFPGrpc.HFPImplBase
		import pandora.HfpProto.*

		private const val TAG = "PandoraHfpHandsfree"

		@kotlinx.coroutines.ExperimentalCoroutinesApi
		class HfpHandsfree(val context: Context) : HFPImplBase() {
		private val scope: CoroutineScope = CoroutineScope(Dispatchers.Default)
		private val flow: Flow<Intent>

		private val bluetoothManager = context.getSystemService(BluetoothManager::class.java)!!
		private val telecomManager = context.getSystemService(TelecomManager::class.java)!!
		private val bluetoothAdapter = bluetoothManager.adapter

		private val bluetoothHfpClient = getProfileProxy<BluetoothHeadsetClient>(context, BluetoothProfile.HEADSET_CLIENT)

		companion object {
		@SuppressLint("StaticFieldLeak") private lateinit var inCallService: InCallService
		}

		init {
		val intentFilter = IntentFilter(BluetoothDevice.ACTION_BOND_STATE_CHANGED)
		flow = intentFlow(context, intentFilter).shareIn(scope, SharingStarted.Eagerly)
		}

		fun deinit() {
		bluetoothAdapter.closeProfileProxy(BluetoothProfile.HEADSET_CLIENT, bluetoothHfpClient)
		scope.cancel()
		}

		override fun answerCallAsHandsfree(
		request: AnswerCallAsHandsfreeRequest,
		responseObserver: StreamObserver<AnswerCallAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.acceptCall(request.connection.toBluetoothDevice(bluetoothAdapter), BluetoothHeadsetClient.CALL_ACCEPT_NONE)
		AnswerCallAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun endCallAsHandsfree(
		request: EndCallAsHandsfreeRequest,
		responseObserver: StreamObserver<EndCallAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		for (call in bluetoothHfpClient.getCurrentCalls(request.connection.toBluetoothDevice(bluetoothAdapter))) {
		bluetoothHfpClient.terminateCall(request.connection.toBluetoothDevice(bluetoothAdapter), call)
		}
		EndCallAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun declineCallAsHandsfree(
		request: DeclineCallAsHandsfreeRequest,
		responseObserver: StreamObserver<DeclineCallAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.rejectCall(request.connection.toBluetoothDevice(bluetoothAdapter))
		DeclineCallAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun connectToAudioAsHandsfree(
		request: ConnectToAudioAsHandsfreeRequest,
		responseObserver: StreamObserver<ConnectToAudioAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.connectAudio(request.connection.toBluetoothDevice(bluetoothAdapter))
		ConnectToAudioAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun disconnectFromAudioAsHandsfree(
		request: DisconnectFromAudioAsHandsfreeRequest,
		responseObserver: StreamObserver<DisconnectFromAudioAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.disconnectAudio(request.connection.toBluetoothDevice(bluetoothAdapter))
		DisconnectFromAudioAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun makeCallAsHandsfree(
		request: MakeCallAsHandsfreeRequest,
		responseObserver: StreamObserver<MakeCallAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.dial(request.connection.toBluetoothDevice(bluetoothAdapter), request.number)
		MakeCallAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun callTransferAsHandsfree(
		request: CallTransferAsHandsfreeRequest,
		responseObserver: StreamObserver<CallTransferAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.explicitCallTransfer(request.connection.toBluetoothDevice(bluetoothAdapter))
		CallTransferAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun enableSlcAsHandsfree(
		request: EnableSlcAsHandsfreeRequest,
		responseObserver: StreamObserver<Empty>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.setConnectionPolicy(request.connection.toBluetoothDevice(bluetoothAdapter), BluetoothProfile.CONNECTION_POLICY_ALLOWED)
		Empty.getDefaultInstance()
		}
		}

		override fun disableSlcAsHandsfree(
		request: DisableSlcAsHandsfreeRequest,
		responseObserver: StreamObserver<Empty>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.setConnectionPolicy(request.connection.toBluetoothDevice(bluetoothAdapter), BluetoothProfile.CONNECTION_POLICY_FORBIDDEN)
		Empty.getDefaultInstance()
		}
		}

		override fun setVoiceRecognitionAsHandsfree(
		request: SetVoiceRecognitionAsHandsfreeRequest,
		responseObserver: StreamObserver<SetVoiceRecognitionAsHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		if (request.enabled) {
		bluetoothHfpClient.startVoiceRecognition(request.connection.toBluetoothDevice(bluetoothAdapter))
		} else {
		bluetoothHfpClient.stopVoiceRecognition(request.connection.toBluetoothDevice(bluetoothAdapter))
		}
		SetVoiceRecognitionAsHandsfreeResponse.getDefaultInstance()
		}
		}

		override fun sendDtmfFromHandsfree(
		request: SendDtmfFromHandsfreeRequest,
		responseObserver: StreamObserver<SendDtmfFromHandsfreeResponse>
		) {
		grpcUnary(scope, responseObserver) {
		bluetoothHfpClient.sendDTMF(request.connection.toBluetoothDevice(bluetoothAdapter), request.code.toByte())
		SendDtmfFromHandsfreeResponse.getDefaultInstance()
		}
		}
		}