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

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

Merge changes from topic "topshim_security_api" 

* changes:
  Topshim Tests: Add a central function for perfoming async functions.
  Topshim API: Add SecurityTest framework.
parents b19f7934 52d84167

system/blueberry/facade/topshim/facade.proto

+5 −0
Original line number Diff line number Diff line
@@ -15,6 +15,7 @@ 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) {}
@@ -144,3 +145,7 @@ message A2dpSourceSetActiveDeviceRequest { 
}



message A2dpSourceSetActiveDevicetResponse {}



message RemoveBondRequest {

  string address = 1;

}

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

+1 −1
Original line number Diff line number Diff line
@@ -33,7 +33,7 @@ class AdapterTest(TopshimBaseTest): 
        print("Adapter is verified when test starts")



    def test_enable_page_scan(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_enable_page_scan())

        self.post(self.__verify_enable_page_scan())





if __name__ == "__main__":

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

+3 −0
Original line number Diff line number Diff line
@@ -111,6 +111,9 @@ 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/topshim_base_test.py

+3 −0
Original line number Diff line number Diff line
@@ -181,6 +181,9 @@ class TopshimBaseTest(base_test.BaseTestClass): 
        await self.cert_adapter.terminate()

        await self.cert_gatt.terminate()



    def post(self, async_fn):

        asyncio.get_event_loop().run_until_complete(async_fn)



    def setup_class(self):

        """

        Configure rootcanal and setup test parameters

system/blueberry/tests/topshim/power/suspend_test.py

+14 −16
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
@@ -70,34 +68,34 @@ class SuspendTest(TopshimBaseTest): 
        return await self.dut_adapter.le_rand()



    def test_no_wake_suspend(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_suspend())

        self.post(self.__verify_no_wake_suspend())



    def test_no_wake_resume(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_resume())

        self.post(self.__verify_no_wake_resume())



    def test_no_wake_suspend_then_resume(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_suspend())

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_resume())

        self.post(self.__verify_no_wake_suspend())

        self.post(self.__verify_no_wake_resume())



    def test_no_wake_suspend_then_resume_then_suspend(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_suspend())

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_resume())

        asyncio.get_event_loop().run_until_complete(self.__verify_no_wake_suspend())

        self.post(self.__verify_no_wake_suspend())

        self.post(self.__verify_no_wake_resume())

        self.post(self.__verify_no_wake_suspend())



    def test_wakeful_suspend_no_a2dp(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_suspend(False))

        self.post(self.__verify_wakeful_suspend(False))



    def test_wakeful_resume_no_a2dp(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_resume(False))

        self.post(self.__verify_wakeful_resume(False))



    def test_wakeful_suspend_then_resume_no_a2dp(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_suspend(False))

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_resume(False))

        self.post(self.__verify_wakeful_suspend(False))

        self.post(self.__verify_wakeful_resume(False))



    def test_wakeful_suspend_then_resume_then_suspend_no_a2dp(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_suspend(False))

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_resume(False))

        asyncio.get_event_loop().run_until_complete(self.__verify_wakeful_suspend(False))

        self.post(self.__verify_wakeful_suspend(False))

        self.post(self.__verify_wakeful_resume(False))

        self.post(self.__verify_wakeful_suspend(False))





if __name__ == "__main__":