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

Commit 1811a0e5 authored by Martin Brabham's avatar Martin Brabham
Browse files

Simplify test writing

Bug: 234756905
Test: system/gd/cert/run --clean --topshim
Tag: #floss
Change-Id: I7eb7dbb5665072c2a69efa4dbcfd8f38e2706c1e
parent 37efc46a
Loading
Loading
Loading
Loading
+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__":
+7 −1
Original line number Diff line number Diff line
@@ -75,9 +75,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())
+22 −4
Original line number Diff line number Diff line
@@ -33,6 +33,7 @@ from blueberry.tests.gd.cert.tracelogger import TraceLogger
from blueberry.tests.gd.cert.truth import assertThat
from blueberry.tests.topshim.lib.adapter_client import AdapterClient
from blueberry.tests.topshim.lib.gatt_client import GattClient
from blueberry.tests.topshim.lib.topshim_device import TopshimDevice

from mobly import asserts
from mobly import base_test
@@ -157,6 +158,9 @@ def dump_crashes_core(dut, cert, rootcanal_running, rootcanal_process, rootcanal

class TopshimBaseTest(base_test.BaseTestClass):

    __dut = None
    __cert = None

    # TODO(optedoblivion): Make TopshimTestStack class
    dut_adapter = None
    dut_gatt = None
@@ -169,10 +173,12 @@ class TopshimBaseTest(base_test.BaseTestClass):
        self.cert_adapter = AdapterClient(port=self.cert_port)
        started = await self.dut_adapter._verify_adapter_started()
        assertThat(started).isTrue()
        started = await self.cert_adapter._verify_adapter_started()
        started = started and await self.cert_adapter._verify_adapter_started()
        assertThat(started).isTrue()
        self.dut_gatt = GattClient(port=self.dut_port)
        self.cert_gatt = GattClient(port=self.cert_port)
        self.__dut = TopshimDevice(self.dut_adapter, self.dut_gatt)
        self.__cert = TopshimDevice(self.cert_adapter, self.cert_gatt)
        return started

    async def _teardown_adapter(self):
@@ -181,7 +187,19 @@ class TopshimBaseTest(base_test.BaseTestClass):
        await self.cert_adapter.terminate()
        await self.cert_gatt.terminate()

    def post(self, async_fn):
    def dut(self):
        """
        Get a handle on the DUT device
        """
        return self.__dut

    def cert(self):
        """
        Get a handle on the CERT device
        """
        return self.__cert

    def __post(self, async_fn):
        asyncio.get_event_loop().run_until_complete(async_fn)

    def setup_class(self):
@@ -217,7 +235,7 @@ class TopshimBaseTest(base_test.BaseTestClass):
        self.cert_port = controllers[0].grpc_port
        self.dut_port = controllers[1].grpc_port
        asyncio.set_event_loop(asyncio.new_event_loop())
        asyncio.get_event_loop().run_until_complete(self._setup_adapter())
        self.__post(self._setup_adapter())

    def teardown_class(self):
        _teardown_class_core(
@@ -225,4 +243,4 @@ class TopshimBaseTest(base_test.BaseTestClass):
            rootcanal_process=self.rootcanal_process,
            rootcanal_logger=self.rootcanal_logger,
            subprocess_wait_timeout_seconds=1)
        asyncio.get_event_loop().run_until_complete(self._teardown_adapter())
        self.__post(self._teardown_adapter())
+19 −0
Original line number Diff line number Diff line
@@ -14,6 +14,7 @@
#   See the License for the specific language governing permissions and
#   limitations under the License.

import asyncio
import logging

from blueberry.tests.gd.cert.gd_device import GdHostOnlyDevice
@@ -62,3 +63,21 @@ def get_instances_with_configs(configs):
        device.setup()
        devices.append(device)
    return devices


class TopshimDevice():
    __adapter = None
    __gatt = None

    def __post(self, async_fn):
        asyncio.get_event_loop().run_until_complete(async_fn)

    def __init__(self, adapter, gatt):
        self.__adapter = adapter
        self.__gatt = gatt

    def enable_page_scan(self):
        self.__post(self.__adapter.enable_page_scan())

    def disable_page_scan(self):
        self.__post(self.__adapter.disable_page_scan())
+1 −1
Original line number Diff line number Diff line
@@ -21,7 +21,7 @@ from blueberry.tests.topshim.lib.adapter_client import AdapterClient
from mobly import test_runner


class SecurityTest(TopshimBaseTest):
class ClassicSecurityTest(TopshimBaseTest):

    DEFAULT_ADDRESS = "01:02:03:04:05:06"

Loading