Donate to e Foundation | Murena handsets with /e/OS | Own a part of Murena! Learn more

Commit ad8a5c0f authored by David Duarte's avatar David Duarte Committed by Gerrit Code Review
Browse files

Merge changes from topic "bbt_config_file" into main

* changes:
  pandora: change the identity_address_type accordingly to the test
  avatar: add hfpProtocol class into test file
  BumbleBluetoothTest: provide Bumble configuration file
parents 1bf2c51b 72c2fb4c
Loading
Loading
Loading
Loading
+18 −0
Original line number Diff line number Diff line
@@ -86,6 +86,21 @@ class AshaTest(base_test.BaseTestClass): # type: ignore[misc]

    @avatar.asynchronous
    async def setup_test(self) -> None:
        # TODO(b/296927588): Test should pass with a random identity address
        # We have to update the config before the reset otherwise it'll be overridden
        random_identity_address_type_list = [
            "test_auto_connection(1,1)",
            "test_advertising_advertisement_data(1,1)",
            "test_advertising_scan_response",
            "test_auto_connection_dual_device(1,1,0)",
            "test_auto_connection_dual_device(1,1,1)",
        ]
        if self.current_test_info.name in random_identity_address_type_list:
            self.ref_left._bumble.config.update({'server': {'identity_address_type': 'random'}})
            self.ref_right._bumble.config.update({'server': {'identity_address_type': 'random'}})
        else:
            self.ref_left._bumble.config.update({'server': {'identity_address_type': 'public'}})
            self.ref_right._bumble.config.update({'server': {'identity_address_type': 'public'}})
        await asyncio.gather(self.dut.reset(), self.ref_left.reset(), self.ref_right.reset())

        # ASHA hearing aid's IO capability is NO_OUTPUT_NO_INPUT
@@ -683,6 +698,9 @@ class AshaTest(base_test.BaseTestClass): # type: ignore[misc]
           2. The disconnected peripheral starts sending ASHA advertisements.
           3. Verify that DUT auto-connects to the peripheral.
        """
        # This tests need to be reactivated ASAP
        if ref_address_type == PUBLIC:
            raise signals.TestSkip('TODO: b/296927588')

        advertisement_left = await self.ref_advertise_asha(
            ref_device=self.ref_left, ref_address_type=ref_address_type, ear=Ear.LEFT
+58 −2
Original line number Diff line number Diff line
@@ -18,13 +18,14 @@ import logging

from avatar import BumblePandoraDevice, PandoraDevice, PandoraDevices
from avatar.pandora_server import AndroidPandoraServer
from bumble.colors import color
from bumble.core import (
    BT_GENERIC_AUDIO_SERVICE,
    BT_HANDSFREE_AUDIO_GATEWAY_SERVICE,
    BT_L2CAP_PROTOCOL_ID,
    BT_RFCOMM_PROTOCOL_ID,
)
from bumble.hfp import HfpProtocol
from bumble import rfcomm
from bumble.rfcomm import DLC, Server as RfcommServer
from bumble.sdp import (
    SDP_BLUETOOTH_PROFILE_DESCRIPTOR_LIST_ATTRIBUTE_ID,
@@ -34,6 +35,7 @@ from bumble.sdp import (
    DataElement,
    ServiceAttribute,
)
import collections
from mobly import base_test, test_runner
from mobly.asserts import assert_equal  # type: ignore
from mobly.asserts import assert_in  # type: ignore
@@ -41,7 +43,7 @@ from mobly.asserts import assert_not_equal # type: ignore
from mobly.asserts import assert_not_in  # type: ignore
from pandora.host_pb2 import Connection as PandoraConnection
from pandora.security_pb2 import LEVEL2
from typing import Dict, List, Optional, Tuple
from typing import Dict, List, Optional, Tuple, Union

SDP_PROFILE_SUPPORTED_FEATURES_ID = 0x0311

@@ -58,6 +60,58 @@ PROPERTY_HF_INDICATOR_ENHANCED_DRIVER_SAFETY = 'bluetooth.headset_client.indicat
HFP_VERSION_1_7 = 0x0107


# Stub for Audio Gateway implementation
# TODO: b/296471045
logger = logging.getLogger(__name__)
class HfpProtocol:
    dlc: rfcomm.DLC
    buffer: str
    lines: collections.deque
    lines_available: asyncio.Event

    def __init__(self, dlc: rfcomm.DLC) -> None:
        self.dlc = dlc
        self.buffer = ''
        self.lines = collections.deque()
        self.lines_available = asyncio.Event()

        dlc.sink = self.feed

    def feed(self, data: Union[bytes, str]) -> None:
        # Convert the data to a string if needed
        if isinstance(data, bytes):
            data = data.decode('utf-8')

        logger.debug(f'<<< Data received: {data}')

        # Add to the buffer and look for lines
        self.buffer += data
        while (separator := self.buffer.find('\r')) >= 0:
            line = self.buffer[:separator].strip()
            self.buffer = self.buffer[separator + 1:]
            if len(line) > 0:
                self.on_line(line)

    def on_line(self, line: str) -> None:
        self.lines.append(line)
        self.lines_available.set()

    def send_command_line(self, line: str) -> None:
        logger.debug(color(f'>>> {line}', 'yellow'))
        self.dlc.write(line + '\r')

    def send_response_line(self, line: str) -> None:
        logger.debug(color(f'>>> {line}', 'yellow'))
        self.dlc.write('\r\n' + line + '\r\n')

    async def next_line(self) -> str:
        await self.lines_available.wait()
        line = self.lines.popleft()
        if not self.lines:
            self.lines_available.clear()
        logger.debug(color(f'<<< {line}', 'green'))
        return line

class HfpClientTest(base_test.BaseTestClass):  # type: ignore[misc]
    devices: Optional[PandoraDevices] = None

@@ -98,6 +152,7 @@ class HfpClientTest(base_test.BaseTestClass): # type: ignore[misc]

    @avatar.asynchronous
    async def setup_test(self) -> None:
        self.ref._bumble.config.update({'server': {'identity_address_type': 'public'}})
        await asyncio.gather(self.dut.reset(), self.ref.reset())

    # TODO(b/286338264): Moving connecting and bonding methods to a shared util scripts
@@ -306,6 +361,7 @@ class HfpAgServer:
        self.send_response_line('OK')



if __name__ == '__main__':
    logging.basicConfig(level=logging.DEBUG)
    test_runner.main()  # type: ignore
+1 −0
Original line number Diff line number Diff line
@@ -50,6 +50,7 @@ java_test_host {

    data: [
        ":BumbleBluetoothTestsApp",
        "bumble_config.json",
    ],
    data_native_bins: [
        "bumble_pandora_server",
+1 −1
Original line number Diff line number Diff line
@@ -16,7 +16,7 @@
      <option name="python-virtualenv" value="true"/>
      <option name="host-setup-command" value="adb -s $SERIAL reverse tcp:7999 tcp:7999" />
      <option name="host-background-command"
        value="$EXTRA_FILE(host_testcases)/BumbleBluetoothTests/bumble_pandora_server"/>
        value="$EXTRA_FILE(host_testcases)/BumbleBluetoothTests/bumble_pandora_server --config $EXTRA_FILE(host_testcases)/BumbleBluetoothTests/bumble_config.json"/>
      <option name="host-teardown-command" value="adb -s $SERIAL reverse --remove tcp:7999" />
    </target_preparer>

+1 −0
Original line number Diff line number Diff line
{}
 No newline at end of file