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

Commit c9e456a0 authored by Henri Chataing's avatar Henri Chataing Committed by Gerrit Code Review
Browse files

Merge changes from topic "pts-vcp" into main

* changes:
  Pandora: Add bindings for VCP profile tests
  VolumeControlService: Push registerCallback and unregisterCallback to the handler thread
parents 0d605c74 997de039
Loading
Loading
Loading
Loading
+19 −7
Original line number Diff line number Diff line
@@ -85,8 +85,7 @@ public class VolumeControlService extends ProfileService {
    private Handler mHandler = null;
    private FeatureFlags mFeatureFlags;

    @VisibleForTesting
    RemoteCallbackList<IBluetoothVolumeControlCallback> mCallbacks;
    @VisibleForTesting RemoteCallbackList<IBluetoothVolumeControlCallback> mCallbacks;

    @VisibleForTesting
    static class VolumeControlOffsetDescriptor {
@@ -1775,11 +1774,18 @@ public class VolumeControlService extends ProfileService {
                }

                enforceBluetoothPrivilegedPermission(service);
                service.mHandler.post(
                        () -> {
                            try {
                                service.registerCallback(callback);
                                receiver.send(null);
                            } catch (RuntimeException e) {
                                receiver.propagateException(e);
                            }
                        });
            } catch (RuntimeException e) {
                receiver.propagateException(e);
            }
        }

        @Override
@@ -1796,12 +1802,18 @@ public class VolumeControlService extends ProfileService {
                }

                enforceBluetoothPrivilegedPermission(service);

                service.mHandler.post(
                        () -> {
                            try {
                                service.mCallbacks.unregister(callback);
                                receiver.send(null);
                            } catch (RuntimeException e) {
                                receiver.propagateException(e);
                            }
                        });
            } catch (RuntimeException e) {
                receiver.propagateException(e);
            }
        }
    }

+8 −0
Original line number Diff line number Diff line
@@ -37,6 +37,7 @@ from mmi2grpc.pbap import PBAPProxy
from mmi2grpc.rfcomm import RFCOMMProxy
from mmi2grpc.sdp import SDPProxy
from mmi2grpc.sm import SMProxy
from mmi2grpc.vcp import VCPProxy
from mmi2grpc._helpers import format_proxy
from mmi2grpc._rootcanal import RootCanal
from mmi2grpc._modem import Modem
@@ -89,6 +90,7 @@ class IUT:
        self._rfcomm = None
        self._sdp = None
        self._sm = None
        self._vcp = None

    def __enter__(self):
        """Resets the IUT when starting a PTS test."""
@@ -125,6 +127,7 @@ class IUT:
        self._rfcomm = None
        self._sdp = None
        self._sm = None
        self._vcp = None

    def _retry(self, func):

@@ -273,6 +276,11 @@ class IUT:
            if not self._sm:
                self._sm = SMProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"), self.rootcanal)
            return self._sm.interact(test, interaction, description, pts_address)
        # HandlesVCP MMIs.
        if profile in ("VCP"):
            if not self._vcp:
                self._vcp = VCPProxy(grpc.insecure_channel(f"localhost:{self.pandora_server_port}"), self.rootcanal)
            return self._vcp.interact(test, interaction, description, pts_address)

        # Handles unsupported profiles.
        code = format_proxy(profile, interaction, description)
+2 −0
Original line number Diff line number Diff line
@@ -145,6 +145,7 @@ class SDPProxy(ProfileProxy):
            "HandsfreeAudioGateway",
            "GenericAudio",
            "Message Access Server",
            "TMAS",
            "NAP",
            "PANU",
            "Phonebook Access - PSE",
