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

Commit a48faab2 authored by Martin Brabham's avatar Martin Brabham
Browse files

Floss: Topshim test updates. 

Bug: 224602924
Bug: 224603540
Bug: 224603198
Test: mma -j $(nproc)
Test: ./build.py
Test: system/gd/cert/run --clean --topshim
Tag: #floss
Change-Id: Ib3ebafdff8bf09c22866b54f93a048603e88de1a
parent ca3bce0b

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

+6 −1
Original line number Diff line number Diff line
@@ -94,7 +94,6 @@ class AdapterClient(): 
    async def le_rand(self):

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

        future = await self._listen_for_event(facade_pb2.EventType.LE_RAND)

        #        await asyncio.wait_for(future, AdapterClient.DEFAULT_TIMEOUT)

        return future.result()



    async def restore_filter_accept_list(self):
@@ -106,6 +105,12 @@ class AdapterClient(): 
    async def set_event_filter_inquiry_result_all_devices(self):

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



    async def set_event_filter_connection_setup_all_devices(self):

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



    async def allow_wake_by_hid(self):

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





class A2dpAutomationHelper():

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

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

+67 −19
Original line number Diff line number Diff line
@@ -25,34 +25,82 @@ from mobly import test_runner 


class SuspendTest(TopshimBaseTest):



    async def __verify_disconnected_suspend(self):

    async def __verify_no_wake_suspend(self):

        # Start suspend work

        await self.dut_adapter.clear_event_filter()

        await self.dut_adapter.clear_event_mask()

        await self.dut_adapter.clear_event_filter()

        await self.dut_adapter.clear_filter_accept_list()

        await self.dut_adapter.disconnect_all_acls()

        random = await self.dut_adapter.le_rand()

        return random

        # In the real module it will iterate advertiser ids

        await self.dut_gatt.unregister_advertiser()

        # In the real module it will iterate scanner ids

        await self.dut_gatt.stop_scan()

        return await self.dut_adapter.le_rand()



    async def __verify_disconnected_resume(self):

    async def __verify_no_wake_resume(self):

        # Start resume work

        await self.dut_adapter.set_default_event_mask()

        await self.dut_adapter.set_event_filter_inquiry_result_all_devices()

        await self.dut_adapter.set_event_filter_connection_setup_all_devices()

        return await self.dut_adapter.le_rand()



    async def __verify_wakeful_suspend(self, is_a2dp_connected):

        await self.dut_adapter.clear_event_mask()

        await self.dut_adapter.clear_event_filter()

        await self.dut_adapter.clear_filter_accept_list()

        await self.dut_adapter.disconnect_all_acls()

        # In the real module it will iterate advertiser ids

        await self.dut_gatt.unregister_advertiser()

        await self.dut_gatt.stop_scan()

        if is_a2dp_connected:

            # await self.dut_adapter.disconnect_a2dp()

            pass

        await self.dut_adapter.allow_wake_by_hid()

        return await self.dut_adapter.le_rand()



    async def __verify_wakeful_resume(self, was_a2dp_connected):

        # Start resume work

        await self.dut_adapter.set_default_event_mask()

        await self.dut_adapter.set_event_filter_inquiry_result_all_devices()

        await self.dut_adapter.set_event_filter_connection_setup_all_devices()

        if was_a2dp_connected:

            # restore filter accept list?

            await self.dut_adapter.restore_filter_accept_list()

        random = await self.dut_adapter.le_rand()

        return random

            # reconnect a2dp

            # await self.dut_adapter.reconnect_last_a2dp()

            # await self.dut_adapter.restart_all_advertising()

        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())



    def test_no_wake_resume(self):

        asyncio.get_event_loop().run_until_complete(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())



    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())



    def test_wakeful_suspend_no_a2dp(self):

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



    def test_wakeful_resume_no_a2dp(self):

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



    def test_disconnected_suspend(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_suspend())

    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))



    def test_disconnected_resume(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_resume())

    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))



    def test_disconnected_suspend_then_resume(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_suspend())

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_resume())



    def test_disconnected_suspend_then_resume_then_suspend(self):

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_suspend())

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_resume())

        asyncio.get_event_loop().run_until_complete(self.__verify_disconnected_suspend())
 
if __name__ == "__main__":

    test_runner.main()