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

Commit ec134ad7 authored by Duo Ho's avatar Duo Ho
Browse files

Add asha dual device test - test_disconnect_acceptor_dual_device into asha_test.py 

Test: avatar run --mobly-std-log --include-filter 'ASHATest#test_disconnect_acceptor_dual_device(1,1,0)'

Change-Id: I02617b8ceb6ae7e17651fb0815fd56983dc8c43d
parent 5a00b850

android/pandora/test/asha_test.py

+90 −24
Original line number Diff line number Diff line
@@ -14,6 +14,7 @@ 


import asyncio

import avatar

import grpc

import logging

import time
@@ -35,16 +36,24 @@ CAPABILITY: int = 0x0 
COMPLETE_LOCAL_NAME: str = "Bumble"





class Device:

    """Reference devices type"""



    LEFT = 0

    RIGHT = 1





class ASHATest(base_test.BaseTestClass):  # type: ignore[misc]

    devices: Optional[PandoraDevices] = None



    # pandora devices.

    dut: PandoraDevice

    ref: PandoraDevice

    ref_left: PandoraDevice

    ref_right: PandoraDevice



    def setup_class(self) -> None:

        self.devices = PandoraDevices(self)

        self.dut, self.ref, *_ = self.devices

        self.dut, self.ref_left, self.ref_right, *_ = self.devices



    def teardown_class(self) -> None:

        if self.devices:
@@ -52,24 +61,29 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 


    @avatar.asynchronous

    async def setup_test(self) -> None:

        await asyncio.gather(self.dut.reset(), self.ref.reset())

        await asyncio.gather(self.dut.reset(), self.ref_left.reset(), self.ref_right.reset())



        if isinstance(self.dut, BumblePandoraDevice):

            raise signals.TestSkip('DUT Bumble does not support Asha source')

        if not isinstance(self.ref, BumblePandoraDevice):

        if not isinstance(self.ref_left, BumblePandoraDevice):

            raise signals.TestSkip('Test require Bumble as reference device(s)')

        if not isinstance(self.ref_right, BumblePandoraDevice):

            raise signals.TestSkip('Test require Bumble as reference device(s)')



        # ASHA hearing aid's IO capability is NO_OUTPUT_NO_INPUT

        setattr(self.ref.device, "io_capability", PairingDelegate.NO_OUTPUT_NO_INPUT)

        setattr(self.ref_left.device, "io_capability", PairingDelegate.NO_OUTPUT_NO_INPUT)

        setattr(self.ref_right.device, "io_capability", PairingDelegate.NO_OUTPUT_NO_INPUT)



    def ref_advertise_asha(self, ref_address_type: OwnAddressType) -> Stream[AdvertiseResponse]:

    def ref_advertise_asha(

        self, ref_device: PandoraDevice, ref_address_type: OwnAddressType

    ) -> Stream[AdvertiseResponse]:

        """

        Ref device starts to advertise

        :return: Ref device's advertise response

        """

        # Ref starts advertising with ASHA service data

        self.ref.asha.Register(capability=CAPABILITY, hisyncid=HISYCNID)

        return self.ref.host.Advertise(

        ref_device.asha.Register(capability=CAPABILITY, hisyncid=HISYCNID)

        return ref_device.host.Advertise(

            legacy=True,

            connectable=True,

            data=DataTypes(
@@ -106,6 +120,14 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        advertisement.cancel()

        return dut_ref, ref_dut



    def is_device_connected(self, device: PandoraDevice, connection: Connection, timeout: float) -> bool:

        try:

            device.host.WaitDisconnection(connection=connection, timeout=timeout)

            return False

        except grpc.RpcError as e:

            assert e.code() == grpc.StatusCode.DEADLINE_EXCEEDED  # type: ignore

            return True



    def test_advertising_advertisement_data(self) -> None:

        """

        Ref starts ASHA advertisements with service data in advertisement data.
@@ -115,7 +137,7 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        protocol_version = 0x01

        truncated_hisyncid = HISYCNID[:4]



        advertisement = self.ref_advertise_asha(ref_address_type=RANDOM)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=RANDOM)



        # DUT starts a service discovery

        scan_result = self.dut_scan_for_asha(dut_address_type=RANDOM)
@@ -143,10 +165,10 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        protocol_version = 0x01

        truncated_hisyncid = HISYCNID[:4]



        self.ref.asha.Register(capability=CAPABILITY, hisyncid=HISYCNID)

        self.ref_left.asha.Register(capability=CAPABILITY, hisyncid=HISYCNID)



        # advertise with ASHA service data in scan response

        advertisement = self.ref.host.Advertise(

        advertisement = self.ref_left.host.Advertise(

            legacy=True,

            scan_response_data=DataTypes(

                complete_local_name=COMPLETE_LOCAL_NAME,
@@ -183,7 +205,7 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        DUT initiates connection to Ref.

        Verify that DUT and Ref are bonded and connected.

        """

        advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)



        ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)
@@ -211,8 +233,8 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        """

        raise signals.TestSkip("TODO: update rootcanal to retry")



        advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

        ref = self.dut_scan_for_asha(dut_address_type=ref_address_type)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

        ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)



        dut_ref, ref_dut = self.dut_connect_to_ref(advertisement, ref, dut_address_type)
@@ -220,13 +242,13 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 


        assert_equal(secure.WhichOneof("result"), "success")

        self.dut.host.Disconnect(dut_ref)

        self.ref.host.WaitDisconnection(ref_dut)

        self.ref_left.host.WaitDisconnection(ref_dut)



        # delete the bond

        if dut_address_type == OwnAddressType.PUBLIC:

            self.dut.security_storage.DeleteBond(public=self.ref.address)

            self.dut.security_storage.DeleteBond(public=self.ref_left.address)

        else:

            self.dut.security_storage.DeleteBond(random=self.ref.random_address)

            self.dut.security_storage.DeleteBond(random=self.ref_left.random_address)



        # DUT connect to REF again

        dut_ref = (self.dut.host.ConnectLE(own_address_type=dut_address_type, **ref.address_asdict())).connection
@@ -250,7 +272,7 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        DUT initiates connection to Ref.

        Verify that DUT and Ref are connected.

        """

        advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

        ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)

        dut_ref, ref_dut = self.dut_connect_to_ref(advertisement, ref, dut_address_type)

        assert dut_ref
