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

Commit 7800c046 authored by Android Build Coastguard Worker's avatar Android Build Coastguard Worker
Browse files

Snap for 8972412 from 66cb6df0 to tm-qpr1-release

Change-Id: Iaba7fd390d5727e01f96d09f5a24201d4c16d73c
parents 93ba1442 66cb6df0
Loading
Loading
Loading
Loading
+8 −0
Original line number Diff line number Diff line
@@ -22,6 +22,7 @@ import sys
import grpc

from mmi2grpc.a2dp import A2DPProxy
from mmi2grpc.avrcp import AVRCPProxy
from mmi2grpc.gatt import GATTProxy
from mmi2grpc.hfp import HFPProxy
from mmi2grpc.sdp import SDPProxy
@@ -54,6 +55,7 @@ class IUT:

        # Profile proxies.
        self._a2dp = None
        self._avrcp = None
        self._gatt = None
        self._hfp = None
        self._sdp = None
@@ -68,6 +70,7 @@ class IUT:

    def __exit__(self, exc_type, exc_value, exc_traceback):
        self._a2dp = None
        self._avrcp = None
        self._gatt = None
        self._hfp = None
        self._sdp = None
@@ -115,6 +118,11 @@ class IUT:
            if not self._a2dp:
                self._a2dp = A2DPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
            return self._a2dp.interact(test, interaction, description, pts_address)
        # Handles AVRCP and AVCTP MMIs.
        if profile in ('AVRCP', 'AVCTP'):
            if not self._avrcp:
                self._avrcp = AVRCPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
            return self._avrcp.interact(test, interaction, description, pts_address)
        # Handles GATT MMIs.
        if profile in ('GATT'):
            if not self._gatt:
+167 −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.
"""AVRCP proxy module."""

import time
from typing import Optional

from grpc import RpcError

from mmi2grpc._audio import AudioSignal
from mmi2grpc._helpers import assert_description
from mmi2grpc._proxy import ProfileProxy
from pandora.a2dp_grpc import A2DP
from pandora.a2dp_pb2 import Sink, Source
from pandora.avrcp_grpc import AVRCP
from pandora.host_grpc import Host
from pandora.host_pb2 import Connection


class AVRCPProxy(ProfileProxy):
    """AVRCP proxy.

    Implements AVRCP and AVCTP PTS MMIs.
    """

    connection: Optional[Connection] = None
    sink: Optional[Sink] = None
    source: Optional[Source] = None

    def __init__(self, channel):
        super().__init__()

        self.host = Host(channel)
        self.a2dp = A2DP(channel)
        self.avrcp = AVRCP(channel)

    @assert_description
    def TSC_AVDTP_mmi_iut_accept_connect(self, test: str, pts_addr: bytes, **kwargs):
        """
        If necessary, take action to accept the AVDTP Signaling Channel
        Connection initiated by the tester.

        Description: Make sure the IUT
        (Implementation Under Test) is in a state to accept incoming Bluetooth
        connections.  Some devices may need to be on a specific screen, like a
        Bluetooth settings screen, in order to pair with PTS.  If the IUT is
        still having problems pairing with PTS, try running a test case where
        the IUT connects to PTS to establish pairing.

        """
        if "CT" in test:

            self.connection = self.host.WaitConnection(address=pts_addr).connection
            try:
                self.source = self.a2dp.WaitSource(connection=self.connection).source
            except RpcError:
                pass
        else:
            self.connection = self.host.WaitConnection(address=pts_addr).connection
            try:
                self.sink = self.a2dp.WaitSink(connection=self.connection).sink
            except RpcError:
                pass
        return "OK"

    @assert_description
    def TSC_AVCTP_mmi_iut_accept_connect_control(self, **kwargs):
        """
        Please wait while PTS creates an AVCTP control channel connection.
        Action: Make sure the IUT is in a connectable state.

        """
        #TODO: Wait for connection to be established and AVCTP control channel to be open
        return "OK"

    @assert_description
    def TSC_AVCTP_mmi_iut_accept_disconnect_control(self, **kwargs):
        """
        Please wait while PTS disconnects the AVCTP control channel connection.

        """
        return "OK"

    @assert_description
    def TSC_AVRCP_mmi_iut_accept_unit_info(self, **kwargs):
        """
        Take action to send a valid response to the [Unit Info] command sent by
        the PTS.

        """
        return "OK"

    @assert_description
    def TSC_AVRCP_mmi_iut_accept_subunit_info(self, **kwargs):
        """
        Take action to send a valid response to the [Subunit Info] command sent
        by the PTS.

        """
        return "OK"

    @assert_description
    def TSC_AVCTP_mmi_iut_accept_connect_browsing(self, **kwargs):
        """
        Please wait while PTS creates an AVCTP browsing channel connection.
        Action: Make sure the IUT is in a connectable state.

        """
        return "OK"

    @assert_description
    def TSC_AVRCP_mmi_iut_accept_get_folder_items_media_player_list(self, **kwargs):
        """
        Take action to send a valid response to the [Get Folder Items] with the
        scope <Media Player List> command sent by the PTS.

        """
        return "OK"

    @assert_description
    def TSC_AVRCP_mmi_user_confirm_media_players(self, **kwargs):
        """
        Do the following media players exist on the IUT?

        Media Player:
        Bluetooth Player


        Note: Some media players may not be listed above.

        """
        #TODO: Verify the media players available
        return "OK"

    @assert_description
    def TSC_AVP_mmi_iut_initiate_disconnect(self, **kwargs):
        """
        Take action to disconnect all A2DP and/or AVRCP connections.

        """
        if self.connection is None:
            self.connection = self.host.GetConnection(address=pts_addr).connection
        self.host.Disconnect(connection=self.connection)
        self.connection = None
        self.sink = None
        self.source = None
        return "OK"

    @assert_description
    def TSC_AVRCP_mmi_iut_accept_set_addressed_player(self, **kwargs):
        """
        Take action to send a valid response to the [Set Addressed Player]
        command sent by the PTS.

        """
        return "OK"
 No newline at end of file
