• Home
  • Line#
  • Scopes#
  • Navigate#
  • Raw
  • Download
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
5import base64
6import json
7
8from autotest_lib.client.cros import constants
9from autotest_lib.server import autotest
10from autotest_lib.server import hosts
11from autotest_lib.server.cros import dnsname_mangler
12
13
14class 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
224def 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