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

Commit 47cc2954 authored by Aritra Sen's avatar Aritra Sen
Browse files

Add API to create bond between two devices using existing btif API. 

Successfully create an SSP connection between the DUT and the CERT using classic bluetooth.
This will be required for HFP profile related tests in the future.
The SSP is done using Passkey Confirmation.

Bug: 261221333
Test: mma -j $(nproc)
Test: ./build.py
Test: system/gd/cert/run --clean --topshim
Tag: #floss
Change-Id: Ib0b874ea18fc81d845ba422d339b97126d84f883
parent 69e0a821

system/blueberry/facade/topshim/facade.proto

+11 −0
Original line number Diff line number Diff line
@@ -25,6 +25,7 @@ service AdapterService { 
service SecurityService {

    rpc RemoveBond(RemoveBondRequest) returns (google.protobuf.Empty) {}

    rpc GenerateLocalOobData(GenerateOobDataRequest) returns (google.protobuf.Empty) {}

    rpc CreateBond(CreateBondRequest) returns (CreateBondResponse) {}

}



service GattService {
@@ -128,6 +129,7 @@ enum EventType { 
  ADAPTER_PROPERTY = 5;

  DISCOVERY_STATE = 6;

  DEVICE_FOUND = 7;

  BOND_STATE = 8;

}



message FetchEventsRequest {}
@@ -257,3 +259,12 @@ message ToggleDiscoveryRequest { 
message ToggleDiscoveryResponse {

  int32 status = 1;

}



message CreateBondRequest {

  string address = 1;

  int32 transport = 2;

}



message CreateBondResponse {

  int32 status = 1;

}
 
 No newline at end of file

system/blueberry/tests/topshim/lib/security_client.py

+10 −0
Original line number Diff line number Diff line
@@ -49,6 +49,16 @@ class SecurityClient(AsyncClosable): 
        self.__task_list.clear()

        await self.__channel.close()



    async def create_bond(self, address, transport):

        """

        Create a bonding entry for a given address with a particular transport type

        """

        await self.__security.CreateBond(facade_pb2.CreateBondRequest(address=address, transport=transport))

        # State change to Bonding

        await self.__adapter._listen_for_event(facade_pb2.EventType.BOND_STATE)

        # State change to Bonded or None (based on success and failure)

        return await self.__adapter._listen_for_event(facade_pb2.EventType.BOND_STATE)



    async def remove_bond(self, address):

        """

        Removes a bonding entry for a given address

system/blueberry/tests/topshim/lib/topshim_device.py

+13 −1
Original line number Diff line number Diff line
@@ -168,12 +168,24 @@ class TopshimDevice(AsyncClosable): 
    def le_rand(self):

        self.__post(self.__adapter.le_rand())



    def create_bond(self, address, transport=1):

        """

        Create a bonding entry for a given address with a particular transport type.

        """

        f = self.__post(self.__security.create_bond(address, transport))

        return self.__post(self.__bond_change_waiter(f))



    def remove_bonded_device(self, address):

        """

        Removes a bonding entry for a given address.

        """

        self.__post(self.__security.remove_bond(address))



    async def __bond_change_waiter(self, f):

        params = await f

        state, address = params["bond_state"].data[0], params["address"].data[0]

        return (state, address)



    def generate_local_oob_data(self, transport=TRANSPORT_LE):

        """

        Generate local Out of Band data
@@ -206,7 +218,7 @@ class TopshimDevice(AsyncClosable): 


        async def waiter(f):

            params = await f

            return params["state"].data[0]

            return params["discovery_state"].data[0]



        return self.__post(waiter(f))

system/blueberry/tests/topshim/security/classic_security_test.py

0 → 100644
+38 −0
Original line number Diff line number Diff line 
#!/usr/bin/env python3

#

#   Copyright 2022 - The Android Open Source Project

#

#   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

#

#       http://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.



from blueberry.tests.gd.cert.truth import assertThat

from blueberry.tests.topshim.lib.adapter_client import AdapterClient

from blueberry.tests.topshim.lib.topshim_base_test import TopshimBaseTest

from blueberry.tests.topshim.lib.topshim_device import TRANSPORT_CLASSIC



from mobly import test_runner





class ClassicSecurityTest(TopshimBaseTest):



    def test_create_classic_bond(self):

        self.dut().enable_inquiry_scan()

        self.cert().enable_inquiry_scan()

        self.dut().toggle_discovery(True)

        address = self.dut().find_device()

        state, conn_addr = self.dut().create_bond(address=address, transport=TRANSPORT_CLASSIC)

        assertThat(state).isEqualTo("Bonded")

        assertThat(conn_addr).isEqualTo(address)





if __name__ == "__main__":

    test_runner.main()

system/blueberry/tests/topshim/topshim_test_runner.py

+2 −1
Original line number Diff line number Diff line
@@ -16,12 +16,13 @@ 


from blueberry.tests.topshim.adapter.adapter_test import AdapterTest

from blueberry.tests.topshim.power.suspend_test import SuspendTest

from blueberry.tests.topshim.security.classic_security_test import ClassicSecurityTest

from blueberry.tests.topshim.security.le_security_test import LeSecurityTest



from mobly import suite_runner

import argparse



ALL_TESTS = [AdapterTest, LeSecurityTest, SuspendTest]

ALL_TESTS = [AdapterTest, ClassicSecurityTest, LeSecurityTest, SuspendTest]





def main():