+31 −3
Original line number Diff line number Diff line
@@ -24,6 +24,17 @@ NEEDS_PAIRING_CONFIRMATION = {
    "SM/CEN/EKS/BV-01-C",
    "SM/CEN/JW/BI-04-C",
    "SM/CEN/JW/BI-01-C",
    "SM/CEN/KDU/BV-04-C",
    "SM/CEN/KDU/BV-05-C",
    "SM/CEN/KDU/BV-06-C",
    "SM/CEN/KDU/BV-10-C",
    "SM/CEN/KDU/BV-11-C",
}

ACCEPTS_REMOTE_PAIRING_CONFIRMATION = {
    "SM/CEN/KDU/BI-01-C",
    "SM/CEN/KDU/BI-02-C",
    "SM/CEN/KDU/BI-03-C",
}


@@ -36,11 +47,13 @@ class SMProxy(ProfileProxy):
        self.connection = None

    @assert_description
    def MMI_IUT_ENABLE_CONNECTION_SM(self, pts_addr: bytes, **kwargs):
    def MMI_IUT_ENABLE_CONNECTION_SM(self, test, pts_addr: bytes, **kwargs):
        """
        Initiate an connection from the IUT to the PTS.
        """
        self.connection = self.host.ConnectLE(address=pts_addr).connection
        if self.connection and test in ACCEPTS_REMOTE_PAIRING_CONFIRMATION:
            self.sm.ProvidePairingConfirmation(connection=self.connection, pairing_confirmation_value=True)
        return "OK"

    @assert_description
@@ -68,3 +81,18 @@ class SMProxy(ProfileProxy):
            self.host.DisconnectLE(connection=self.connection)
            self.connection = None
        return "OK"

    def MMI_LESC_NUMERIC_COMPARISON(self, **kwargs):
        """
        Please confirm the following number matches IUT: 385874.
        """

        return "OK"

    @assert_description
    def MMI_ASK_IUT_PERFORM_RESET(self, **kwargs):
        """
        Please reset your device.
        """
        self.host.Reset()
        return "OK"
+3 −0
Original line number Diff line number Diff line
@@ -134,6 +134,7 @@ genrule {
         "    $(in)",
    srcs: [
        "proto/pandora/a2dp.proto",
        "proto/pandora/avrcp.proto",
        "proto/pandora/gatt.proto",
        "proto/pandora/hfp.proto",
        "proto/pandora/host.proto",
@@ -142,6 +143,8 @@ genrule {
    out: [
        "pandora/a2dp_grpc.py",
        "pandora/a2dp_pb2.py",
        "pandora/avrcp_grpc.py",
        "pandora/avrcp_pb2.py",
        "pandora/gatt_grpc.py",
        "pandora/gatt_pb2.py",
        "pandora/hfp_grpc.py",
+3 −0
Original line number Diff line number Diff line
@@ -22,6 +22,7 @@
        <option name="physical" value="false" />
        <option name="profile" value="A2DP/SRC" />
        <option name="profile" value="AVDTP/SRC" />
        <option name="profile" value="AVRCP" />
        <option name="profile" value="GATT/CL/GAC" />
        <option name="profile" value="GATT/CL/GAD" />
        <option name="profile" value="HFP/AG/DIS" />
@@ -31,5 +32,7 @@
        <option name="profile" value="SDP/SR" />
        <option name="profile" value="SM/CEN/EKS" />
        <option name="profile" value="SM/CEN/JW" />
        <option name="profile" value="SM/CEN/KDU" />

    </test>
</configuration>
Loading