@@ -155,6 +156,7 @@ class SDPProxy(ProfileProxy):
            "Generic Attribute service",
            "A/V_RemoteControlController",
            "Android Auto Compatibility",
            "TMAS",
        ]
        movable_services = [
            "Message Access Server",
+121 −0
Original line number Diff line number Diff line
# Copyright 2022 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.
"""VCP proxy module."""
import threading

from mmi2grpc._helpers import assert_description, match_description
from mmi2grpc._proxy import ProfileProxy
from mmi2grpc._rootcanal import Dongle

from pandora.security_grpc import Security
from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer
from pandora.host_grpc import Host
from pandora.host_pb2 import PUBLIC, RANDOM
from pandora_experimental.le_audio_grpc import LeAudio


class VCPProxy(ProfileProxy):

    def __init__(self, channel, rootcanal):
        super().__init__(channel)
        self.host = Host(channel)
        self.security = Security(channel)
        self.le_audio = LeAudio(channel)
        self.rootcanal = rootcanal
        self.connection = None
        self.pairing_stream = None

    def test_started(self, test: str, description: str, pts_addr: bytes):
        self.rootcanal.select_pts_dongle(Dongle.LAIRD_BL654)

        return "OK"

    @assert_description
    def IUT_INITIATE_CONNECTION(self, pts_addr: bytes, **kwargs):
        """
        Please initiate a GATT connection to the PTS.

        Description: Verify that
        the Implementation Under Test (IUT) can initiate a GATT connect request
        to the PTS.
        """
        self.connection = self.host.ConnectLE(own_address_type=RANDOM, public=pts_addr).connection
        self.pairing_stream = self.security.OnPairing()

        def secure():
            self.security.Secure(connection=self.connection, le=LE_LEVEL3)

        threading.Thread(target=secure).start()
        return "OK"

    @match_description
    def _mmi_2004(self, pts_addr: bytes, passkey: str, **kwargs):
        """
        Please confirm that 6 digit number is matched with (?P<passkey>[0-9]*).
        """
        received = []
        for event in self.pairing_stream:
            if event.address == pts_addr and event.numeric_comparison == int(passkey):
                self.pairing_stream.send(PairingEventAnswer(
                    event=event,
                    confirm=True,
                ))
                return "OK"
            received.append(event.numeric_comparison)

        assert False, f"mismatched passcode: expected {passkey}, received {received}"

    @match_description
    def IUT_INITIATE_DISCOVER_CHARACTERISTIC(self, **kwargs):
        """
        Please take action to discover the
        (Volume Control Point|Volume State|Volume Flags|Offset State|Volume Offset Control Point)
        characteristic from the Volume (Offset)? Control. Discover the primary service if needed.
        Description: Verify that the Implementation Under Test \(IUT\) can send
        Discover All Characteristics command.
        """
        return "OK"

    @match_description
    def IUT_READ_CHARACTERISTIC(self, name: str, handle: str, **kwargs):
        """
        Please send Read Request to read (?P<name>(Volume State|Volume Flags|Offset State)) characteristic with handle
        = (?P<handle>(0x[0-9A-Fa-f]{4})).
        """
        return "OK"

    @assert_description
    def USER_CONFIRM_SUPPORTED_CHARACTERISTIC(self, characteristics: str, **kwargs):
        """
        Please verify that for each supported characteristic, attribute
        handle/UUID pair(s) is returned to the upper tester.(?P<characteristics>(.|\n)*)
        """

        return "OK"

    @match_description
    def IUT_CONFIG_NOTIFICATION(self, name: str, **kwargs):
        """
        Please write to Client Characteristic Configuration Descriptor of
        (?P<name>(Volume State|Offset State)) characteristic to enable notification.
        """
        return "OK"

    @assert_description
    def _mmi_20501(self, **kwargs):
        """
        Please start general inquiry. Click 'Yes' If IUT does discovers PTS
        otherwise click 'No'.
        """
        return "OK"
+1 −0
Original line number Diff line number Diff line
@@ -56,5 +56,6 @@
        <option name="profile" value="RFCOMM" />
        <option name="profile" value="SDP" />
        <option name="profile" value="SM" />
        <option name="profile" value="VCP" />
    </test>
</configuration>
Loading