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

Commit 13ec1a82 authored by JohnLai's avatar JohnLai
Browse files

Floss: Initialize folder for floss pandora server 

This server is used by the pandora (pts-bot and avatar) with Floss.

Design doc: https://docs.google.com/document/d/1-mRTZKDFhZYe4Iqf_YRXhNduwyCfTN5dz105iycfyTE/edit?resourcekey=0-WzM7muW9I05lz1uzskNl5Q#heading=h.17wg41voij6q

Bug: 289480188
Test: mma packages/modules/Bluetooth
Tag: #floss
Change-Id: I061babc5dad3baa6c32e265f34744dd3bbb25c06
parent 2673180e

floss/pandora/server/__init__.py

0 → 100644
+0 −0

Empty file added.

floss/pandora/server/host.py

0 → 100644
+116 −0
Original line number Diff line number Diff line 
# Copyright 2023 Google LLC

#

# 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

#

#     https://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.

"""Host grpc interface."""



from typing import AsyncGenerator



from google.protobuf import empty_pb2

import grpc

from pandora import host_grpc_aio

from pandora import host_pb2





class HostService(host_grpc_aio.HostServicer):

    """Service to trigger Bluetooth Host procedures.



    This class implements the Pandora bluetooth test interfaces,

    where the meta class definition is automatically generated by the protobuf.

    The interface definition can be found in:

    https://cs.android.com/android/platform/superproject/+/main:external

    /pandora/bt-test-interfaces/pandora/host.proto

    """



    async def FactoryReset(self, request: empty_pb2.Empty, context: grpc.ServicerContext) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def Reset(self, request: empty_pb2.Empty, context: grpc.ServicerContext) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def ReadLocalAddress(self, request: empty_pb2.Empty,

                               context: grpc.ServicerContext) -> host_pb2.ReadLocalAddressResponse:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def Connect(self, request: host_pb2.ConnectRequest,

                      context: grpc.ServicerContext) -> host_pb2.ConnectResponse:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def WaitConnection(self, request: host_pb2.WaitConnectionRequest,

                             context: grpc.ServicerContext) -> host_pb2.WaitConnectionResponse:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def ConnectLE(self, request: host_pb2.ConnectLERequest,

                        context: grpc.ServicerContext) -> host_pb2.ConnectLEResponse:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def Disconnect(self, request: host_pb2.DisconnectRequest, context: grpc.ServicerContext) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def WaitDisconnection(self, request: host_pb2.WaitDisconnectionRequest,

                                context: grpc.ServicerContext) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def Advertise(self, request: host_pb2.AdvertiseRequest,

                        context: grpc.ServicerContext) -> AsyncGenerator[host_pb2.AdvertiseResponse, None]:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")

        yield host_pb2.AdvertiseResponse()  # no-op: to make the linter happy



    async def Scan(self, request: host_pb2.ScanRequest,

                   context: grpc.ServicerContext) -> AsyncGenerator[host_pb2.ScanningResponse, None]:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")

        yield host_pb2.ScanningResponse()  # no-op: to make the linter happy



    async def Inquiry(self, request: empty_pb2.Empty,

                      context: grpc.ServicerContext) -> AsyncGenerator[host_pb2.InquiryResponse, None]:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")

        yield host_pb2.InquiryResponse()  # no-op: to make the linter happy



    async def SetDiscoverabilityMode(

        self,

        request: host_pb2.SetDiscoverabilityModeRequest,

        context: grpc.ServicerContext,

    ) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def SetConnectabilityMode(

        self,

        request: host_pb2.SetConnectabilityModeRequest,

        context: grpc.ServicerContext,

    ) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")

floss/pandora/server/security.py

0 → 100644
+76 −0
Original line number Diff line number Diff line 
# Copyright 2023 Google LLC

#

# 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

#

#     https://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.

"""Security grpc interface."""



from typing import AsyncGenerator

from typing import AsyncIterator



from google.protobuf import empty_pb2

from google.protobuf import wrappers_pb2

import grpc

from pandora import security_pb2

from pandora import security_grpc_aio





class SecurityService(security_grpc_aio.SecurityServicer):

    """Service to trigger Bluetooth Host security pairing procedures.



    This class implements the Pandora bluetooth test interfaces,

    where the meta class definition is automatically generated by the protobuf.

    The interface definition can be found in:

    https://cs.android.com/android/platform/superproject/+/main:external

    /pandora/bt-test-interfaces/pandora/security.proto

    """



    async def OnPairing(self, request: AsyncIterator[security_pb2.PairingEventAnswer],

                        context: grpc.ServicerContext) -> AsyncGenerator[security_pb2.PairingEvent, None]:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")

        yield security_pb2.PairingEvent()  # no-op: to make the linter happy



    async def Secure(self, request: security_pb2.SecureRequest,

                     context: grpc.ServicerContext) -> security_pb2.SecureResponse:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def WaitSecurity(self, request: security_pb2.WaitSecurityRequest,

                           context: grpc.ServicerContext) -> security_pb2.WaitSecurityResponse:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")





class SecurityStorageService(security_grpc_aio.SecurityStorageServicer):

    """Service to trigger Bluetooth Host security persistent storage procedures.



    This class implements the Pandora bluetooth test interfaces,

    where the meta class definition is automatically generated by the protobuf.

    The interface definition can be found in:

    https://cs.android.com/android/platform/superproject/+/main:external

    /pandora/bt-test-interfaces/pandora/security.proto

    """



    async def IsBonded(self, request: security_pb2.IsBondedRequest,

                       context: grpc.ServicerContext) -> wrappers_pb2.BoolValue:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")



    async def DeleteBond(self, request: security_pb2.DeleteBondRequest,

                         context: grpc.ServicerContext) -> empty_pb2.Empty:

        context.set_code(grpc.StatusCode.UNIMPLEMENTED)  # type: ignore

        context.set_details("Method not implemented!")  # type: ignore

        raise NotImplementedError("Method not implemented!")