| #!/usr/bin/env python3 |
| # |
| # Copyright 2019 - 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. |
| |
| import os |
| import sys |
| import logging |
| |
| from bluetooth_packets_python3 import hci_packets |
| from cert.event_stream import EventStream |
| from cert.closable import Closable |
| from cert.closable import safeClose |
| from cert.matchers import AdvertisingMatchers |
| from cert.py_hci import PyHci |
| from cert.truth import assertThat |
| from facade import common_pb2 as common |
| from facade import rootservice_pb2 as facade_rootservice |
| from google.protobuf import empty_pb2 as empty_proto |
| from hci.facade import hci_facade_pb2 as hci_facade |
| from hci.facade import \ |
| le_advertising_manager_facade_pb2 as le_advertising_facade |
| from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade |
| from hci.facade.le_advertising_manager_facade_pb2 import AdvertisingCallbackMsgType |
| from hci.facade.le_advertising_manager_facade_pb2 import AdvertisingStatus |
| |
| |
| class LeAdvertisingManagerTestBase(): |
| |
| def setup_test(self, cert): |
| self.cert_hci = PyHci(cert, acl_streaming=True) |
| self.dut.callback_event_stream = EventStream( |
| self.dut.hci_le_advertising_manager.FetchCallbackEvents(empty_proto.Empty())) |
| self.dut.address_event_stream = EventStream( |
| self.dut.hci_le_advertising_manager.FetchAddressEvents(empty_proto.Empty())) |
| |
| def teardown_test(self): |
| self.cert_hci.close() |
| if self.dut.callback_event_stream is not None: |
| safeClose(self.dut.callback_event_stream) |
| else: |
| logging.info("DUT: Callback Event Stream is None!") |
| if self.dut.address_event_stream is not None: |
| safeClose(self.dut.address_event_stream) |
| else: |
| logging.info("DUT: address Event Stream is None!") |
| |
| def set_address_policy_with_static_address(self): |
| privacy_policy = le_initiator_address_facade.PrivacyPolicy( |
| address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS, |
| address_with_type=common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=bytes(b'd0:05:04:03:02:01')), |
| type=common.RANDOM_DEVICE_ADDRESS), |
| rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', |
| minimum_rotation_time=0, |
| maximum_rotation_time=0) |
| self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(privacy_policy) |
| |
| def create_advertiser(self): |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_The_DUT')) |
| gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize())) |
| config = le_advertising_facade.AdvertisingConfig( |
| advertisement=[gap_data], |
| interval_min=512, |
| interval_max=768, |
| advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND, |
| own_address_type=common.USE_RANDOM_DEVICE_ADDRESS, |
| channel_map=7, |
| filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES) |
| request = le_advertising_facade.CreateAdvertiserRequest(config=config) |
| create_response = self.dut.hci_le_advertising_manager.CreateAdvertiser(request) |
| return create_response |
| |
| def test_le_ad_scan_dut_advertises(self): |
| self.set_address_policy_with_static_address() |
| self.cert_hci.register_for_le_events(hci_packets.SubeventCode.ADVERTISING_REPORT, |
| hci_packets.SubeventCode.EXTENDED_ADVERTISING_REPORT) |
| |
| # CERT Scans |
| self.cert_hci.send_command(hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01')) |
| scan_parameters = hci_packets.PhyScanParameters() |
| scan_parameters.le_scan_type = hci_packets.LeScanType.ACTIVE |
| scan_parameters.le_scan_interval = 40 |
| scan_parameters.le_scan_window = 20 |
| self.cert_hci.send_command( |
| hci_packets.LeSetExtendedScanParametersBuilder(hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.LeScanningFilterPolicy.ACCEPT_ALL, 1, |
| [scan_parameters])) |
| self.cert_hci.send_command( |
| hci_packets.LeSetExtendedScanEnableBuilder(hci_packets.Enable.ENABLED, |
| hci_packets.FilterDuplicates.DISABLED, 0, 0)) |
| |
| create_response = self.create_advertiser() |
| |
| assertThat(self.cert_hci.get_le_event_stream()).emits(lambda packet: b'Im_The_DUT' in packet.payload) |
| |
| remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id) |
| self.dut.hci_le_advertising_manager.RemoveAdvertiser(remove_request) |
| self.cert_hci.send_command( |
| hci_packets.LeSetScanEnableBuilder(hci_packets.Enable.DISABLED, hci_packets.Enable.DISABLED)) |
| |
| def test_extended_create_advertises(self): |
| self.set_address_policy_with_static_address() |
| self.cert_hci.register_for_le_events(hci_packets.SubeventCode.ADVERTISING_REPORT, |
| hci_packets.SubeventCode.EXTENDED_ADVERTISING_REPORT) |
| |
| # CERT Scans |
| self.cert_hci.send_command(hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01')) |
| scan_parameters = hci_packets.PhyScanParameters() |
| scan_parameters.le_scan_type = hci_packets.LeScanType.ACTIVE |
| scan_parameters.le_scan_interval = 40 |
| scan_parameters.le_scan_window = 20 |
| self.cert_hci.send_command( |
| hci_packets.LeSetExtendedScanParametersBuilder(hci_packets.OwnAddressType.RANDOM_DEVICE_ADDRESS, |
| hci_packets.LeScanningFilterPolicy.ACCEPT_ALL, 1, |
| [scan_parameters])) |
| self.cert_hci.send_command( |
| hci_packets.LeSetExtendedScanEnableBuilder(hci_packets.Enable.ENABLED, |
| hci_packets.FilterDuplicates.DISABLED, 0, 0)) |
| |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_The_DUT')) |
| gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize())) |
| config = le_advertising_facade.AdvertisingConfig( |
| advertisement=[gap_data], |
| interval_min=512, |
| interval_max=768, |
| advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND, |
| own_address_type=common.USE_RANDOM_DEVICE_ADDRESS, |
| channel_map=7, |
| filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES) |
| extended_config = le_advertising_facade.ExtendedAdvertisingConfig( |
| advertising_config=config, |
| connectable=True, |
| scannable=False, |
| directed=False, |
| high_duty_directed_connectable=False, |
| legacy_pdus=True, |
| anonymous=False, |
| include_tx_power=True, |
| use_le_coded_phy=False, |
| secondary_max_skip=0x00, |
| secondary_advertising_phy=0x01, |
| sid=0x00, |
| enable_scan_request_notifications=0x00) |
| request = le_advertising_facade.ExtendedCreateAdvertiserRequest(config=extended_config) |
| create_response = self.dut.hci_le_advertising_manager.ExtendedCreateAdvertiser(request) |
| |
| assertThat(self.cert_hci.get_le_event_stream()).emits(lambda packet: b'Im_The_DUT' in packet.payload) |
| |
| remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id) |
| self.dut.hci_le_advertising_manager.RemoveAdvertiser(remove_request) |
| self.cert_hci.send_command( |
| hci_packets.LeSetScanEnableBuilder(hci_packets.Enable.DISABLED, hci_packets.Enable.DISABLED)) |
| |
| def test_advertising_set_started_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_SET_STARTED, |
| create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS, |
| 0x00)) |
| |
| def test_enable_advertiser_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| enable_advertiser_request = le_advertising_facade.EnableAdvertiserRequest( |
| advertiser_id=create_response.advertiser_id, enable=True) |
| self.dut.hci_le_advertising_manager.EnableAdvertiser(enable_advertiser_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_ENABLED, |
| create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS, |
| 0x01)) |
| |
| def test_disable_advertiser_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| disable_advertiser_request = le_advertising_facade.EnableAdvertiserRequest( |
| advertiser_id=create_response.advertiser_id, enable=False) |
| self.dut.hci_le_advertising_manager.EnableAdvertiser(disable_advertiser_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_ENABLED, |
| create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS, |
| 0x00)) |
| |
| def test_set_advertising_data_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_The_DUT2')) |
| gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize())) |
| |
| set_data_request = le_advertising_facade.SetDataRequest( |
| advertiser_id=create_response.advertiser_id, set_scan_rsp=False, data=[gap_data]) |
| self.dut.hci_le_advertising_manager.SetData(set_data_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_DATA_SET, |
| create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS)) |
| |
| def test_set_scan_response_data_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_The_DUT2')) |
| gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize())) |
| |
| set_data_request = le_advertising_facade.SetDataRequest( |
| advertiser_id=create_response.advertiser_id, set_scan_rsp=True, data=[gap_data]) |
| self.dut.hci_le_advertising_manager.SetData(set_data_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.SCAN_RESPONSE_DATA_SET, |
| create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS)) |
| |
| def test_set_parameters_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| |
| # The Host shall not issue set parameters command when advertising is enabled |
| disable_advertiser_request = le_advertising_facade.EnableAdvertiserRequest( |
| advertiser_id=create_response.advertiser_id, enable=False) |
| self.dut.hci_le_advertising_manager.EnableAdvertiser(disable_advertiser_request) |
| |
| config = le_advertising_facade.AdvertisingConfig( |
| interval_min=512, |
| interval_max=768, |
| advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND, |
| own_address_type=common.USE_RANDOM_DEVICE_ADDRESS, |
| channel_map=7, |
| filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES) |
| |
| set_parameters_request = le_advertising_facade.SetParametersRequest( |
| advertiser_id=create_response.advertiser_id, config=config) |
| self.dut.hci_le_advertising_manager.SetParameters(set_parameters_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.ADVERTISING_PARAMETERS_UPDATED, |
| create_response.advertiser_id, AdvertisingStatus.ADV_SUCCESS)) |
| |
| def test_set_periodic_parameters_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| |
| config = le_advertising_facade.PeriodicAdvertisingParameters( |
| min_interval=512, |
| max_interval=768, |
| advertising_property=le_advertising_facade.AdvertisingProperty.INCLUDE_TX_POWER) |
| |
| set_periodic_parameters_request = le_advertising_facade.SetPeriodicParametersRequest( |
| advertiser_id=create_response.advertiser_id, config=config) |
| self.dut.hci_le_advertising_manager.SetPeriodicParameters(set_periodic_parameters_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg( |
| AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_PARAMETERS_UPDATED, create_response.advertiser_id)) |
| |
| def test_set_periodic_data_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| gap_name = hci_packets.GapData() |
| gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME |
| gap_name.data = list(bytes(b'Im_The_DUT2')) |
| gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize())) |
| |
| set_periodic_data_request = le_advertising_facade.SetPeriodicDataRequest( |
| advertiser_id=create_response.advertiser_id, data=[gap_data]) |
| self.dut.hci_le_advertising_manager.SetPeriodicData(set_periodic_data_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_DATA_SET, |
| create_response.advertiser_id)) |
| |
| def test_enable_periodic_advertising_callback(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| enable_periodic_advertising_request = le_advertising_facade.EnablePeriodicAdvertisingRequest( |
| advertiser_id=create_response.advertiser_id, enable=True) |
| self.dut.hci_le_advertising_manager.EnablePeriodicAdvertising(enable_periodic_advertising_request) |
| |
| assertThat(self.dut.callback_event_stream).emits( |
| AdvertisingMatchers.AdvertisingCallbackMsg(AdvertisingCallbackMsgType.PERIODIC_ADVERTISING_ENABLED, |
| create_response.advertiser_id)) |
| |
| def test_get_own_address(self): |
| self.set_address_policy_with_static_address() |
| create_response = self.create_advertiser() |
| get_own_address_request = le_advertising_facade.GetOwnAddressRequest( |
| advertiser_id=create_response.advertiser_id) |
| self.dut.hci_le_advertising_manager.GetOwnAddress(get_own_address_request) |
| address_with_type = common.BluetoothAddressWithType( |
| address=common.BluetoothAddress(address=bytes(b'd0:05:04:03:02:01')), type=common.RANDOM_DEVICE_ADDRESS) |
| assertThat(self.dut.address_event_stream).emits( |
| AdvertisingMatchers.AddressMsg(AdvertisingCallbackMsgType.OWN_ADDRESS_READ, create_response.advertiser_id, |
| address_with_type)) |
