• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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 le_scanning_manager_facade_pb2 as le_scanning_facade
28from hci.facade import 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 LeScanningManagerTest(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_INTERFACES'),))
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_scans(self):
71        with EventCallbackStream(
72                # DUT Scans
73                self.device_under_test.hci_le_scanning_manager.StartScan(
74                    empty_proto.Empty())) as advertising_event_stream:
75
76            hci_event_asserts = EventAsserts(advertising_event_stream)
77
78            # CERT Advertises
79            gap_name = hci_packets.GapData()
80            gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
81            gap_name.data = list(bytes(b'Im_The_CERT!'))
82            gap_data = le_advertising_facade.GapDataMsg(
83                data=bytes(gap_name.Serialize()))
84            config = le_advertising_facade.AdvertisingConfig(
85                advertisement=[gap_data],
86                random_address=common.BluetoothAddress(
87                    address=bytes(b'A6:A5:A4:A3:A2:A1')),
88                interval_min=512,
89                interval_max=768,
90                event_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
91                address_type=common.RANDOM_DEVICE_ADDRESS,
92                peer_address_type=common.PUBLIC_DEVICE_OR_IDENTITY_ADDRESS,
93                peer_address=common.BluetoothAddress(
94                    address=bytes(b'0C:05:04:03:02:01')),
95                channel_map=7,
96                filter_policy=le_advertising_facade.AdvertisingFilterPolicy.
97                ALL_DEVICES)
98            request = le_advertising_facade.CreateAdvertiserRequest(
99                config=config)
100
101            create_response = self.cert_device.hci_le_advertising_manager.CreateAdvertiser(
102                request)
103
104            hci_event_asserts.assert_event_occurs(
105                lambda packet: b'Im_The_CERT' in packet.event)
106
107            remove_request = le_advertising_facade.RemoveAdvertiserRequest(
108                advertiser_id=create_response.advertiser_id)
109            self.cert_device.hci_le_advertising_manager.RemoveAdvertiser(
110                remove_request)
111