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

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

Floss: Suspend Topshim Test

6 seconds is a REALLY long time.
The default has been modified to 2.

Should be overridden for scanning possibly, but everything else should
happen within 2 seconds.

Additionally, the suspend timeout is 2 seconds so we need to be done
within 2 seconds of calling le_rand.

Bug: 229264266
Test: mma -j $(nproc)
Test: ./build.py
Test: system/gd/cert/run_topshim
Tag: #floss
Change-Id: I39da69403589c25b6de8745aae3effbe20c1df53
parent a89fd654
Loading
Loading
Loading
Loading
+1 −1
Original line number Diff line number Diff line
@@ -25,7 +25,7 @@ from google.protobuf import empty_pb2 as empty_proto

class AdapterAutomationHelper():
    # Timeout for async wait
    DEFAULT_TIMEOUT = 6
    DEFAULT_TIMEOUT = 2
    """Invoke gRPC on topshim for Adapter testing"""

    def __init__(self, port=8999):
+17 −3
Original line number Diff line number Diff line
@@ -23,16 +23,30 @@ from blueberry.tests.gd.rust.topshim.facade.automation_helper import AdapterAuto
class SuspendTest(topshim_base_test.TopshimBaseTest):

    async def _test_verify_suspend(self):

        # Setup
        self.dut_adapter = AdapterAutomationHelper(port=self.dut_port)
        event_loop = asyncio.get_running_loop()
        self.dut_adapter.fetch_events(event_loop)

        # Get 'ON' event
        self.dut_adapter.pending_future = event_loop.create_future()
        self.dut_adapter.fetch_events(event_loop)
        await asyncio.wait_for(self.dut_adapter.pending_future, AdapterAutomationHelper.DEFAULT_TIMEOUT)

        # Start suspend work
        await self.dut_adapter.clear_event_filter()
        await self.dut_adapter.clear_event_mask()
        await self.dut_adapter.clear_filter_accept_list()
        # TODO(optedoblivion): Find a better way to disconnect active ACLs
        #await self.dut_adapter.disconnect_all_acls()
        await self.dut_adapter.le_rand()
        await asyncio.wait_for(self.dut_adapter.pending_future, AdapterAutomationHelper.DEFAULT_TIMEOUT)
        self.dut_adapter.fetch_events(event_loop)

        # Get 'LE RAND' event
        self.dut_adapter.pending_future = event_loop.create_future()
        self.dut_adapter.fetch_events(event_loop)
        await asyncio.wait_for(self.dut_adapter.pending_future, AdapterAutomationHelper.DEFAULT_TIMEOUT)

        # Teardown
        self.dut_adapter.event_handler.cancel()

    def test_verify_suspend(self):