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 import GdBaseTestClass
22from cert.event_stream import EventStream
23from google.protobuf import empty_pb2 as empty_proto
24from facade import rootservice_pb2 as facade_rootservice
25from hci.facade import hci_facade_pb2 as hci_facade
26from hci.facade import le_scanning_manager_facade_pb2 as le_scanning_facade
27from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade
28from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade
29from bluetooth_packets_python3 import hci_packets
30from facade import common_pb2 as common
31
32
33class LeScanningManagerTest(GdBaseTestClass):
34
35    def setup_class(self):
36        super().setup_class(dut_module='HCI_INTERFACES', cert_module='HCI_INTERFACES')
37
38    def register_for_event(self, event_code):
39        msg = hci_facade.EventCodeMsg(code=int(event_code))
40        self.cert.hci.RegisterEventHandler(msg)
41
42    def register_for_le_event(self, event_code):
43        msg = hci_facade.LeSubeventCodeMsg(code=int(event_code))
44        self.cert.hci.RegisterLeEventHandler(msg)
45
46    def enqueue_hci_command(self, command, expect_complete):
47        cmd_bytes = bytes(command.Serialize())
48        cmd = common.Data(payload=cmd_bytes)
49        if (expect_complete):
50            self.cert.hci.EnqueueCommandWithComplete(cmd)
51        else:
52            self.cert.hci.EnqueueCommandWithStatus(cmd)
53
54    def test_le_ad_scan_dut_scans(self):
55        privacy_policy = le_initiator_address_facade.PrivacyPolicy(
56            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
57            address_with_type=common.BluetoothAddressWithType(
58                address=common.BluetoothAddress(address=bytes(b'D0:05:04:03:02:01')),
59                type=common.RANDOM_DEVICE_ADDRESS),
60            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
61            minimum_rotation_time=0,
62            maximum_rotation_time=0)
63        self.dut.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(privacy_policy)
64        cert_privacy_policy = le_initiator_address_facade.PrivacyPolicy(
65            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
66            address_with_type=common.BluetoothAddressWithType(
67                address=common.BluetoothAddress(address=bytes(b'C0:05:04:03:02:01')),
68                type=common.RANDOM_DEVICE_ADDRESS),
69            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
70            minimum_rotation_time=0,
71            maximum_rotation_time=0)
72        self.cert.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(cert_privacy_policy)
73        with EventStream(
74                # DUT Scans
75                self.dut.hci_le_scanning_manager.StartScan(empty_proto.Empty())) as advertising_event_stream:
76
77            # CERT Advertises
78            gap_name = hci_packets.GapData()
79            gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
80            gap_name.data = list(bytes(b'Im_The_CERT!'))
81            gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
82            config = le_advertising_facade.AdvertisingConfig(
83                advertisement=[gap_data],
84                interval_min=512,
85                interval_max=768,
86                advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
87                own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
88                channel_map=7,
89                filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
90            request = le_advertising_facade.CreateAdvertiserRequest(config=config)
91
92            create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)
93
94            advertising_event_stream.assert_event_occurs(lambda packet: b'Im_The_CERT' in packet.event)
95
96            remove_request = le_advertising_facade.RemoveAdvertiserRequest(advertiser_id=create_response.advertiser_id)
97            self.cert.hci_le_advertising_manager.RemoveAdvertiser(remove_request)
98