diff options
Diffstat (limited to 'tests/hfp_test.py')
| -rw-r--r-- | tests/hfp_test.py | 470 |
1 files changed, 444 insertions, 26 deletions
diff --git a/tests/hfp_test.py b/tests/hfp_test.py index dc28180..83b0d35 100644 --- a/tests/hfp_test.py +++ b/tests/hfp_test.py @@ -19,8 +19,9 @@ import asyncio import logging import os import pytest +import pytest_asyncio -from typing import Tuple +from typing import Tuple, Optional from .test_utils import TwoDevices from bumble import core @@ -36,9 +37,93 @@ logger = logging.getLogger(__name__) # ----------------------------------------------------------------------------- +def _default_hf_configuration() -> hfp.HfConfiguration: + return hfp.HfConfiguration( + supported_hf_features=[ + hfp.HfFeature.CODEC_NEGOTIATION, + hfp.HfFeature.ESCO_S4_SETTINGS_SUPPORTED, + hfp.HfFeature.HF_INDICATORS, + hfp.HfFeature.ENHANCED_CALL_STATUS, + hfp.HfFeature.THREE_WAY_CALLING, + hfp.HfFeature.CLI_PRESENTATION_CAPABILITY, + ], + supported_hf_indicators=[ + hfp.HfIndicator.ENHANCED_SAFETY, + hfp.HfIndicator.BATTERY_LEVEL, + ], + supported_audio_codecs=[ + hfp.AudioCodec.CVSD, + hfp.AudioCodec.MSBC, + ], + ) + + +# ----------------------------------------------------------------------------- +def _default_hf_sdp_features() -> hfp.HfSdpFeature: + return ( + hfp.HfSdpFeature.WIDE_BAND + | hfp.HfSdpFeature.THREE_WAY_CALLING + | hfp.HfSdpFeature.CLI_PRESENTATION_CAPABILITY + ) + + +# ----------------------------------------------------------------------------- +def _default_ag_configuration() -> hfp.AgConfiguration: + return hfp.AgConfiguration( + supported_ag_features=[ + hfp.AgFeature.HF_INDICATORS, + hfp.AgFeature.IN_BAND_RING_TONE_CAPABILITY, + hfp.AgFeature.REJECT_CALL, + hfp.AgFeature.CODEC_NEGOTIATION, + hfp.AgFeature.ESCO_S4_SETTINGS_SUPPORTED, + hfp.AgFeature.ENHANCED_CALL_STATUS, + hfp.AgFeature.THREE_WAY_CALLING, + ], + supported_ag_indicators=[ + hfp.AgIndicatorState.call(), + hfp.AgIndicatorState.service(), + hfp.AgIndicatorState.callsetup(), + hfp.AgIndicatorState.callsetup(), + hfp.AgIndicatorState.signal(), + hfp.AgIndicatorState.roam(), + hfp.AgIndicatorState.battchg(), + ], + supported_hf_indicators=[ + hfp.HfIndicator.ENHANCED_SAFETY, + hfp.HfIndicator.BATTERY_LEVEL, + ], + supported_ag_call_hold_operations=[ + hfp.CallHoldOperation.ADD_HELD_CALL, + hfp.CallHoldOperation.HOLD_ALL_ACTIVE_CALLS, + hfp.CallHoldOperation.HOLD_ALL_CALLS_EXCEPT, + hfp.CallHoldOperation.RELEASE_ALL_ACTIVE_CALLS, + hfp.CallHoldOperation.RELEASE_ALL_HELD_CALLS, + hfp.CallHoldOperation.RELEASE_SPECIFIC_CALL, + hfp.CallHoldOperation.CONNECT_TWO_CALLS, + ], + supported_audio_codecs=[hfp.AudioCodec.CVSD, hfp.AudioCodec.MSBC], + ) + + +# ----------------------------------------------------------------------------- +def _default_ag_sdp_features() -> hfp.AgSdpFeature: + return ( + hfp.AgSdpFeature.WIDE_BAND + | hfp.AgSdpFeature.IN_BAND_RING_TONE_CAPABILITY + | hfp.AgSdpFeature.THREE_WAY_CALLING + ) + + +# ----------------------------------------------------------------------------- async def make_hfp_connections( - hf_config: hfp.Configuration, -) -> Tuple[hfp.HfProtocol, hfp.HfpProtocol]: + hf_config: Optional[hfp.HfConfiguration] = None, + ag_config: Optional[hfp.AgConfiguration] = None, +): + if not hf_config: + hf_config = _default_hf_configuration() + if not ag_config: + ag_config = _default_ag_configuration() + # Setup devices devices = TwoDevices() await devices.setup_connection() @@ -55,38 +140,371 @@ async def make_hfp_connections( # Setup HFP connection hf = hfp.HfProtocol(client_dlc, hf_config) - ag = hfp.HfpProtocol(server_dlc) - return hf, ag + ag = hfp.AgProtocol(server_dlc, ag_config) + + await hf.initiate_slc() + return (hf, ag) # ----------------------------------------------------------------------------- +@pytest_asyncio.fixture +async def hfp_connections(): + hf, ag = await make_hfp_connections() + hf_loop_task = asyncio.create_task(hf.run()) + + try: + yield (hf, ag) + finally: + # Close the coroutine. + hf.unsolicited_queue.put_nowait(None) + await hf_loop_task +# ----------------------------------------------------------------------------- @pytest.mark.asyncio -async def test_slc(): - hf_config = hfp.Configuration( - supported_hf_features=[], supported_hf_indicators=[], supported_audio_codecs=[] - ) - hf, ag = await make_hfp_connections(hf_config) - - async def ag_loop(): - while line := await ag.next_line(): - if line.startswith('AT+BRSF'): - ag.send_response_line('+BRSF: 0') - elif line.startswith('AT+CIND=?'): - ag.send_response_line( - '+CIND: ("call",(0,1)),("callsetup",(0-3)),("service",(0-1)),' - '("signal",(0-5)),("roam",(0,1)),("battchg",(0-5)),' - '("callheld",(0-2))' +async def test_slc_with_minimal_features(): + hf, ag = await make_hfp_connections( + hfp.HfConfiguration( + supported_audio_codecs=[], + supported_hf_features=[], + supported_hf_indicators=[], + ), + hfp.AgConfiguration( + supported_ag_call_hold_operations=[], + supported_ag_features=[], + supported_ag_indicators=[ + hfp.AgIndicatorState( + indicator=hfp.AgIndicator.CALL, + supported_values={0, 1}, + current_status=0, ) - elif line.startswith('AT+CIND?'): - ag.send_response_line('+CIND: 0,0,1,4,1,5,0') - ag.send_response_line('OK') + ], + supported_hf_indicators=[], + supported_audio_codecs=[], + ), + ) - ag_task = asyncio.create_task(ag_loop()) + assert hf.supported_ag_features == ag.supported_ag_features + assert hf.supported_hf_features == ag.supported_hf_features + assert hf.supported_ag_call_hold_operations == ag.supported_ag_call_hold_operations + for a, b in zip(hf.ag_indicators, ag.ag_indicators): + assert a.indicator == b.indicator + assert a.current_status == b.current_status - await hf.initiate_slc() - ag_task.cancel() + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_slc(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + + assert hf.supported_ag_features == ag.supported_ag_features + assert hf.supported_hf_features == ag.supported_hf_features + assert hf.supported_ag_call_hold_operations == ag.supported_ag_call_hold_operations + for a, b in zip(hf.ag_indicators, ag.ag_indicators): + assert a.indicator == b.indicator + assert a.current_status == b.current_status + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_ag_indicator(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + + future = asyncio.get_running_loop().create_future() + hf.on('ag_indicator', future.set_result) + + ag.update_ag_indicator(hfp.AgIndicator.CALL, 1) + + indicator: hfp.AgIndicatorState = await future + assert indicator.current_status == 1 + assert indicator.indicator == hfp.AgIndicator.CALL + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_hf_indicator(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + + future = asyncio.get_running_loop().create_future() + ag.on('hf_indicator', future.set_result) + + await hf.execute_command('AT+BIEV=2,100') + + indicator: hfp.HfIndicatorState = await future + assert indicator.current_status == 100 + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_codec_negotiation( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + + futures = [ + asyncio.get_running_loop().create_future(), + asyncio.get_running_loop().create_future(), + ] + hf.on('codec_negotiation', futures[0].set_result) + ag.on('codec_negotiation', futures[1].set_result) + await ag.negotiate_codec(hfp.AudioCodec.MSBC) + + assert await futures[0] == await futures[1] + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_dial(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + NUMBER = 'ATD123456789' + + future = asyncio.get_running_loop().create_future() + ag.on('dial', future.set_result) + await hf.execute_command(f'ATD{NUMBER}') + + number: str = await future + assert number == NUMBER + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_answer(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + + future = asyncio.get_running_loop().create_future() + ag.on('answer', lambda: future.set_result(None)) + await hf.answer_incoming_call() + + await future + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_reject_incoming_call( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + + future = asyncio.get_running_loop().create_future() + ag.on('hang_up', lambda: future.set_result(None)) + await hf.reject_incoming_call() + + await future + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_terminate_call(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + + future = asyncio.get_running_loop().create_future() + ag.on('hang_up', lambda: future.set_result(None)) + await hf.terminate_call() + + await future + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_query_calls_without_calls( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + + assert await hf.query_current_calls() == [] + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_query_calls_with_calls( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + ag.calls.append( + hfp.CallInfo( + index=1, + direction=hfp.CallInfoDirection.MOBILE_ORIGINATED_CALL, + status=hfp.CallInfoStatus.ACTIVE, + mode=hfp.CallInfoMode.VOICE, + multi_party=hfp.CallInfoMultiParty.NOT_IN_CONFERENCE, + number='123456789', + ) + ) + + assert await hf.query_current_calls() == ag.calls + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +@pytest.mark.parametrize( + "operation,", + ( + hfp.CallHoldOperation.RELEASE_ALL_HELD_CALLS, + hfp.CallHoldOperation.RELEASE_ALL_ACTIVE_CALLS, + hfp.CallHoldOperation.HOLD_ALL_ACTIVE_CALLS, + hfp.CallHoldOperation.ADD_HELD_CALL, + hfp.CallHoldOperation.CONNECT_TWO_CALLS, + ), +) +async def test_hold_call_without_call_index( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol], + operation: hfp.CallHoldOperation, +): + hf, ag = hfp_connections + call_hold_future = asyncio.get_running_loop().create_future() + ag.on("call_hold", lambda op, index: call_hold_future.set_result((op, index))) + + await hf.execute_command(f"AT+CHLD={operation.value}") + + assert (await call_hold_future) == (operation, None) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +@pytest.mark.parametrize( + "operation,", + ( + hfp.CallHoldOperation.RELEASE_SPECIFIC_CALL, + hfp.CallHoldOperation.HOLD_ALL_CALLS_EXCEPT, + ), +) +async def test_hold_call_with_call_index( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol], + operation: hfp.CallHoldOperation, +): + hf, ag = hfp_connections + call_hold_future = asyncio.get_running_loop().create_future() + ag.on("call_hold", lambda op, index: call_hold_future.set_result((op, index))) + ag.calls.append( + hfp.CallInfo( + index=1, + direction=hfp.CallInfoDirection.MOBILE_ORIGINATED_CALL, + status=hfp.CallInfoStatus.ACTIVE, + mode=hfp.CallInfoMode.VOICE, + multi_party=hfp.CallInfoMultiParty.NOT_IN_CONFERENCE, + number='123456789', + ) + ) + + await hf.execute_command(f"AT+CHLD={operation.value.replace('x', '1')}") + + assert (await call_hold_future) == (operation, 1) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_ring(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + ring_future = asyncio.get_running_loop().create_future() + hf.on("ring", lambda: ring_future.set_result(None)) + + ag.send_ring() + + await ring_future + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_speaker_volume(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + speaker_volume_future = asyncio.get_running_loop().create_future() + hf.on("speaker_volume", speaker_volume_future.set_result) + + ag.set_speaker_volume(10) + + assert await speaker_volume_future == 10 + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_microphone_volume( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + microphone_volume_future = asyncio.get_running_loop().create_future() + hf.on("microphone_volume", microphone_volume_future.set_result) + + ag.set_microphone_volume(10) + + assert await microphone_volume_future == 10 + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_cli_notification(hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol]): + hf, ag = hfp_connections + cli_notification_future = asyncio.get_running_loop().create_future() + hf.on("cli_notification", cli_notification_future.set_result) + + ag.send_cli_notification( + hfp.CallLineIdentification(number="\"123456789\"", type=129, alpha="\"Bumble\"") + ) + + assert await cli_notification_future == hfp.CallLineIdentification( + number="123456789", type=129, alpha="Bumble", subaddr="", satype=None + ) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_voice_recognition_from_hf( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + voice_recognition_future = asyncio.get_running_loop().create_future() + ag.on("voice_recognition", voice_recognition_future.set_result) + + await hf.execute_command("AT+BVRA=1") + + assert await voice_recognition_future == hfp.VoiceRecognitionState.ENABLE + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_voice_recognition_from_ag( + hfp_connections: Tuple[hfp.HfProtocol, hfp.AgProtocol] +): + hf, ag = hfp_connections + voice_recognition_future = asyncio.get_running_loop().create_future() + hf.on("voice_recognition", voice_recognition_future.set_result) + + ag.send_response("+BVRA: 1") + + assert await voice_recognition_future == hfp.VoiceRecognitionState.ENABLE + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_hf_sdp_record(): + devices = TwoDevices() + await devices.setup_connection() + + devices[0].sdp_service_records[1] = hfp.make_hf_sdp_records( + 1, 2, _default_hf_configuration(), hfp.ProfileVersion.V1_8 + ) + + assert await hfp.find_hf_sdp_record(devices.connections[1]) == ( + 2, + hfp.ProfileVersion.V1_8, + _default_hf_sdp_features(), + ) + + +# ----------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_ag_sdp_record(): + devices = TwoDevices() + await devices.setup_connection() + + devices[0].sdp_service_records[1] = hfp.make_ag_sdp_records( + 1, 2, _default_ag_configuration(), hfp.ProfileVersion.V1_8 + ) + + assert await hfp.find_ag_sdp_record(devices.connections[1]) == ( + 2, + hfp.ProfileVersion.V1_8, + _default_ag_sdp_features(), + ) # ----------------------------------------------------------------------------- |