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

Commit 9891512f authored by William Escande's avatar William Escande
Browse files

AVATAR: add test for gatt server timeout 

Note that this require hacking bumble as there is not yet the
implementation of L2CAP_Credit_Based_Connection_Request

Bug: 289460863
Test: atest avatar GattTest#test_eatt_when_not_encrypted_no_timeout
Change-Id: I03330f820770ee6fcf4d6d8c7386008d2ca0e853
parent 00008b43

android/pandora/test/gatt_test.py

+57 −0
Original line number Diff line number Diff line
@@ -14,6 +14,7 @@ 


import asyncio

import avatar

import grpc

import logging



from avatar import BumblePandoraDevice, PandoraDevice, PandoraDevices
@@ -23,6 +24,7 @@ from bumble.pairing import PairingConfig 
from bumble_experimental.gatt import GATTService

from mobly import base_test, signals, test_runner

from mobly.asserts import assert_equal  # type: ignore

from mobly.asserts import assert_false  # type: ignore

from mobly.asserts import assert_in  # type: ignore

from mobly.asserts import assert_is_not_none  # type: ignore

from mobly.asserts import assert_not_in  # type: ignore
@@ -238,6 +240,61 @@ class GattTest(base_test.BaseTestClass): # type: ignore[misc] 
        assert_in(SERVICE_UUID_1, (service.uuid for service in second_discovery.services))

        assert_in(SERVICE_UUID_2, (service.uuid for service in second_discovery.services))



    @avatar.asynchronous

    async def test_eatt_when_not_encrypted_no_timeout(self) -> None:

        if not isinstance(self.ref, BumblePandoraDevice):

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

        advertise = self.dut.aio.host.Advertise(

            legacy=True,

            connectable=True,

            own_address_type=RANDOM,

            data=DataTypes(manufacturer_specific_data=b'pause cafe'),

        )



        scan = self.ref.aio.host.Scan()

        dut = await anext((x async for x in scan if b'pause cafe' in x.data.manufacturer_specific_data))

        scan.cancel()



        ref_dut = (await self.ref.aio.host.ConnectLE(own_address_type=RANDOM, **dut.address_asdict())).connection

        assert_is_not_none(ref_dut)

        assert ref_dut

        advertise.cancel()



        connection = self.ref.device.lookup_connection(int.from_bytes(ref_dut.cookie.value, 'big'))

        assert connection



        connection_request = (

            b"\x17"  # code of L2CAP_CREDIT_BASED_CONNECTION_REQ

            b"\x01"  # identifier

            b"\x0a\x00"  # data length

            b"\x27\x00"  # psm(EATT)

            b"\x64\x00"  # MTU

            b"\x64\x00"  # MPS

            b"\x64\x00"  # initial credit

            b"\x40\x00"  # source cid[0]

        )



        fut = asyncio.get_running_loop().create_future()

        setattr(self.ref.device.l2cap_channel_manager, "on_[0x18]", lambda _, _1, frame: fut.set_result(frame))

        self.ref.device.l2cap_channel_manager.send_control_frame(  # type:ignore

            connection, 0x05, connection_request

        )

        control_frame = await fut



        assert_equal(bytes(control_frame)[10], 0x05)  # All connections refused – insufficient authentication



        # TODO(b/289460863): change to assert_true when the test is fixed

        assert_false(await is_connected(self.ref, ref_dut), "Device is no longer connected")





async def is_connected(device: PandoraDevice, connection: Connection) -> bool:

    try:

        await device.aio.host.WaitDisconnection(connection=connection, timeout=5)

        return False

    except grpc.RpcError as e:

        assert_equal(e.code(), grpc.StatusCode.DEADLINE_EXCEEDED)  # type: ignore

        return True





if __name__ == '__main__':

    logging.basicConfig(level=logging.DEBUG)