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

Commit 43ab3cae authored by Krzysztof Kopyściński's avatar Krzysztof Kopyściński
Browse files

pandora: add support to VCP tests 

Implements VCP tests that are supported by bluetooth stack. Tests
that are skipped require mechanism to trigger sending write to Volume
Control Point with "Unmute" OP code

Bug: 369938109
Flag: TEST_ONLY
Test: atest pts-bot:VCP -v
Change-Id: I1fa01b7e184abe164da0420a70c46b4c1c2fb75d
parent 41e81e10

android/pandora/mmi2grpc/mmi2grpc/vcp.py

+70 −6
Original line number Diff line number Diff line
@@ -17,23 +17,33 @@ import threading 
from mmi2grpc._helpers import assert_description, match_description

from mmi2grpc._proxy import ProfileProxy

from mmi2grpc._rootcanal import Dongle



from pandora_experimental.vcp_grpc import VCP

from pandora_experimental.gatt_grpc import GATT

from pandora.security_grpc import Security, SecurityStorage

from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer

from pandora.host_grpc import Host

from pandora.host_pb2 import PUBLIC, RANDOM

from pandora.security_grpc import Security

from pandora.security_pb2 import LE_LEVEL3, PairingEventAnswer

from pandora_experimental.le_audio_grpc import LeAudio



from time import sleep





class VCPProxy(ProfileProxy):



    def __init__(self, channel, rootcanal):

        super().__init__(channel)

        self.vcp = VCP(channel)

        self.gatt = GATT(channel)

        self.security_storage = SecurityStorage(channel)

        self.host = Host(channel)

        self.security = Security(channel)

        self.le_audio = LeAudio(channel)

        self.rootcanal = rootcanal

        self.connection = None

        self.pairing_stream = None

        self.pairing_stream = self.security.OnPairing()



    def test_started(self, test: str, description: str, pts_addr: bytes):

        self.rootcanal.select_pts_dongle(Dongle.LAIRD_BL654)
@@ -49,13 +59,18 @@ class VCPProxy(ProfileProxy): 
        the Implementation Under Test (IUT) can initiate a GATT connect request

        to the PTS.

        """

        self.security_storage.DeleteBond(public=pts_addr)

        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)



        def vcp_connect():

            self.vcp.WaitConnect(connection=self.connection)



        threading.Thread(target=secure).start()

        threading.Thread(target=vcp_connect).start()



        return "OK"



    @match_description
@@ -84,6 +99,10 @@ class VCPProxy(ProfileProxy): 
        Description: Verify that the Implementation Under Test \(IUT\) can send

        Discover All Characteristics command.

        """

        # PTS expects us to do discovery after bonding, but in fact Android does it as soon as

        # encryption is completed. Invalidate GATT cache so the discovery takes place again

        self.gatt.ClearCache(connection=self.connection)



        return "OK"



    @match_description
@@ -92,13 +111,26 @@ class VCPProxy(ProfileProxy): 
        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})).

        """

        # After discovery Android reads these values by itself, after profile connection.

        # Although, for some tests, this is used as validation, for example for tests with invalid

        # behavior (BI tests). Just send GATT read to sattisfy this conditions, as VCP has no exposed

        # (or even existing, native) interface to trigger read on demand.

        def read():

            nonlocal handle

            self.gatt.ReadCharacteristicFromHandle(\

                    connection=self.connection, handle=int(handle, base=16))



        worker = threading.Thread(target=read)

        worker.start()

        worker.join(timeout=30)



        return "OK"



    @assert_description

    def USER_CONFIRM_SUPPORTED_CHARACTERISTIC(self, characteristics: str, **kwargs):

    @match_description

    def USER_CONFIRM_SUPPORTED_CHARACTERISTIC(self, body: str, **kwargs):

        """

        Please verify that for each supported characteristic, attribute

        handle/UUID pair(s) is returned to the upper tester.(?P<characteristics>(.|\n)*)

        handle/UUID pair\(s\) is returned to the (.*)\.(?P<body>.*)

        """



        return "OK"
@@ -109,6 +141,38 @@ class VCPProxy(ProfileProxy): 
        Please write to Client Characteristic Configuration Descriptor of

        (?P<name>(Volume State|Offset State)) characteristic to enable notification.

        """



        # After discovery Android subscribes by itself, after profile connection

        return "OK"



    def IUT_SEND_WRITE_REQUEST(self, description: str, **kwargs):

        r"""

        Please send write request to handle 0xXXXX with following value.

        Characteristic name:

            Op Code: [X (0xXX)] Op code name

            Change Counter: <WildCard: Exists>

            Value: <WildCard: Exists>

        """



        # Wait a couple seconds so the VCP is ready (subscriptions and reads are completed)

        sleep(2)



        if ("Set Absolute Volume" in description):

            self.vcp.SetDeviceVolume(connection=self.connection, volume=42)

        elif ("Unmute" in description):

            # for now, there is no way to trigger this, and tests are skipped

            return "No"

        elif ("Set Volume Offset" in description):

            self.vcp.SetVolumeOffset(connection=self.connection, offset=42)

        elif ("Volume Control Point" in description and

              "Op Code: <WildCard: Exists>" in description):

            # Handles sending *any* OP Code on Volume Control Point

            self.vcp.SetDeviceVolume(connection=self.connection, volume=42)

        elif ("Volume Offset Control Point" in description and

              "Op Code: <WildCard: Exists>" in description):

            self.vcp.SetVolumeOffset(connection=self.connection, offset=42)





        return "OK"



    @assert_description

