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