1 # Copyright (c) 2013 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import base64 6 import json 7 8 from autotest_lib.client.cros import constants 9 from autotest_lib.server import autotest 10 from autotest_lib.server import hosts 11 from autotest_lib.server.cros import dnsname_mangler 12 13 14 class BluetoothTester(object): 15 """BluetoothTester is a thin layer of logic over a remote tester. 16 17 The Autotest host object representing the remote tester, passed to this 18 class on initialization, can be accessed from its host property. 19 20 """ 21 22 23 XMLRPC_BRINGUP_TIMEOUT_SECONDS = 60 24 XMLRPC_LOG_PATH = '/var/log/bluetooth_xmlrpc_tester.log' 25 26 def __init__(self, tester_host): 27 """Construct a BluetoothTester. 28 29 @param tester_host: host object representing a remote host. 30 31 """ 32 self.host = tester_host 33 # Make sure the client library is on the device so that the proxy code 34 # is there when we try to call it. 35 client_at = autotest.Autotest(self.host) 36 client_at.install() 37 # Start up the XML-RPC proxy on the tester. 38 self._proxy = self.host.rpc_server_tracker.xmlrpc_connect( 39 constants.BLUETOOTH_TESTER_XMLRPC_SERVER_COMMAND, 40 constants.BLUETOOTH_TESTER_XMLRPC_SERVER_PORT, 41 command_name= 42 constants.BLUETOOTH_TESTER_XMLRPC_SERVER_CLEANUP_PATTERN, 43 ready_test_name= 44 constants.BLUETOOTH_TESTER_XMLRPC_SERVER_READY_METHOD, 45 timeout_seconds=self.XMLRPC_BRINGUP_TIMEOUT_SECONDS, 46 logfile=self.XMLRPC_LOG_PATH) 47 48 49 def setup(self, profile): 50 """Set up the tester with the given profile. 51 52 @param profile: Profile to use for this test, valid values are: 53 computer - a standard computer profile 54 55 @return True on success, False otherwise. 56 57 """ 58 return self._proxy.setup(profile) 59 60 61 def set_discoverable(self, discoverable, timeout=0): 62 """Set the discoverable state of the controller. 63 64 @param discoverable: Whether controller should be discoverable. 65 @param timeout: Timeout in seconds before disabling discovery again, 66 ignored when discoverable is False, must not be zero when 67 discoverable is True. 68 69 @return True on success, False otherwise. 70 71 """ 72 return self._proxy.set_discoverable(discoverable, timeout) 73 74 75 def read_info(self): 76 """Read the adapter information from the Kernel. 77 78 @return the information as a JSON-encoded tuple of: 79 ( address, bluetooth_version, manufacturer_id, 80 supported_settings, current_settings, class_of_device, 81 name, short_name ) 82 83 """ 84 return json.loads(self._proxy.read_info()) 85 86 87 def set_advertising(self, advertising): 88 """Set the whether the controller is advertising via LE. 89 90 @param advertising: Whether controller should advertise via LE. 91 92 @return True on success, False otherwise. 93 94 """ 95 return self._proxy.set_advertising(advertising) 96 97 98 def discover_devices(self, br_edr=True, le_public=True, le_random=True): 99 """Discover remote devices. 100 101 Activates device discovery and collects the set of devices found, 102 returning them as a list. 103 104 @param br_edr: Whether to detect BR/EDR devices. 105 @param le_public: Whether to detect LE Public Address devices. 106 @param le_random: Whether to detect LE Random Address devices. 107 108 @return List of devices found as tuples with the format 109 (address, address_type, rssi, flags, base64-encoded eirdata), 110 or False if discovery could not be started. 111 112 """ 113 devices = self._proxy.discover_devices(br_edr, le_public, le_random) 114 if devices == False: 115 return False 116 117 return ( 118 (address, address_type, rssi, flags, 119 base64.decodestring(eirdata)) 120 for address, address_type, rssi, flags, eirdata 121 in json.loads(devices) 122 ) 123 124 125 def copy_logs(self, destination): 126 """Copy the logs generated by this tester to a given location. 127 128 @param destination: destination directory for the logs. 129 130 """ 131 self.host.collect_logs(self.XMLRPC_LOG_PATH, destination) 132 133 134 def close(self): 135 """Tear down state associated with the client.""" 136 # This kills the RPC server. 137 self.host.close() 138 139 140 def connect(self, address): 141 """Connect to device with the given address 142 143 @param address: Bluetooth address. 144 145 """ 146 self._proxy.connect(address) 147 148 149 def service_search_request(self, uuids, max_rec_cnt, preferred_size=32, 150 forced_pdu_size=None, invalid_request=False): 151 """Send a Service Search Request 152 153 @param uuids: List of UUIDs (as integers) to look for. 154 @param max_rec_cnt: Maximum count of returned service records. 155 @param preferred_size: Preffered size of UUIDs in bits (16, 32, or 128). 156 @param forced_pdu_size: Use certain PDU size parameter instead of 157 calculating actual length of sequence. 158 @param invalid_request: Whether to send request with intentionally 159 invalid syntax for testing purposes (bool flag). 160 161 @return list of found services' service record handles or Error Code 162 163 """ 164 return json.loads( 165 self._proxy.service_search_request( 166 uuids, max_rec_cnt, preferred_size, forced_pdu_size, 167 invalid_request) 168 ) 169 170 171 def service_attribute_request(self, handle, max_attr_byte_count, attr_ids, 172 forced_pdu_size=None, invalid_request=None): 173 """Send a Service Attribute Request 174 175 @param handle: service record from which attribute values are to be 176 retrieved. 177 @param max_attr_byte_count: maximum number of bytes of attribute data to 178 be returned in the response to this request. 179 @param attr_ids: a list, where each element is either an attribute ID 180 or a range of attribute IDs. 181 @param forced_pdu_size: Use certain PDU size parameter instead of 182 calculating actual length of sequence. 183 @param invalid_request: Whether to send request with intentionally 184 invalid syntax for testing purposes (string with raw request). 185 186 @return list of found attributes IDs and their values or Error Code 187 188 """ 189 return json.loads( 190 self._proxy.service_attribute_request( 191 handle, max_attr_byte_count, attr_ids, forced_pdu_size, 192 invalid_request) 193 ) 194 195 196 def service_search_attribute_request(self, uuids, max_attr_byte_count, 197 attr_ids, preferred_size=32, 198 forced_pdu_size=None, 199 invalid_request=None): 200 """Send a Service Search Attribute Request 201 202 @param uuids: list of UUIDs (as integers) to look for. 203 @param max_attr_byte_count: maximum number of bytes of attribute data to 204 be returned in the response to this request. 205 @param attr_ids: a list, where each element is either an attribute ID 206 or a range of attribute IDs. 207 @param preferred_size: Preffered size of UUIDs in bits (16, 32, or 128). 208 @param forced_pdu_size: Use certain PDU size parameter instead of 209 calculating actual length of sequence. 210 @param invalid_request: Whether to send request with intentionally 211 invalid syntax for testing purposes (string to be prepended 212 to correct request). 213 214 @return list of found attributes IDs and their values or Error Code 215 216 """ 217 return json.loads( 218 self._proxy.service_search_attribute_request( 219 uuids, max_attr_byte_count, attr_ids, preferred_size, 220 forced_pdu_size, invalid_request) 221 ) 222 223 224 def create_host_from(device_host, args=None): 225 """Creates a host object for the Tester associated with a DUT. 226 227 The IP address or the hostname can be specified in the 'tester' member of 228 the argument dictionary. When not present it is derived from the hostname 229 of the DUT by appending '-router' to the first part. 230 231 Will raise an exception if there isn't a tester for the DUT, or if the DUT 232 is specified as an IP address and thus the hostname cannot be derived. 233 234 @param device_host: Autotest host object for the DUT. 235 @param args: Dictionary of arguments passed to the test. 236 237 @return Autotest host object for the Tester. 238 239 """ 240 cmdline_override = args.get('tester', None) 241 hostname = dnsname_mangler.get_tester_addr( 242 device_host.hostname, 243 cmdline_override=cmdline_override) 244 return hosts.create_host(hostname) 245
