• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
1#!/usr/bin/env python3
2#
3#   Copyright 2020 - 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
17from bluetooth_packets_python3 import hci_packets
18from blueberry.tests.gd.cert.event_stream import EventStream
19from blueberry.tests.gd.cert.event_stream import IEventStream
20from blueberry.tests.gd.cert.closable import Closable
21from blueberry.tests.gd.cert.closable import safeClose
22from blueberry.tests.gd.cert.truth import assertThat
23from google.protobuf import empty_pb2 as empty_proto
24from blueberry.facade.hci import hci_facade_pb2 as hci_facade
25from blueberry.facade.neighbor import facade_pb2 as neighbor_facade
26
27
28class InquirySession(Closable, IEventStream):
29
30    def __init__(self, device, inquiry_msg):
31        self.inquiry_event_stream = EventStream(device.neighbor.SetInquiryMode(inquiry_msg))
32
33    def get_event_queue(self):
34        return self.inquiry_event_stream.get_event_queue()
35
36    def close(self):
37        safeClose(self.inquiry_event_stream)
38
39
40class GetRemoteNameSession(Closable):
41
42    def __init__(self, device):
43        self.remote_name_stream = EventStream(device.neighbor.GetRemoteNameEvents(empty_proto.Empty()))
44
45    def verify_name(self, name):
46        assertThat(self.remote_name_stream).emits(lambda msg: bytes(name) in msg.name)
47
48    def close(self):
49        safeClose(self.remote_name_stream)
50
51
52class PyNeighbor(object):
53
54    def __init__(self, device):
55        self.device = device
56        self.remote_host_supported_features_notification_registered = False
57
58    def set_inquiry_mode(self, inquiry_msg):
59        """
60        Set the inquiry mode and return a session which can be used for event queue assertion
61        """
62        return InquirySession(self.device, inquiry_msg)
63
64    def _register_remote_host_supported_features_notification(self):
65        """
66        REMOTE_HOST_SUPPORTED_FEATURES_NOTIFICATION event will be sent when a device sends remote name request
67        """
68        if self.remote_host_supported_features_notification_registered:
69            return
70        msg = hci_facade.EventRequest(code=int(hci_packets.EventCode.REMOTE_HOST_SUPPORTED_FEATURES_NOTIFICATION))
71        self.device.hci.RequestEvent(msg)
72        self.remote_host_supported_features_notification_registered = True
73
74    def get_remote_name(self, remote_address):
75        """
76        Get the remote name and return a session which can be used for event queue assertion
77        """
78        self._register_remote_host_supported_features_notification()
79        self.device.neighbor.ReadRemoteName(
80            neighbor_facade.RemoteNameRequestMsg(
81                address=remote_address.encode('utf8'), page_scan_repetition_mode=1, clock_offset=0x6855))
82        return GetRemoteNameSession(self.device)
83