server/cros/bluetooth/bluetooth_tester.py

# Copyright (c) 2013 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import base64
import json

from autotest_lib.client.cros import constants
from autotest_lib.server import autotest
from autotest_lib.server import hosts
from autotest_lib.server.cros import dnsname_mangler


class BluetoothTester(object):
    """BluetoothTester is a thin layer of logic over a remote tester.

    The Autotest host object representing the remote tester, passed to this
    class on initialization, can be accessed from its host property.

    """


    XMLRPC_BRINGUP_TIMEOUT_SECONDS = 60

    def __init__(self, tester_host):
        """Construct a BluetoothTester.

        @param tester_host: host object representing a remote host.

        """
        self.host = tester_host
        # Make sure the client library is on the device so that the proxy code
        # is there when we try to call it.
        client_at = autotest.Autotest(self.host)
        client_at.install()
        # Start up the XML-RPC proxy on the tester.
        self._proxy = self.host.xmlrpc_connect(
                constants.BLUETOOTH_TESTER_XMLRPC_SERVER_COMMAND,
                constants.BLUETOOTH_TESTER_XMLRPC_SERVER_PORT,
                command_name=
                  constants.BLUETOOTH_TESTER_XMLRPC_SERVER_CLEANUP_PATTERN,
                ready_test_name=
                  constants.BLUETOOTH_TESTER_XMLRPC_SERVER_READY_METHOD,
                timeout_seconds=self.XMLRPC_BRINGUP_TIMEOUT_SECONDS)


    def setup(self, profile):
        """Set up the tester with the given profile.

        @param profile: Profile to use for this test, valid values are:
                computer - a standard computer profile

        @return True on success, False otherwise.

        """
        return self._proxy.setup(profile)


    def set_discoverable(self, discoverable, timeout=0):
        """Set the discoverable state of the controller.

        @param discoverable: Whether controller should be discoverable.
        @param timeout: Timeout in seconds before disabling discovery again,
                ignored when discoverable is False, must not be zero when
                discoverable is True.

        @return True on success, False otherwise.

        """
        return self._proxy.set_discoverable(discoverable, timeout)


    def read_info(self):
        """Read the adapter information from the Kernel.

        @return the information as a JSON-encoded tuple of:
          ( address, bluetooth_version, manufacturer_id,
            supported_settings, current_settings, class_of_device,
            name, short_name )

        """
        return json.loads(self._proxy.read_info())


    def set_advertising(self, advertising):
        """Set the whether the controller is advertising via LE.

        @param advertising: Whether controller should advertise via LE.

        @return True on success, False otherwise.

        """
        return self._proxy.set_advertising(advertising)


    def discover_devices(self, br_edr=True, le_public=True, le_random=True):
        """Discover remote devices.

        Activates device discovery and collects the set of devices found,
        returning them as a list.

        @param br_edr: Whether to detect BR/EDR devices.
        @param le_public: Whether to detect LE Public Address devices.
        @param le_random: Whether to detect LE Random Address devices.

        @return List of devices found as tuples with the format
                (address, address_type, rssi, flags, base64-encoded eirdata),
                or False if discovery could not be started.

        """
        devices = self._proxy.discover_devices(br_edr, le_public, le_random)
        if devices == False:
            return False

        return (
                (address, address_type, rssi, flags,
                 base64.decodestring(eirdata))
                for address, address_type, rssi, flags, eirdata
                in json.loads(devices)
        )


    def close(self):
        """Tear down state associated with the client."""
        # This kills the RPC server.
        self.host.close()


    def connect(self, address):
        """Connect to device with the given address

        @param address: Bluetooth address.

        """
        self._proxy.connect(address)


    def service_search_request(self, uuids, max_rec_cnt, preferred_size=32,
                               forced_pdu_size=None, invalid_request=False):
        """Send a Service Search Request

        @param uuids: List of UUIDs (as integers) to look for.
        @param max_rec_cnt: Maximum count of returned service records.
        @param preferred_size: Preffered size of UUIDs in bits (16, 32, or 128).
        @param forced_pdu_size: Use certain PDU size parameter instead of
               calculating actual length of sequence.
        @param invalid_request: Whether to send request with intentionally
               invalid syntax for testing purposes (bool flag).

        @return list of found services' service record handles or Error Code

        """
        return json.loads(
                self._proxy.service_search_request(
                 uuids, max_rec_cnt, preferred_size, forced_pdu_size,
                 invalid_request)
        )


    def service_attribute_request(self, handle, max_attr_byte_count, attr_ids,
                                  forced_pdu_size=None, invalid_request=None):
        """Send a Service Attribute Request

        @param handle: service record from which attribute values are to be
               retrieved.
        @param max_attr_byte_count: maximum number of bytes of attribute data to
               be returned in the response to this request.
        @param attr_ids: a list, where each element is either an attribute ID
               or a range of attribute IDs.
        @param forced_pdu_size: Use certain PDU size parameter instead of
               calculating actual length of sequence.
        @param invalid_request: Whether to send request with intentionally
               invalid syntax for testing purposes (string with raw request).

        @return list of found attributes IDs and their values or Error Code

        """
        return json.loads(
                self._proxy.service_attribute_request(
                 handle, max_attr_byte_count, attr_ids, forced_pdu_size,
                 invalid_request)
        )


    def service_search_attribute_request(self, uuids, max_attr_byte_count,
                                         attr_ids, preferred_size=32,
                                         forced_pdu_size=None,
                                         invalid_request=None):
        """Send a Service Search Attribute Request

        @param uuids: list of UUIDs (as integers) to look for.
        @param max_attr_byte_count: maximum number of bytes of attribute data to
               be returned in the response to this request.
        @param attr_ids: a list, where each element is either an attribute ID
               or a range of attribute IDs.
        @param preferred_size: Preffered size of UUIDs in bits (16, 32, or 128).
        @param forced_pdu_size: Use certain PDU size parameter instead of
               calculating actual length of sequence.
        @param invalid_request: Whether to send request with intentionally
               invalid syntax for testing purposes (string to be prepended
               to correct request).

        @return list of found attributes IDs and their values or Error Code

        """
        return json.loads(
                self._proxy.service_search_attribute_request(
                 uuids, max_attr_byte_count, attr_ids, preferred_size,
                 forced_pdu_size, invalid_request)
        )


def create_host_from(device_host, args=None):
    """Creates a host object for the Tester associated with a DUT.

    The IP address or the hostname can be specified in the 'tester' member of
    the argument dictionary. When not present it is derived from the hostname
    of the DUT by appending '-router' to the first part.

    Will raise an exception if there isn't a tester for the DUT, or if the DUT
    is specified as an IP address and thus the hostname cannot be derived.

    @param device_host: Autotest host object for the DUT.
    @param args: Dictionary of arguments passed to the test.

    @return Autotest host object for the Tester.

    """
    cmdline_override = args.get('tester', None)
    hostname = dnsname_mangler.get_tester_addr(
            device_host.hostname,
            cmdline_override=cmdline_override)
    return hosts.create_host(hostname)