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

Commit a5dfff8b authored by Aritra Sen's avatar Aritra Sen
Browse files

Creating a Python wrapper for gRPC service for HF Client. 

Bug: 262045500
Test: system/gd/cert/run --topshim --clean
Tag: #floss
Change-Id: I0c9ffe07233c9ff92d2396b84705c997e5ce595b
parent 75e09327

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

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





class HfClientClient(AsyncClosable):

    """

    Wrapper gRPC interface to the HF Client Service

    """

    __channel = None

    __hf_client_stub = None



    def __init__(self, port=8999):

        self.__channel = grpc.aio.insecure_channel("localhost:%d" % port)

        self.__hf_client_stub = facade_pb2_grpc.HfClientServiceStub(self.__channel)



    async def close(self):

        await self.__channel.close()



    async def start_slc(self, address):

        """

        """

        await self.__hf_client_stub.StartSlc(

            facade_pb2.StartSlcRequest(connection=facade_pb2.Connection(cookie=address.encode())))



    async def stop_slc(self, address):

        """

        """

        await self.__hf_client_stub.StopSlc(

            facade_pb2.StopSlcRequest(connection=facade_pb2.Connection(cookie=address.encode())))



    async def connect_audio(self, address):

        """

        """

        await self.__hf_client_stub.ConnectAudio(

            facade_pb2.ConnectAudioRequest(connection=facade_pb2.Connection(cookie=address.encode())))



    async def disconnect_audio(self, address):

        """

        """

        await self.__hf_client_stub.DisconnectAudio(

            facade_pb2.DisconnectAudioRequest(connection=facade_pb2.Connection(cookie=address.encode())))

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

+8 −2
Original line number Diff line number Diff line
@@ -34,6 +34,8 @@ from blueberry.tests.gd.cert.truth import assertThat 
from blueberry.tests.topshim.lib.adapter_client import AdapterClient

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

from blueberry.tests.topshim.lib.gatt_client import GattClient

from blueberry.tests.topshim.lib.hf_client_client import HfClientClient

from blueberry.tests.topshim.lib.hfp_client import HfpClient

from blueberry.tests.topshim.lib.security_client import SecurityClient

from blueberry.tests.topshim.lib.topshim_device import TopshimDevice
@@ -169,9 +171,13 @@ class TopshimBaseTest(base_test.BaseTestClass): 
        started = started and await cert_adapter._verify_adapter_started()

        assertThat(started).isTrue()

        self.__dut = TopshimDevice(dut_adapter, GattClient(port=self.dut_port),

                                   SecurityClient(dut_adapter, port=self.dut_port))

                                   SecurityClient(dut_adapter, port=self.dut_port),

                                   HfpClient(port=self.dut_port),

                                   HfClientClient(port=self.dut_port))

        self.__cert = TopshimDevice(cert_adapter, GattClient(port=self.cert_port),

                                    SecurityClient(cert_adapter, port=self.cert_port))

                                    SecurityClient(cert_adapter, port=self.cert_port),

                                    HfpClient(port=self.cert_port),

                                    HfClientClient(port=self.cert_port))

        return started



    async def __teardown_adapter(self):

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

+7 −1
Original line number Diff line number Diff line
@@ -76,6 +76,8 @@ class TopshimDevice(AsyncClosable): 
    __adapter = None

    __gatt = None

    __security = None

    __hfp = None

    __hf_client = None



    async def __le_rand_wrapper(self, async_fn):

        result = await async_fn
@@ -86,10 +88,12 @@ class TopshimDevice(AsyncClosable): 
    def __post(self, async_fn):

        return asyncio.get_event_loop().run_until_complete(async_fn)



    def __init__(self, adapter, gatt, security):

    def __init__(self, adapter, gatt, security, hfp, hf_client):

        self.__adapter = adapter

        self.__gatt = gatt

        self.__security = security

        self.__hfp = hfp

        self.__hf_client = hf_client



    async def close(self):

        """
@@ -98,6 +102,8 @@ class TopshimDevice(AsyncClosable): 
        await asyncSafeClose(self.__adapter)

        await asyncSafeClose(self.__gatt)

        await asyncSafeClose(self.__security)

        await asyncSafeClose(self.__hfp)

        await asyncSafeClose(self.__hf_client)



    def enable_page_scan(self):

        self.__post(self.__adapter.enable_page_scan())