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

Commit 4d04803f authored by Treehugger Robot's avatar Treehugger Robot Committed by Gerrit Code Review
Browse files

Merge changes from topic "topshim_security_api" 

* changes:
  Topshim Testing: Classic Security Test Framework
  Topshim Testing: Further simplification
  Simplify test writing
parents 5fbb8839 17eef576

system/blueberry/facade/topshim/facade.proto

+4 −1
Original line number Diff line number Diff line
@@ -15,12 +15,15 @@ service AdapterService { 
  rpc LeRand(google.protobuf.Empty) returns (google.protobuf.Empty) {}

  rpc SetEventFilterConnectionSetupAllDevices(google.protobuf.Empty) returns (google.protobuf.Empty) {}

  rpc AllowWakeByHid(google.protobuf.Empty) returns (google.protobuf.Empty) {}

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

  rpc RestoreFilterAcceptList(google.protobuf.Empty) returns (google.protobuf.Empty) {}

  rpc SetDefaultEventMask(google.protobuf.Empty) returns (google.protobuf.Empty) {}

  rpc SetEventFilterInquiryResultAllDevices(google.protobuf.Empty) returns (google.protobuf.Empty) {}

}



service SecurityService {

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

}



service GattService {

  // Advertiser

  rpc RegisterAdvertiser(google.protobuf.Empty) returns (google.protobuf.Empty) {}

system/blueberry/tests/topshim/adapter/adapter_test.py

+1 −7
Original line number Diff line number Diff line
@@ -14,8 +14,6 @@ 
#   See the License for the specific language governing permissions and

#   limitations under the License.



import asyncio



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

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

from blueberry.tests.topshim.lib.adapter_client import AdapterClient
@@ -25,15 +23,11 @@ from mobly import test_runner 


class AdapterTest(TopshimBaseTest):



    async def __verify_enable_page_scan(self):

        await self.dut_adapter.set_enable_page_scan()

        return await self.dut_adapter.le_rand()



    def test_verify_adapter_started(self):

        print("Adapter is verified when test starts")



    def test_enable_page_scan(self):

        self.post(self.__verify_enable_page_scan())

        self.dut().enable_page_scan()





if __name__ == "__main__":

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

+10 −6
Original line number Diff line number Diff line
@@ -19,11 +19,12 @@ import grpc 


from blueberry.facade.topshim import facade_pb2

from blueberry.facade.topshim import facade_pb2_grpc

from blueberry.tests.topshim.lib.async_closable import AsyncClosable



from google.protobuf import empty_pb2 as empty_proto





class AdapterClient():

class AdapterClient(AsyncClosable):

    """

    Wrapper gRPC interface to the Topshim/BTIF layer

    """
@@ -39,7 +40,7 @@ class AdapterClient(): 
        self.__adapter_stub = facade_pb2_grpc.AdapterServiceStub(self.__channel)

        self.__adapter_event_stream = self.__adapter_stub.FetchEvents(facade_pb2.FetchEventsRequest())



    async def terminate(self):

    async def close(self):

        for task in self.__task_list:

            task.cancel()

            task = None
@@ -75,9 +76,15 @@ class AdapterClient(): 
        await self.__adapter_stub.ToggleStack(facade_pb2.ToggleStackRequest(start_stack=is_start))

        return await self._verify_adapter_started()



    async def set_enable_page_scan(self):

    async def enable_page_scan(self):

        """Enable page scan (might be used for A2dp sink to be discoverable)"""

        await self.__adapter_stub.SetDiscoveryMode(facade_pb2.SetDiscoveryModeRequest(enable_page_scan=True))

        return await self.le_rand()



    async def disable_page_scan(self):

        """Enable page scan (might be used for A2dp sink to be discoverable)"""

        await self.__adapter_stub.SetDiscoveryMode(facade_pb2.SetDiscoveryModeRequest(enable_page_scan=False))

        return await self.le_rand()



    async def clear_event_filter(self):

        await self.__adapter_stub.ClearEventFilter(empty_proto.Empty())
@@ -111,9 +118,6 @@ class AdapterClient(): 
    async def allow_wake_by_hid(self):

        await self.__adapter_stub.AllowWakeByHid(empty_proto.Empty())



    async def remove_bond(self, address):

        await self.__adapter_stub.RemoveBond(facade_pb2.RemoveBondRequest(address=address))





class A2dpAutomationHelper():

    """Invoke gRPC on topshim for A2DP testing"""

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

0 → 100644
+50 −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.



import asyncio

import time

from abc import ABC, abstractmethod

import logging





class AsyncClosable(ABC):



    async def __async_exit(self, type=None, value=None, traceback=None):

        try:

            return await self.close()

        except Exception:

            logging.warning("Failed to close or already closed")



    def __enter__(self):

        return self



    def __exit__(self, type, value, traceback):

        asyncio.run_until_complete(self.__async_exit(type, value, traceback))

        return traceback is None



    def __del__(self):

        asyncio.get_event_loop().run_until_complete(self.__async_exit())



    @abstractmethod

    async def close(self):

        pass





async def asyncSafeClose(closable):

    if closable is None:

        logging.warn("Tried to close an object that is None")

        return

    await closable.close()

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

+4 −2
Original line number Diff line number Diff line
@@ -19,11 +19,13 @@ import grpc 


from blueberry.facade.topshim import facade_pb2

from blueberry.facade.topshim import facade_pb2_grpc

from blueberry.tests.topshim.lib.async_closable import AsyncClosable

from blueberry.tests.topshim.lib.async_closable import asyncSafeClose



from google.protobuf import empty_pb2 as empty_proto





class GattClient():

class GattClient(AsyncClosable):

    """

    Wrapper gRPC interface to the GATT Service

    """
@@ -39,7 +41,7 @@ class GattClient(): 
        self.__gatt_stub = facade_pb2_grpc.GattServiceStub(self.__channel)

        #self.__gatt_event_stream = self.__gatt_stub.FetchEvents(facade_pb2.FetchEventsRequest())



    async def terminate(self):

    async def close(self):

        """

        Terminate the current tasks

        """