@@ -269,7 +291,7 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        DUT initiates disconnection to Ref.

        Verify that DUT and Ref are disconnected.

        """

        advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

        ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)

        dut_ref, _ = self.dut_connect_to_ref(advertisement, ref, dut_address_type)
@@ -288,12 +310,12 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        Ref initiates disconnection to DUT (typically when put back in its box).

        Verify that Ref is disconnected.

        """

        advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

        ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)

        dut_ref, ref_dut = self.dut_connect_to_ref(advertisement, ref, dut_address_type)

        assert dut_ref

        assert ref_dut

        self.ref.host.Disconnect(connection=ref_dut)

        self.ref_left.host.Disconnect(connection=ref_dut)



    @avatar.parameterized(

        (RANDOM, RANDOM, 0),
@@ -315,7 +337,7 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        """



        def connect_and_disconnect() -> None:

            advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

            advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

            ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)

            dut_ref, _ = self.dut_connect_to_ref(advertisement, ref, dut_address_type)

            self.dut.host.Disconnect(connection=dut_ref)
@@ -339,7 +361,7 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        Ref starts sending ASHA advertisements.

        Verify that DUT auto-connects to Ref.

        """

        advertisement = self.ref_advertise_asha(ref_address_type=ref_address_type)

        advertisement = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

        ref = self.dut_scan_for_asha(dut_address_type=dut_address_type)



        # manually connect and not cancel advertisement
@@ -352,12 +374,56 @@ class ASHATest(base_test.BaseTestClass): # type: ignore[misc] 
        secure = self.dut.security.Secure(connection=dut_ref, le=LE_LEVEL3)

        assert_equal(secure.WhichOneof("result"), "success")



        self.ref.host.Disconnect(connection=ref_dut)

        self.ref_left.host.Disconnect(connection=ref_dut)



        ref_dut = next(advertisement).connection

        advertisement.cancel()

        assert ref_dut



    @avatar.parameterized(

        (RANDOM, RANDOM, Device.LEFT),

        (RANDOM, PUBLIC, Device.RIGHT),

    )  # type: ignore[misc]

    def test_disconnect_acceptor_dual_device(

        self,

        dut_address_type: OwnAddressType,

        ref_address_type: OwnAddressType,

        disconnect_device: Device,

    ) -> None:

        """

        Prerequisites: DUT and Ref are connected and bonded.

        Description:

           1. One peripheral of Ref initiates disconnection to DUT.

           2. Verify that it is disconnected and that the other peripheral is still connected.

        """



        advertisement_left = self.ref_advertise_asha(ref_device=self.ref_left, ref_address_type=ref_address_type)

        ref_left = self.dut_scan_for_asha(dut_address_type=dut_address_type)

        dut_ref_left, ref_left_dut = self.dut_connect_to_ref(

            advertisement=advertisement_left, ref=ref_left, dut_address_type=dut_address_type

        )

        advertisement_left.cancel()

        assert dut_ref_left

        assert ref_left_dut



        advertisement_right = self.ref_advertise_asha(ref_device=self.ref_right, ref_address_type=ref_address_type)

        ref_right = self.dut_scan_for_asha(dut_address_type=dut_address_type)

        dut_ref_right, ref_right_dut = self.dut_connect_to_ref(

            advertisement=advertisement_right, ref=ref_right, dut_address_type=dut_address_type

        )

        advertisement_right.cancel()

        assert dut_ref_right

        assert ref_right_dut



        if disconnect_device == Device.LEFT:

            self.ref_left.host.Disconnect(connection=ref_left_dut)

            assert self.is_device_connected(device=self.ref_right, connection=ref_right_dut, timeout=5.0) == True

            assert self.is_device_connected(device=self.ref_left, connection=ref_left_dut, timeout=5.0) == False

        else:

            self.ref_right.host.Disconnect(connection=ref_right_dut)

            assert self.is_device_connected(device=self.ref_right, connection=ref_right_dut, timeout=5.0) == False

            assert self.is_device_connected(device=self.ref_left, connection=ref_left_dut, timeout=5.0) == True





if __name__ == "__main__":

    logging.basicConfig(level=logging.DEBUG)

android/pandora/test/avatar.sh

+1 −1
Original line number Diff line number Diff line
@@ -12,7 +12,7 @@ _USAGE="avatar [OPTIONS] <COMMAND> ... 
      OPTIONS: (subset, see 'avatar run --help-all')

        --include-filter=<ClassA[#test_a]>

                         Add a test filter in form of 'ClassA[#test_a]'.

        --test-bed       Set mobly test bed (default is 'android.bumble').

        --test-bed       Set mobly test bed (default is 'android.bumbles').

        --mobly-std-log  Print mobly logs to standard outputs.

        --mobly-options=<'--opt-a --opt-b'>

                         Pass additional options to mobly, like '--verbose' or '--list'.