1#!/usr/bin/env python3 2# 3# Copyright 2019 - The Android Open Source Project 4# 5# Licensed under the Apache License, Version 2.0 (the "License"); 6# you may not use this file except in compliance with the License. 7# You may obtain a copy of the License at 8# 9# http://www.apache.org/licenses/LICENSE-2.0 10# 11# Unless required by applicable law or agreed to in writing, software 12# distributed under the License is distributed on an "AS IS" BASIS, 13# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14# See the License for the specific language governing permissions and 15# limitations under the License. 16 17import os 18import sys 19import logging 20 21from cert.gd_base_test_facade_only import GdFacadeOnlyBaseTestClass 22from cert.event_callback_stream import EventCallbackStream 23from cert.event_asserts import EventAsserts 24from google.protobuf import empty_pb2 as empty_proto 25from facade import rootservice_pb2 as facade_rootservice 26from hci.facade import facade_pb2 as hci_facade 27from hci.facade import \ 28 le_advertising_manager_facade_pb2 as le_advertising_facade 29from bluetooth_packets_python3 import hci_packets 30from facade import common_pb2 as common 31 32 33class LeAdvertisingManagerTest(GdFacadeOnlyBaseTestClass): 34 35 def setup_test(self): 36 self.device_under_test.rootservice.StartStack( 37 facade_rootservice.StartStackRequest( 38 module_under_test=facade_rootservice.BluetoothModule.Value( 39 'HCI_INTERFACES'),)) 40 self.cert_device.rootservice.StartStack( 41 facade_rootservice.StartStackRequest( 42 module_under_test=facade_rootservice.BluetoothModule.Value( 43 'HCI'),)) 44 45 self.device_under_test.wait_channel_ready() 46 self.cert_device.wait_channel_ready() 47 48 def teardown_test(self): 49 self.device_under_test.rootservice.StopStack( 50 facade_rootservice.StopStackRequest()) 51 self.cert_device.rootservice.StopStack( 52 facade_rootservice.StopStackRequest()) 53 54 def register_for_event(self, event_code): 55 msg = hci_facade.EventCodeMsg(code=int(event_code)) 56 self.cert_device.hci.RegisterEventHandler(msg) 57 58 def register_for_le_event(self, event_code): 59 msg = hci_facade.LeSubeventCodeMsg(code=int(event_code)) 60 self.cert_device.hci.RegisterLeEventHandler(msg) 61 62 def enqueue_hci_command(self, command, expect_complete): 63 cmd_bytes = bytes(command.Serialize()) 64 cmd = hci_facade.CommandMsg(command=cmd_bytes) 65 if (expect_complete): 66 self.cert_device.hci.EnqueueCommandWithComplete(cmd) 67 else: 68 self.cert_device.hci.EnqueueCommandWithStatus(cmd) 69 70 def test_le_ad_scan_dut_advertises(self): 71 self.register_for_le_event(hci_packets.SubeventCode.ADVERTISING_REPORT) 72 self.register_for_le_event( 73 hci_packets.SubeventCode.EXTENDED_ADVERTISING_REPORT) 74 with EventCallbackStream( 75 self.cert_device.hci.FetchLeSubevents( 76 empty_proto.Empty())) as hci_le_event_stream: 77 78 hci_event_asserts = EventAsserts(hci_le_event_stream) 79 80 # CERT Scans 81 self.enqueue_hci_command( 82 hci_packets.LeSetRandomAddressBuilder('0C:05:04:03:02:01'), 83 True) 84 scan_parameters = hci_packets.PhyScanParameters() 85 scan_parameters.le_scan_type = hci_packets.LeScanType.ACTIVE 86 scan_parameters.le_scan_interval = 40 87 scan_parameters.le_scan_window = 20 88 self.enqueue_hci_command( 89 hci_packets.LeSetExtendedScanParametersBuilder( 90 hci_packets.AddressType.RANDOM_DEVICE_ADDRESS, 91 hci_packets.LeSetScanningFilterPolicy.ACCEPT_ALL, 1, 92 [scan_parameters]), True) 93 self.enqueue_hci_command( 94 hci_packets.LeSetExtendedScanEnableBuilder( 95 hci_packets.Enable.ENABLED, 96 hci_packets.FilterDuplicates.DISABLED, 0, 0), True) 97 98 # DUT Advertises 99 gap_name = hci_packets.GapData() 100 gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME 101 gap_name.data = list(bytes(b'Im_The_DUT!')) 102 gap_data = le_advertising_facade.GapDataMsg( 103 data=bytes(gap_name.Serialize())) 104 config = le_advertising_facade.AdvertisingConfig( 105 advertisement=[gap_data], 106 random_address=common.BluetoothAddress( 107 address=bytes(b'0D:05:04:03:02:01')), 108 interval_min=512, 109 interval_max=768, 110 event_type=le_advertising_facade.AdvertisingEventType.ADV_IND, 111 address_type=common.RANDOM_DEVICE_ADDRESS, 112 peer_address_type=common.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS, 113 peer_address=common.BluetoothAddress( 114 address=bytes(b'A6:A5:A4:A3:A2:A1')), 115 channel_map=7, 116 filter_policy=le_advertising_facade.AdvertisingFilterPolicy. 117 ALL_DEVICES) 118 request = le_advertising_facade.CreateAdvertiserRequest( 119 config=config) 120 121 create_response = self.device_under_test.hci_le_advertising_manager.CreateAdvertiser( 122 request) 123 124 hci_event_asserts.assert_event_occurs( 125 lambda packet: b'Im_The_DUT' in packet.event) 126 127 remove_request = le_advertising_facade.RemoveAdvertiserRequest( 128 advertiser_id=create_response.advertiser_id) 129 self.device_under_test.hci_le_advertising_manager.RemoveAdvertiser( 130 remove_request) 131 self.enqueue_hci_command( 132 hci_packets.LeSetScanEnableBuilder(hci_packets.Enable.DISABLED, 133 hci_packets.Enable.DISABLED), 134 True) 135