android/pandora/server/configs/pts_bot_tests_config.json

+16 −15
Original line number Diff line number Diff line
@@ -686,7 +686,19 @@ 
    "SM/PER/SCJW/BV-03-C",

    "SM/PER/SCPK/BI-03-C",

    "SM/PER/SCPK/BV-02-C",

    "VCP/VC/CGGIT/CHA/BV-06-C"

    "VCP/VC/CGGIT/CHA/BV-01-C",

    "VCP/VC/CGGIT/CHA/BV-02-C",

    "VCP/VC/CGGIT/CHA/BV-03-C",

    "VCP/VC/CGGIT/CHA/BV-04-C",

    "VCP/VC/CGGIT/CHA/BV-06-C",

    "VCP/VC/CGGIT/SER/BV-01-C",

    "VCP/VC/CGGIT/SER/BV-02-C",

    "VCP/VC/SPE/BI-05-C",

    "VCP/VC/SPE/BI-13-C",

    "VCP/VC/SPE/BI-15-C",

    "VCP/VC/SPE/BI-16-C",

    "VCP/VC/VCCP/BV-05-C",

    "VCP/VC/VOCP/BV-01-C"

  ],

  "flaky": [

    "A2DP/SRC/SUS/BV-02-I",
@@ -1145,20 +1157,8 @@ 
    "SM/PER/SCPK/BV-03-C",

    "SM/PER/SIE/BV-01-C",

    "SM/PER/SIP/BV-01-C",

    "VCP/VC/CGGIT/CHA/BV-01-C",

    "VCP/VC/CGGIT/CHA/BV-02-C",

    "VCP/VC/CGGIT/CHA/BV-03-C",

    "VCP/VC/CGGIT/CHA/BV-04-C",

    "VCP/VC/CGGIT/SER/BV-01-C",

    "VCP/VC/CGGIT/SER/BV-02-C",

    "VCP/VC/SPE/BI-05-C",

    "VCP/VC/SPE/BI-06-C",

    "VCP/VC/SPE/BI-13-C",

    "VCP/VC/SPE/BI-15-C",

    "VCP/VC/SPE/BI-16-C",

    "VCP/VC/VCCP/BV-05-C",

    "VCP/VC/VCCP/BV-06-C",

    "VCP/VC/VOCP/BV-01-C"

    "VCP/VC/VCCP/BV-06-C"

  ],

  "ics": {

    "TSPC_4.0HCI_1_1": true,
@@ -3168,7 +3168,8 @@ 
    "SM": {},

    "SPP": {},

    "SUM ICS": {},

    "VCP": {}

    "VCP": {

    }

  },

  "flags": [

    {

android/pandora/server/src/Server.kt

+1 −0
Original line number Diff line number Diff line
@@ -68,6 +68,7 @@ class Server(context: Context) { 
                        BluetoothProfile.OPP to ::Opp,

                        BluetoothProfile.MAP to ::Map,

                        BluetoothProfile.LE_AUDIO to ::LeAudio,

                        BluetoothProfile.VOLUME_CONTROL to ::Vcp,

                    )

                    .filter { bluetoothAdapter.isEnabled }

                    .filter { bluetoothAdapter.getSupportedProfiles().contains(it.key) == true }

android/pandora/server/src/Vcp.kt

0 → 100644
+132 −0
Original line number Diff line number Diff line 
/*

 * Copyright (C) 2024 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.

 */



package com.android.pandora



import android.bluetooth.BluetoothAdapter

import android.bluetooth.BluetoothDevice

import android.bluetooth.BluetoothDevice.TRANSPORT_LE

import android.bluetooth.BluetoothVolumeControl

import android.bluetooth.BluetoothManager

import android.bluetooth.BluetoothProfile

import android.bluetooth.BluetoothProfile.CONNECTION_POLICY_ALLOWED

import android.content.Context

import android.content.Intent

import android.content.IntentFilter

import android.util.Log

import com.google.protobuf.Empty

import io.grpc.Status

import io.grpc.stub.StreamObserver

import java.io.Closeable

import java.io.PrintWriter

import java.io.StringWriter

import java.util.concurrent.Executors

import kotlinx.coroutines.CoroutineScope

import kotlinx.coroutines.Dispatchers

import kotlinx.coroutines.cancel

import kotlinx.coroutines.channels.awaitClose

import kotlinx.coroutines.flow.Flow

import kotlinx.coroutines.flow.SharingStarted

import kotlinx.coroutines.flow.callbackFlow

import kotlinx.coroutines.flow.filter

import kotlinx.coroutines.flow.first

import kotlinx.coroutines.flow.map

import kotlinx.coroutines.flow.shareIn

import pandora.vcp.VCPGrpc.VCPImplBase

import pandora.vcp.VcpProto.*

import pandora.HostProto.Connection



@kotlinx.coroutines.ExperimentalCoroutinesApi

class Vcp(val context: Context) : VCPImplBase(), Closeable {

    private val TAG = "PandoraVcp"



    private val scope: CoroutineScope = CoroutineScope(Dispatchers.Default.limitedParallelism(1))



    private val bluetoothManager = context.getSystemService(BluetoothManager::class.java)!!

    private val bluetoothAdapter = bluetoothManager.adapter



    private val bluetoothVolumeControl =

        getProfileProxy<BluetoothVolumeControl>(context, BluetoothProfile.VOLUME_CONTROL)



    private val flow =

        intentFlow(

                context,

                IntentFilter().apply {

                    addAction(BluetoothVolumeControl.ACTION_CONNECTION_STATE_CHANGED)

                },

                scope,

            )

            .shareIn(scope, SharingStarted.Eagerly)



    override fun close() {

        // Deinit the CoroutineScope

        scope.cancel()

    }



    override fun setDeviceVolume(

        request: SetDeviceVolumeRequest,

        responseObserver: StreamObserver<Empty>

    ) {

        grpcUnary<Empty>(scope, responseObserver) {

            val device = request.connection.toBluetoothDevice(bluetoothAdapter)



            Log.i(TAG, "setDeviceVolume(${device}, ${request.volume})")



            bluetoothVolumeControl.setDeviceVolume(device, request.volume, false)



            Empty.getDefaultInstance()

        }

    }



    override fun setVolumeOffset(

        request: SetVolumeOffsetRequest,

        responseObserver: StreamObserver<Empty>

    ) {

        grpcUnary<Empty>(scope, responseObserver) {

            val device = request.connection.toBluetoothDevice(bluetoothAdapter)



            Log.i(TAG, "setVolumeOffset(${device}, ${request.offset})")



            bluetoothVolumeControl.setVolumeOffset(device, 1, request.offset)



            Empty.getDefaultInstance()

        }

    }



    override fun waitConnect(

        request: WaitConnectRequest,

        responseObserver: StreamObserver<Empty>

    ) {

        grpcUnary<Empty>(scope, responseObserver) {

            val device = request.connection.toBluetoothDevice(bluetoothAdapter)

            Log.i(TAG, "waitPeripheral(${device}")

            if (

                bluetoothVolumeControl.getConnectionState(device) != BluetoothProfile.STATE_CONNECTED

            ) {

                Log.d(TAG, "Manual call to setConnectionPolicy")

                bluetoothVolumeControl.setConnectionPolicy(device, CONNECTION_POLICY_ALLOWED)

                Log.d(TAG, "wait for bluetoothVolumeControl profile connection")

                flow

                    .filter { it.getBluetoothDeviceExtra() == device }

                    .map { it.getIntExtra(BluetoothProfile.EXTRA_STATE, BluetoothAdapter.ERROR) }

                    .filter { it == BluetoothProfile.STATE_CONNECTED }

                    .first()

            }



            Empty.getDefaultInstance()

        }

    }

}
 
 No newline at end of file

pandora/interfaces/pandora_experimental/vcp.proto

0 → 100644
+47 −0
Original line number Diff line number Diff line 
// Copyright (C) 2024 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.



syntax = "proto3";



package pandora.vcp;



import "pandora/host.proto";

option java_outer_classname = "VcpProto";

import "google/protobuf/empty.proto";



service VCP {

  // set absolute volume on remote device

  rpc SetDeviceVolume(SetDeviceVolumeRequest) returns (google.protobuf.Empty);

  // set volume offset on remote device

  rpc SetVolumeOffset(SetVolumeOffsetRequest) returns (google.protobuf.Empty);

  // Wait for device to be connected.

  rpc WaitConnect(WaitConnectRequest) returns (google.protobuf.Empty);

}



// Request of the `SetDeviceVolume` method

message SetDeviceVolumeRequest{

  // Connection crafted by grpc server

  Connection connection = 1;

  // Volume value to be set

  int32 volume = 2;

}



// Request of the `SetVolumeOffset` method

message SetVolumeOffsetRequest{

  // Connection crafted by grpc server

  Connection connection = 1;

  // Volume offset value to be set

  int32 offset = 2;

}



message WaitConnectRequest {

  Connection connection = 1;

}