auracaster/bumble_mirror
3
0
Fork 1
mirror of https://github.com/google/bumble.git synced 2026-04-16 00:25:31 +00:00
Code Issues Packages Projects Releases Wiki Activity
Files
42711d3d31f2c808c653c05d096bce1e85337781
bumble_mirror/tests/hfp_test.py
Gilles Boccon-Gibod 7aba36302a use isort when formatting
2025-08-21 16:38:58 -07:00

609 lines
20 KiB
Python
Raw Blame History

# Copyright 2021-2022 Google LLC
#
# 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
#
# https://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.
# -----------------------------------------------------------------------------
# Imports
# -----------------------------------------------------------------------------
import asyncio
import logging
import os
from typing import Optional
import pytest
import pytest_asyncio
from bumble import core, hci, hfp, rfcomm
from .test_utils import TwoDevices
# -----------------------------------------------------------------------------
# Logging
# -----------------------------------------------------------------------------
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_SPEECH
| 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_SPEECH
| hfp.AgSdpFeature.IN_BAND_RING_TONE_CAPABILITY
| hfp.AgSdpFeature.THREE_WAY_CALLING
)
# -----------------------------------------------------------------------------
async def make_hfp_connections(
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()
# Setup RFCOMM channel
wait_dlc = asyncio.get_running_loop().create_future()
rfcomm_channel = rfcomm.Server(devices.devices[0]).listen(wait_dlc.set_result)
assert devices.connections[0]
assert devices.connections[1]
client_mux = await rfcomm.Client(devices.connections[1]).start()
client_dlc = await client_mux.open_dlc(rfcomm_channel)
server_dlc = await wait_dlc
# Setup HFP connection
hf = hfp.HfProtocol(client_dlc, hf_config)
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_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,
)
],
supported_hf_indicators=[],
supported_audio_codecs=[],
),
)
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_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(),
)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_sco_setup():
devices = TwoDevices()
# Enable Classic connections
devices[0].classic_enabled = True
devices[1].classic_enabled = True
# Start
await devices[0].power_on()
await devices[1].power_on()
connections = await asyncio.gather(
devices[0].connect(
devices[1].public_address, transport=core.PhysicalTransport.BR_EDR
),
devices[1].accept(devices[0].public_address),
)
def on_sco_request(_connection, _link_type: int):
connections[1].abort_on(
'disconnection',
devices[1].send_command(
hci.HCI_Enhanced_Accept_Synchronous_Connection_Request_Command(
bd_addr=connections[1].peer_address,
**hfp.ESCO_PARAMETERS[
hfp.DefaultCodecParameters.ESCO_CVSD_S1
].asdict(),
)
),
)
devices[1].on('sco_request', on_sco_request)
sco_connection_futures = [
asyncio.get_running_loop().create_future(),
asyncio.get_running_loop().create_future(),
]
for device, future in zip(devices, sco_connection_futures):
device.on('sco_connection', future.set_result)
await devices[0].send_command(
hci.HCI_Enhanced_Setup_Synchronous_Connection_Command(
connection_handle=connections[0].handle,
**hfp.ESCO_PARAMETERS[hfp.DefaultCodecParameters.ESCO_CVSD_S1].asdict(),
)
)
sco_connections = await asyncio.gather(*sco_connection_futures)
sco_disconnection_futures = [
asyncio.get_running_loop().create_future(),
asyncio.get_running_loop().create_future(),
]
for future, sco_connection in zip(sco_disconnection_futures, sco_connections):
sco_connection.on('disconnection', future.set_result)
await sco_connections[0].disconnect()
await asyncio.gather(*sco_disconnection_futures)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_hf_batched_response(
hfp_connections: tuple[hfp.HfProtocol, hfp.AgProtocol],
):
hf, ag = hfp_connections
ag.dlc.write(b'\r\n+BIND: (1,2)\r\n\r\nOK\r\n')
await hf.execute_command("AT+BIND=?", response_type=hfp.AtResponseType.SINGLE)
# -----------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_ag_batched_commands(
hfp_connections: tuple[hfp.HfProtocol, hfp.AgProtocol],
):
hf, ag = hfp_connections
answer_future = asyncio.get_running_loop().create_future()
ag.on('answer', lambda: answer_future.set_result(None))
hang_up_future = asyncio.get_running_loop().create_future()
ag.on('hang_up', lambda: hang_up_future.set_result(None))
hf.dlc.write(b'ATA\rAT+CHUP\r')
await answer_future
await hang_up_future
# -----------------------------------------------------------------------------
async def run():
await test_slc()
# -----------------------------------------------------------------------------
if __name__ == '__main__':
logging.basicConfig(level=os.environ.get('BUMBLE_LOGLEVEL', 'INFO').upper())
asyncio.run(run())
Reference in New Issue View Git Blame Copy Permalink