#!/usr/bin/env python3 # # Copyright (C) 2020 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. """ Setup: This test only requires two fuchsia devices. """ from acts import signals from acts.base_test import BaseTestClass from acts.test_decorators import test_tracker_info from acts_contrib.test_utils.bt.bt_test_utils import generate_id_by_size from acts_contrib.test_utils.fuchsia.bt_test_utils import bredr_scan_for_device_by_name from acts_contrib.test_utils.fuchsia.bt_test_utils import le_scan_for_device_by_name from acts_contrib.test_utils.fuchsia.bt_test_utils import unbond_all_known_devices from acts_contrib.test_utils.fuchsia.bt_test_utils import verify_device_state_by_name import time class BtFuchsiaEPTest(BaseTestClass): ble_advertise_interval = 50 scan_timeout_seconds = 60 default_iterations = 1000 adv_name = generate_id_by_size(10) test_adv_data = { "name": adv_name, "appearance": None, "service_data": None, "tx_power_level": None, "service_uuids": None, "manufacturer_data": None, "uris": None, } test_connectable = True test_scan_response = None def setup_class(self): super().setup_class() for fd in self.fuchsia_devices: fd.bts_lib.initBluetoothSys() self.pri_dut = self.fuchsia_devices[0] self.sec_dut = self.fuchsia_devices[1] def on_fail(self, test_name, begin_time): for fd in self.fuchsia_devices: fd.take_bug_report(test_name, begin_time) self._unbond_all_known_devices() self.sec_dut.ble_lib.bleStopBleAdvertising() self._kill_media_services() def teardown_class(self): self._kill_media_services() def _kill_media_services(self): """Kill any BT services related to A2DP/AVRCP on all Fuchsia devices. """ ssh_timeout = 30 for fd in self.fuchsia_devices: fd.send_command_ssh("killall bt-a2dp*", timeout=ssh_timeout, skip_status_code_check=True) fd.send_command_ssh("killall bt-avrcp*", timeout=ssh_timeout, skip_status_code_check=True) def _unbond_all_known_devices(self): """For all Fuchsia devices, unbond any known pairings. """ time.sleep(5) for fd in self.fuchsia_devices: unbond_all_known_devices(fd, self.log) def test_ble_awareness(self): """Verify that Fuchsia devices can advertise and scan each other Verify a Fuchsia device that starts a BLE advertisesement can be found by a Fuchsia BLE scanner. Steps: 1. On one Fuchsia device set an advertisement 2. On one Fuchsia device, scan for the advertisement by name Expected Result: Verify that there are no errors after each GATT connection. Returns: signals.TestPass if no errors signals.TestFailure if there are any errors during the test. TAGS: BLE Priority: 0 """ self.sec_dut.ble_lib.bleStartBleAdvertising( self.test_adv_data, self.test_scan_response, self.ble_advertise_interval, self.test_connectable) device = le_scan_for_device_by_name(self.pri_dut, self.log, self.adv_name, self.scan_timeout_seconds) self.sec_dut.ble_lib.bleStopBleAdvertising() if device is None: raise signals.TestFailure("Scanner unable to find advertisement.") raise signals.TestPass("Success") def test_gatt_central_peripheral(self): """Verify that Fuchsia devices can perform GATT operations Verify a Fuchsia devices can perform GATT connections and interactions. Steps: 1. On one Fuchsia device set an advertisement 2. On one Fuchsia device, scan for the advertisement by name 3. Perform GATT connection over LE 4. Pair both devices. 5. Perform GATT read/write operations. 6. Perform GATT disconnection. Expected Result: Verify that there are no errors after each GATT connection. Returns: signals.TestPass if no errors signals.TestFailure if there are any errors during the test. TAGS: BLE Priority: 0 """ self._unbond_all_known_devices() source_device_name = generate_id_by_size(10) self.pri_dut.bts_lib.setName(source_device_name) self.sec_dut.ble_lib.bleStartBleAdvertising( self.test_adv_data, self.test_scan_response, self.ble_advertise_interval, self.test_connectable) device = le_scan_for_device_by_name(self.pri_dut, self.log, self.adv_name, self.scan_timeout_seconds) if device is None: raise signals.TestFailure("Scanner unable to find advertisement.") connect_result = self.pri_dut.gattc_lib.bleConnectToPeripheral( device["id"]) if connect_result.get("error") is not None: raise signals.TestFailure("GATT Connection failed with: {}".format( connect_result.get("error"))) if not verify_device_state_by_name(self.pri_dut, self.log, self.adv_name, "CONNECTED", None): raise signals.TestFailure( "Failed to connect to device {}.".format(target_device_name)) if not verify_device_state_by_name( self.sec_dut, self.log, source_device_name, "CONNECTED", None): raise signals.TestFailure( "Failed to connect to device {}.".format(source_device_name)) security_level = "ENCRYPTED" non_bondable = False transport = 2 #LE self.pri_dut.bts_lib.pair(device["id"], security_level, non_bondable, transport) services = None if not verify_device_state_by_name(self.pri_dut, self.log, self.adv_name, "BONDED", services): raise signals.TestFailure( "Failed to pair device {}.".format(target_device_name)) if not verify_device_state_by_name(self.sec_dut, self.log, source_device_name, "BONDED", services): raise signals.TestFailure( "Failed to pair device {}.".format(source_device_name)) disconnect_result = self.pri_dut.gattc_lib.bleDisconnectPeripheral( device["id"]) if disconnect_result.get("error") is not None: raise signals.TestFailure( "GATT Disconnection failed with: {}".format( connect_result.get("error"))) self.sec_dut.ble_lib.bleStopBleAdvertising() # TODO: Setup Proper GATT server and verify services published are found raise signals.TestPass("Success") def test_pairing_a2dp(self): """Verify that Fuchsia devices can pair to each other and establish an A2DP connection Verify that Fuchsia devices can pair to each other and establish an A2DP connection Steps: 1. Clear out all bonded devices 2. Stop any A2DP services running on the device Needed to take ownership of the services 3. Init sink and source opposite devices 4. Start pairing delegate for all Fuchsia devices 5. Set sink device to be discoverable 6. Discover sink device from source device 7. Connect to sink device from source device 8. Pair to sink device 9. Validate paired devices and services present Expected Result: Verify devices are successfully paired and appropriate a2dp services are running. Returns: signals.TestPass if no errors signals.TestFailure if there are any errors during the test. TAGS: BREDR, A2DP Priority: 0 """ self._unbond_all_known_devices() self._kill_media_services() source_device_name = generate_id_by_size(10) target_device_name = generate_id_by_size(10) self.pri_dut.bts_lib.setName(source_device_name) self.sec_dut.bts_lib.setName(target_device_name) input_capabilities = "NONE" output_capabilities = "NONE" # Initialize a2dp on both devices. self.pri_dut.avdtp_lib.init() self.sec_dut.avdtp_lib.init() self.pri_dut.bts_lib.acceptPairing(input_capabilities, output_capabilities) self.sec_dut.bts_lib.acceptPairing(input_capabilities, output_capabilities) self.sec_dut.bts_lib.setDiscoverable(True) unique_mac_addr_id = bredr_scan_for_device_by_name( self.pri_dut, self.log, target_device_name, self.scan_timeout_seconds) if not unique_mac_addr_id: raise signals.TestFailure( "Failed to find device {}.".format(target_device_name)) connect_result = self.pri_dut.bts_lib.connectDevice(unique_mac_addr_id) if connect_result.get("error") is not None: raise signals.TestFailure("Failed to connect with {}.".format( connect_result.get("error"))) # We pair before checking the CONNECTED status because BR/EDR semantics # were recently changed such that if pairing is not confirmed, then bt # does not report connected = True. security_level = "NONE" bondable = True transport = 1 #BREDR pair_result = self.pri_dut.bts_lib.pair(unique_mac_addr_id, security_level, bondable, transport) if pair_result.get("error") is not None: raise signals.TestFailure("Failed to pair with {}.".format( pair_result.get("error"))) if not verify_device_state_by_name( self.pri_dut, self.log, target_device_name, "CONNECTED", None): raise signals.TestFailure( "Failed to connect to device {}.".format(target_device_name)) if not verify_device_state_by_name( self.sec_dut, self.log, source_device_name, "CONNECTED", None): raise signals.TestFailure( "Failed to connect to device {}.".format(source_device_name)) #TODO: Validation of services and paired devices (b/175641870) # A2DP sink: 0000110b-0000-1000-8000-00805f9b34fb # A2DP source: 0000110a-0000-1000-8000-00805f9b34fb #TODO: Make an easy function for checking/updating devices services = None if not verify_device_state_by_name(self.pri_dut, self.log, target_device_name, "BONDED", services): raise signals.TestFailure( "Failed to pair device {}.".format(target_device_name)) if not verify_device_state_by_name(self.sec_dut, self.log, source_device_name, "BONDED", services): raise signals.TestFailure( "Failed to pair device {}.".format(source_device_name)) raise signals.TestPass("Success")