import json
import logging
import uuid
import xml.etree.ElementTree as ET

from autotest_lib.client.common_lib import error
from autotest_lib.server.cros.bluetooth import bluetooth_adapter_tests


class bluetooth_SDP_Test(object):
    """Base class with Properties and methods common across SDP tests"""
    version = 1

    MIN_ATTR_BYTE_CNT                = 7
    MAX_ATTR_BYTE_CNT                = 300

    SDP_SERVER_CLASS_ID              = 0x1000
    GAP_CLASS_ID                     = 0x1800
    BROWSE_GROUP_LIST_ATTR_ID        = 0x0005
    PUBLIC_BROWSE_ROOT               = 0x1002

    DOCUMENTATION_URL_ATTR_ID        = 0x000A
    CLIENT_EXECUTABLE_URL_ATTR_ID    = 0x000B
    ICON_URL_ATTR_ID                 = 0x000C
    PROTOCOL_DESCRIPTOR_LIST_ATTR_ID = 0x0004
    L2CAP_UUID                       = 0x0100
    ATT_UUID                         = 0x0007
    ATT_PSM                          = 0x001F
    PNP_INFORMATION_CLASS_ID         = 0x1200
    VERSION_NUMBER_LIST_ATTR_ID      = 0x0200
    SERVICE_DATABASE_STATE_ATTR_ID   = 0x0201
    AVRCP_TG_CLASS_ID                = 0x110C
    PROFILE_DESCRIPTOR_LIST_ATTR_ID  = 0x0009
    ADDITIONAL_PROTOCOLLIST_ATTR_ID  = 0x000D

    FAKE_SERVICE_PATH                = '/autotest/fake_service'
    BLUEZ_URL                        = 'http://www.bluez.org/'

    FAKE_SERVICE_CLASS_ID            = 0xCDEF
    FAKE_ATTRIBUTE_VALUE             = 42
    LANGUAGE_BASE_ATTRIBUTE_ID       = 0x0006

    FAKE_GENERAL_ATTRIBUTE_IDS       = [
                                        0x0003, # TP/SERVER/SA/BV-04-C
                                        0x0002, # TP/SERVER/SA/BV-06-C
                                        0x0007, # TP/SERVER/SA/BV-07-C
                                        0x0008, # TP/SERVER/SA/BV-10-C
                                        # TP/SERVER/SA/BV-09-C:
                                        LANGUAGE_BASE_ATTRIBUTE_ID
                                       ]

    FAKE_LANGUAGE_ATTRIBUTE_OFFSETS  = [
                                        0x0000, # TP/SERVER/SA/BV-12-C
                                        0x0001, # TP/SERVER/SA/BV-13-C
                                        0x0002  # TP/SERVER/SA/BV-14-C
                                       ]

    BLUETOOTH_BASE_UUID              = 0x0000000000001000800000805F9B34FB
    SERVICE_CLASS_ID_ATTR_ID         = 0x0001
    ERROR_CODE_INVALID_RECORD_HANDLE = 0x0002
    ERROR_CODE_INVALID_SYNTAX        = 0x0003
    ERROR_CODE_INVALID_PDU_SIZE      = 0x0004
    INVALID_RECORD_HANDLE            = 0xFEEE
    INVALID_SYNTAX_REQUEST           = '123'
    INVALID_PDU_SIZE                 = 11


    def build_service_record(self):
        """Build SDP record manually for the fake service.

        @return resulting record as string

        """
        value = ET.Element('uint16', {'value': str(self.FAKE_ATTRIBUTE_VALUE)})

        sdp_record = ET.Element('record')

        service_id_attr = ET.Element(
            'attribute', {'id': str(self.SERVICE_CLASS_ID_ATTR_ID)})
        service_id_attr.append(
            ET.Element('uuid', {'value': '0x%X' % self.FAKE_SERVICE_CLASS_ID}))
        sdp_record.append(service_id_attr)

        for attr_id in self.FAKE_GENERAL_ATTRIBUTE_IDS:
            attr = ET.Element('attribute', {'id': str(attr_id)})
            attr.append(value)
            sdp_record.append(attr)

        for offset in self.FAKE_LANGUAGE_ATTRIBUTE_OFFSETS:
            attr_id = self.FAKE_ATTRIBUTE_VALUE + offset
            attr = ET.Element('attribute', {'id': str(attr_id)})
            attr.append(value)
            sdp_record.append(attr)

        sdp_record_str = ('<?xml version="1.0" encoding="UTF-8"?>' +
                          ET.tostring(sdp_record))
        return sdp_record_str


class bluetooth_SDP_ServiceAttributeRequest(bluetooth_SDP_Test,
    bluetooth_adapter_tests.BluetoothAdapterTests):
    """
    Verify the correct behaviour of the device when searching for attributes of
    services.
    """
    version = 1

    MAX_REC_CNT                      = 3

    SERVICE_RECORD_HANDLE_ATTR_ID    = 0x0000

    NON_EXISTING_ATTRIBUTE_ID        = 0xFEDC

    @staticmethod
    def assert_equal(actual, expected):
        """Verify that |actual| is equal to |expected|.

        @param actual: The value we got.
        @param expected: The value we expected.
        @raise error.TestFail: If the values are unequal.
        """
        if actual != expected:
            raise error.TestFail(
                'Expected |%s|, got |%s|' % (expected, actual))


    @staticmethod
    def assert_nonempty_list(value):
        """Verify that |value| is a list, and that the list is non-empty.

        @param value: The value to check.
        @raise error.TestFail: If the value is not a list, or is empty.
        """
        if not isinstance(value, list):
            raise error.TestFail('Value is not a list. Got |%s|.' % value)

        if value == []:
            raise error.TestFail('List is empty')


    def get_single_handle(self, class_id):
        """Send a Service Search Request to get a handle for specific class ID.

        @param class_id: The class that we want a handle for.
        @return The record handle, as an int.
        @raise error.TestFail: If we failed to retrieve a handle.
        """
        res = json.loads(self.tester.service_search_request([class_id],
                         self.MAX_REC_CNT, dict()))
        if not (isinstance(res, list) and len(res) > 0):
            raise error.TestFail(
                    'Failed to retrieve handle for 0x%x' % class_id)
        return res[0]


    # TODO(quiche): Place this after get_attribute(), so all the tests are
    # grouped together.
    def test_record_handle_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-01-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        # Send Service Search Request to find out record handle for
        # SDP Server service.
        record_handle = self.get_single_handle(self.SDP_SERVER_CLASS_ID)

        # Send Service Attribute Request for Service Record Handle Attribute.
        res = json.loads(self.tester.service_attribute_request(
                  record_handle,
                  self.MAX_ATTR_BYTE_CNT,
                  [self.SERVICE_RECORD_HANDLE_ATTR_ID], {}))

        # Ensure that returned attribute is correct.
        self.assert_equal(res,
                          [self.SERVICE_RECORD_HANDLE_ATTR_ID, record_handle])


    def get_attribute(self, class_id, attr_id):
        """Get a single attribute of a single service

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.
        @return attribute value if attribute exists, None otherwise

        """
        record_handle = self.get_single_handle(class_id)
        res = json.loads(self.tester.service_attribute_request(
                  record_handle, self.MAX_ATTR_BYTE_CNT, [attr_id], {}))
        if isinstance(res, list) and len(res) == 2 and res[0] == attr_id:
            return res[1]
        return None


    # TODO(quiche): Move this up, to be grouped with the other |assert|
    # methods.
    def assert_attribute_equals(self, class_id, attr_id, expected_value):
        """Verify that |attr_id| of service with |class_id| has |expected_value|

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.
        @param expected_value: The expected value for the attribute.
        @raise error.TestFail: If the actual value differs from |expected_value|
        """
        self.assert_equal(self.get_attribute(class_id, attr_id),
                          expected_value)


    def test_browse_group_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-08-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        self.assert_attribute_equals(self.GAP_CLASS_ID,
                                     self.BROWSE_GROUP_LIST_ATTR_ID,
                                     [self.PUBLIC_BROWSE_ROOT])


    def test_icon_url_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-11-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        self.assert_attribute_equals(self.GAP_CLASS_ID,
                                     self.ICON_URL_ATTR_ID,
                                     self.BLUEZ_URL)


    def test_documentation_url_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-18-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        self.assert_attribute_equals(self.GAP_CLASS_ID,
                                     self.DOCUMENTATION_URL_ATTR_ID,
                                     self.BLUEZ_URL)


    def test_client_executable_url_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-19-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        self.assert_attribute_equals(self.GAP_CLASS_ID,
                                     self.CLIENT_EXECUTABLE_URL_ATTR_ID,
                                     self.BLUEZ_URL)


    def test_protocol_descriptor_list_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-05-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        value = self.get_attribute(self.GAP_CLASS_ID,
                                   self.PROTOCOL_DESCRIPTOR_LIST_ATTR_ID)

        # The first-layer protocol is L2CAP, using the PSM for ATT protocol.
        self.assert_equal(value[0], [self.L2CAP_UUID, self.ATT_PSM])

        # The second-layer protocol is ATT. The additional parameters are
        # ignored, since they may reasonably vary between implementations.
        self.assert_equal(value[1][0], self.ATT_UUID)


    def test_continuation_state(self):
        """Implementation of test TP/SERVER/SA/BV-03-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        record_handle = self.get_single_handle(self.PNP_INFORMATION_CLASS_ID)
        self.assert_nonempty_list(
            json.loads(self.tester.service_attribute_request(
                record_handle, self.MIN_ATTR_BYTE_CNT, [[0, 0xFFFF]], {})))


    def test_version_list_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-15-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        self.assert_nonempty_list(
            self.get_attribute(self.SDP_SERVER_CLASS_ID,
                self.VERSION_NUMBER_LIST_ATTR_ID))


    def test_service_database_state_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-16-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        state = self.get_attribute(self.SDP_SERVER_CLASS_ID,
                                   self.SERVICE_DATABASE_STATE_ATTR_ID)
        if not isinstance(state, int):
            raise error.TestFail('State is not an int: %s' % state)


    def test_profile_descriptor_list_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-17-C from SDP Specification.

        @raise error.TestFail: If list attribute not correct form.

        """
        profile_list = self.get_attribute(self.PNP_INFORMATION_CLASS_ID,
                                          self.PROFILE_DESCRIPTOR_LIST_ATTR_ID)

        if not isinstance(profile_list, list):
            raise error.TestFail('Value is not a list')
        self.assert_equal(len(profile_list), 1)

        if not isinstance(profile_list[0], list):
            raise error.TestFail('Item is not a list')
        self.assert_equal(len(profile_list[0]), 2)

        self.assert_equal(profile_list[0][0], self.PNP_INFORMATION_CLASS_ID)


    def test_additional_protocol_descriptor_list_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-21-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.

        """

        """AVRCP is not supported by Chromebook and no need to run this test
        self.assert_nonempty_list(
            self.get_attribute(self.AVRCP_TG_CLASS_ID,
                self.ADDITIONAL_PROTOCOLLIST_ATTR_ID))
        """

    def test_non_existing_attribute(self):
        """Implementation of test TP/SERVER/SA/BV-20-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        record_handle = self.get_single_handle(self.FAKE_SERVICE_CLASS_ID)
        res = json.loads(self.tester.service_attribute_request(
                  record_handle, self.MAX_ATTR_BYTE_CNT,
                  [self.NON_EXISTING_ATTRIBUTE_ID], {}))
        self.assert_equal(res, [])


    def test_fake_attributes(self):
        """Test values of attributes of the fake service record.

        @raise error.TestFail: If the DUT failed the test.
        """
        for attr_id in self.FAKE_GENERAL_ATTRIBUTE_IDS:
            self.assert_attribute_equals(self.FAKE_SERVICE_CLASS_ID,
                                         attr_id, self.FAKE_ATTRIBUTE_VALUE)

        for offset in self.FAKE_LANGUAGE_ATTRIBUTE_OFFSETS:
            record_handle = self.get_single_handle(self.FAKE_SERVICE_CLASS_ID)

            lang_base = json.loads(self.tester.service_attribute_request(
                            record_handle, self.MAX_ATTR_BYTE_CNT,
                            [self.LANGUAGE_BASE_ATTRIBUTE_ID], {}))
            attr_id = lang_base[1] + offset

            response = json.loads(self.tester.service_attribute_request(
                record_handle, self.MAX_ATTR_BYTE_CNT, [attr_id], {}))
            self.assert_equal(response, [attr_id, self.FAKE_ATTRIBUTE_VALUE])


    def test_invalid_record_handle(self):
        """Implementation of test TP/SERVER/SA/BI-01-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        res = json.loads(self.tester.service_attribute_request(
                          self.INVALID_RECORD_HANDLE, self.MAX_ATTR_BYTE_CNT,
                          [self.NON_EXISTING_ATTRIBUTE_ID], {}))
        self.assert_equal(res, self.ERROR_CODE_INVALID_RECORD_HANDLE)


    def test_invalid_request_syntax(self):
        """Implementation of test TP/SERVER/SA/BI-02-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        record_handle = self.get_single_handle(self.SDP_SERVER_CLASS_ID)
        res = json.loads(self.tester.service_attribute_request(
                          record_handle,
                          self.MAX_ATTR_BYTE_CNT,
                          [self.SERVICE_RECORD_HANDLE_ATTR_ID],
                          {'invalid_request':self.INVALID_SYNTAX_REQUEST}))
        self.assert_equal(res, self.ERROR_CODE_INVALID_SYNTAX)


    def test_invalid_pdu_size(self):
        """Implementation of test TP/SERVER/SA/BI-03-C from SDP Specification.

        @raise error.TestFail: If the DUT failed the test.
        """
        opts = dict({'forced_pdu_size':self.INVALID_PDU_SIZE})
        record_handle = self.get_single_handle(self.SDP_SERVER_CLASS_ID)
        res = json.loads(self.tester.service_attribute_request(record_handle,
            self.MAX_ATTR_BYTE_CNT, [self.SERVICE_RECORD_HANDLE_ATTR_ID], opts))
        self.assert_equal(res, self.ERROR_CODE_INVALID_PDU_SIZE)


    def correct_request_att_request_test(self):
        """Run basic tests for Service Attribute Request."""
        # Connect to the DUT via L2CAP using SDP socket.
        self.tester.connect(self.adapter['Address'])

        self.test_record_handle_attribute()
        self.test_browse_group_attribute()
        self.test_icon_url_attribute()
        self.test_documentation_url_attribute()
        self.test_client_executable_url_attribute()
        self.test_protocol_descriptor_list_attribute()
        self.test_continuation_state()
        self.test_version_list_attribute()
        self.test_service_database_state_attribute()
        self.test_profile_descriptor_list_attribute()
        self.test_additional_protocol_descriptor_list_attribute()
        self.test_fake_attributes()
        self.test_non_existing_attribute()
        self.test_invalid_record_handle()
        self.test_invalid_pdu_size()
        self.test_invalid_request_syntax()


    def sdp_service_attribute_request_test(self, device):
        """Runs service attribute request test"""

        if self.host.btpeer.get_platform() != 'RASPI':
            raise error.TestNAError('Test only runs on Raspi')

        self.tester = device
        # Reset the adapter to the powered on, discoverable state.
        if not self.bluetooth_facade.reset_on():
            raise error.TestFail('DUT adapter could not be powered on')
        if not self.bluetooth_facade.set_discoverable(True):
            raise error.TestFail('DUT could not be set as discoverable')

        self.adapter = self.bluetooth_facade.get_adapter_properties()

        # Create a fake service record in order to test attributes,
        # that are not present in any of existing services.
        uuid128 = ((self.FAKE_SERVICE_CLASS_ID << 96) +
                   self.BLUETOOTH_BASE_UUID)
        uuid_str = str(uuid.UUID(int=uuid128))
        sdp_record = self.build_service_record()
        self.bluetooth_facade.register_profile(self.FAKE_SERVICE_PATH,
                                     uuid_str,
                                     {"ServiceRecord": sdp_record})

        # Setup the tester as a generic computer.
        if not self.tester.setup('computer'):
            raise error.TestFail('Tester could not be initialized')

        self.correct_request_att_request_test()


class bluetooth_SDP_ServiceBrowse(bluetooth_SDP_Test,
    bluetooth_adapter_tests.BluetoothAdapterTests):
    """
    Verify that the IUT behave correct during Service Browse procedure.
    """
    version = 1

    MAX_BROWSE_REC_CNT             = 100
    MAX_ATTR_BYTE_CNT       = 300
    SERVICE_CLASS_ID_LIST   = 0x0001
    BROWSE_GROUP_DESCRIPTOR = 0x1001
    GROUP_ID                = 0x0200


    def get_attribute_ssr_sar(self, class_id, attr_id, size):
        """Get service attributes using Service Search Request and Service
        Attribute Request.

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.
        @param size: Preferred size of UUID.

        @return attribute value if attribute exists, None otherwise

        """
        handles = json.loads(self.tester.service_search_request(
                      [class_id], self.MAX_BROWSE_REC_CNT,
                      {'preferred_size':size}))

        if not (isinstance(handles, list) and len(handles) > 0):
            return None

        res = []
        for record_handle in handles:
            value = json.loads(self.tester.service_attribute_request(
                        record_handle, self.MAX_ATTR_BYTE_CNT, [attr_id], {}))
            if not (isinstance(value, list) and len(value) == 2 and
                value[0] == attr_id):
                return None
            res.append(value[1])

        return res


    def get_attribute_ssar(self, class_id, attr_id, size):
        """Get service attributes using Service Search Attribute Request.

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.
        @param size: Preferred size of UUID.

        @return attribute value if attribute exists, None otherwise

        """
        response = json.loads(self.tester.service_search_attribute_request(
                       [class_id], self.MAX_ATTR_BYTE_CNT, [attr_id],
                       {'preferred_size':size}))

        if not isinstance(response, list):
            return None

        res = []
        for elem in response:
            if not (isinstance(elem, list) and len(elem) == 2 and
                    elem[0] == attr_id):
                return None
            res.append(elem[1])

        return res


    def test_attribute(self, class_id, attr_id, get_attribute):
        """Test service attributes using 16-bit, 32-bit and 128-bit
        size of UUID.

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.
        @param get_attribute: Method to use to get an attribute value.

        @return attribute value if attribute exists and values from three tests
        are equal, None otherwise

        """
        result_16 = get_attribute(class_id, attr_id, 16)

        for size in 32, 128:
            result_cur = get_attribute(class_id, attr_id, size)
            if result_16 != result_cur:
                return None

        return result_16


    def service_browse(self, get_attribute):
        """Execute a Service Browse procedure.

        @param get_attribute: Method to use to get an attribute value.

        @return sorted list of unique services on the DUT, or False if browse
        did not finish correctly

        """
        # Find services on top of hierarchy.
        root_services = self.test_attribute(self.PUBLIC_BROWSE_ROOT,
                                            self.SERVICE_CLASS_ID_LIST,
                                            get_attribute)

        if not root_services:
            return False

        # Find additional browse groups.
        group_ids = self.test_attribute(self.BROWSE_GROUP_DESCRIPTOR,
                                        self.GROUP_ID,
                                        get_attribute)
        if not group_ids:
            return False

        # Find services from all browse groups.
        all_services = []
        for group_id in group_ids:
            services = self.test_attribute(group_id,
                                           self.SERVICE_CLASS_ID_LIST,
                                           get_attribute)
            if not services:
                return False
            all_services.extend(services)

        # Ensure that root services are among all services.
        for service in root_services:
            if service not in all_services:
                return False

        # Sort all services and remove duplicates.
        all_services.sort()
        last = 0
        for service in all_services[1:]:
            if all_services[last] != service:
                last += 1
                all_services[last] = service

        return all_services[:last + 1]


    def correct_request_browse_test(self):
        """Run basic tests for Service Browse procedure.

        @return True if all tests finishes correctly, False otherwise

        """

        # Connect to the DUT via L2CAP using SDP socket.
        self.tester.connect(self.adapter['Address'])

        browse_ssar = self.service_browse(self.get_attribute_ssar)
        if not browse_ssar:
            return False

        browse_ssr_sar = self.service_browse(self.get_attribute_ssr_sar)


        # Ensure that two different browse methods return the same results.
        return browse_ssar == browse_ssr_sar


    def sdp_service_browse_test(self, device):
        """Runs service browse test"""

        if self.host.btpeer.get_platform() != 'RASPI':
            raise error.TestNAError('Test only runs on Raspi')

        self.tester = device
        # Reset the adapter to the powered on, discoverable state.
        if not (self.bluetooth_facade.reset_on() and
                self.bluetooth_facade.set_discoverable(True)):
            raise error.TestFail('DUT could not be reset to initial state')

        self.adapter = self.bluetooth_facade.get_adapter_properties()

        # Setup the tester as a generic computer.
        if not self.tester.setup('computer'):
            raise error.TestFail('Tester could not be initialized')

        # Since radio is involved, this test is not 100% reliable; instead we
        # repeat a few times until it succeeds.
        for failed_attempts in range(0, 5):
            if self.correct_request_browse_test():
                break
        else:
            raise error.TestFail('Expected device was not found')

        # Record how many attempts this took, hopefully we'll one day figure out
        # a way to reduce this to zero and then the loop above can go away.
        self.write_perf_keyval({'failed_attempts': failed_attempts })


class bluetooth_SDP_ServiceSearchAttributeRequest(bluetooth_SDP_Test,
    bluetooth_adapter_tests.BluetoothAdapterTests):
    """
    Verify the correct behaviour of the device when searching for services and
    attributes.
    """


    NON_EXISTING_SERVICE_CLASS_ID    = 0x9875
    PUBLIC_BROWSE_GROUP_CLASS_ID     = 0x1002

    NON_EXISTING_ATTRIBUTE_ID        = 0xABCD
    SERVICE_CLASS_ID_ATTRIBUTE_ID    = 0x0001
    VERSION_NUMBER_LIST_ATTR_ID      = 0x0200


    def fail_test(self, testname, value):
        """Raise an error for a particular SDP test.

        @param testname: a string representation of the test name.
        @param value: the value that did not pass muster.

        """
        raise error.TestFail('SDP test %s failed: got %s.' % (testname, value))


    def test_non_existing(self, class_id, attr_id):
        """Check that a single attribute of a single service does not exist

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.

        @raises error.TestFail if service or attribute does exists.

        """
        for size in 16, 32, 128:
            result = json.loads(self.tester.service_search_attribute_request(
                         [class_id],
                         self.MAX_ATTR_BYTE_CNT,
                         [attr_id],
                         {'preferred_size':size}))
            if result != []:
                raise error.TestFail('Attribute %s of class %s exists when it '
                                     'should not!' % (class_id, attr_id))


    def get_attribute_sssar(self, class_id, attr_id, size):
        """Get a single attribute of a single service using Service Search
        Attribute Request.

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.
        @param size: Preferred size of UUID.

        @return attribute value if attribute exists

        @raises error.TestFail if attribute does not exist

        """
        res = json.loads(self.tester.service_search_attribute_request(
                  [class_id], self.MAX_ATTR_BYTE_CNT, [attr_id],
                  {'preferred_size':size}))

        if (isinstance(res, list) and len(res) == 1 and
            isinstance(res[0], list) and len(res[0]) == 2 and
            res[0][0] == attr_id):
            return res[0][1]

        raise error.TestFail('Attribute %s of class %s does not exist! (size '
                             '%s)' % (class_id, attr_id, size))


    def test_attribute_sssar(self, class_id, attr_id):
        """Test a single attribute of a single service using 16-bit, 32-bit and
        128-bit size of UUID.

        @param class_id: Class ID of service to check.
        @param attr_id: ID of attribute to check.

        @return attribute value if attribute exists and values from three tests
        are equal

        @raises error.TestFail if attribute doesn't exist or values not equal

        """
        result_16 = self.get_attribute_sssar(class_id, attr_id, 16)
        for size in 32, 128:
            result_cur = self.get_attribute_sssar(class_id, attr_id, size)
            if result_16 != result_cur:
                raise error.TestFail('Attribute test failed %s: expected %s, '
                                     'got %s' % (size, result_16, result_cur))

        return result_16


    def test_non_existing_service(self):
        """Implementation of test TP/SERVER/SSA/BV-01-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        self.test_non_existing(self.NON_EXISTING_SERVICE_CLASS_ID,
                               self.SERVICE_CLASS_ID_ATTRIBUTE_ID)


    def test_non_existing_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-02-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        self.test_non_existing(self.PUBLIC_BROWSE_GROUP_CLASS_ID,
                               self.NON_EXISTING_ATTRIBUTE_ID)


    def test_non_existing_service_attribute(self):
        """Implementation of test TP/SERVER/SSA/BV-03-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        self.test_non_existing(self.NON_EXISTING_SERVICE_CLASS_ID,
                               self.NON_EXISTING_ATTRIBUTE_ID)


    def test_existing_service_attribute(self):
        """Implementation of test TP/SERVER/SSA/BV-04-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.SDP_SERVER_CLASS_ID,
                                    self.SERVICE_CLASS_ID_ATTRIBUTE_ID)
        if not value == [self.SDP_SERVER_CLASS_ID]:
            self.fail_test('TP/SERVER/SSA/BV-04-C', value)


    def test_service_database_state_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-08-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.SDP_SERVER_CLASS_ID,
                                    self.SERVICE_DATABASE_STATE_ATTR_ID)
        if not isinstance(value, int):
            self.fail_test('TP/SERVER/SSA/BV-08-C', value)


    def test_protocol_descriptor_list_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-11-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
                                    self.PROTOCOL_DESCRIPTOR_LIST_ATTR_ID)

        # The first-layer protocol is L2CAP, using the PSM for ATT protocol.
        if value[0] != [self.L2CAP_UUID, self.ATT_PSM]:
            self.fail_test('TP/SERVER/SSA/BV-11-C', value)

        # The second-layer protocol is ATT. The additional parameters are
        # ignored, since they may reasonably vary between implementations.
        if value[1][0] != self.ATT_UUID:
            self.fail_test('TP/SERVER/SSA/BV-11-C', value)


    def test_browse_group_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-12-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
                                    self.BROWSE_GROUP_LIST_ATTR_ID)
        if not value == [self.PUBLIC_BROWSE_ROOT]:
            self.fail_test('TP/SERVER/SSA/BV-12-C', value)


    def test_icon_url_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-15-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
                                    self.ICON_URL_ATTR_ID)
        if not value == self.BLUEZ_URL:
            self.fail_test('TP/SERVER/SSA/BV-15-C', value)


    def test_version_list_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-19-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.SDP_SERVER_CLASS_ID,
                                    self.VERSION_NUMBER_LIST_ATTR_ID)
        if not isinstance(value, list) and value != []:
            self.fail_test('TP/SERVER/SSA/BV-19-C', value)


    def test_profile_descriptor_list_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-20-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.PNP_INFORMATION_CLASS_ID,
                                    self.PROFILE_DESCRIPTOR_LIST_ATTR_ID)
        if not (isinstance(value, list) and len(value) == 1 and
                isinstance(value[0], list) and len(value[0]) == 2 and
                value[0][0] == self.PNP_INFORMATION_CLASS_ID):
            self.fail_test('TP/SERVER/SSA/BV-20-C', value)


    def test_documentation_url_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-21-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
                                    self.DOCUMENTATION_URL_ATTR_ID)
        if not value == self.BLUEZ_URL:
            self.fail_test('TP/SERVER/SSA/BV-21-C', value)


    def test_client_executable_url_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-22-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
                                    self.CLIENT_EXECUTABLE_URL_ATTR_ID)
        if not value == self.BLUEZ_URL:
            self.fail_test('TP/SERVER/SSA/BV-22-C', value)


    def test_additional_protocol_descriptor_list_attribute_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-23-C from SDP Specification.

        @raises error.TestFail if test fails

        """

        """AVRCP is not supported by Chromebook and no need to run this test
        value = self.test_attribute_sssar(self.AVRCP_TG_CLASS_ID,
                                    self.ADDITIONAL_PROTOCOLLIST_ATTR_ID)
        if not isinstance(value, list) and value != []:
            self.fail_test('TP/SERVER/SSA/BV-23-C', value)
        """

    def test_fake_attributes_sssar(self):
        """Test values of attributes of the fake service record.

        @raises error.TestFail if test fails

        """
        for attr_id in self.FAKE_GENERAL_ATTRIBUTE_IDS:
            value = self.test_attribute_sssar(self.FAKE_SERVICE_CLASS_ID,
                                              attr_id)
            if value != self.FAKE_ATTRIBUTE_VALUE:
                self.fail_test('fake service attributes', value)

        for offset in self.FAKE_LANGUAGE_ATTRIBUTE_OFFSETS:
            lang_base = self.test_attribute_sssar(self.FAKE_SERVICE_CLASS_ID,
                                            self.LANGUAGE_BASE_ATTRIBUTE_ID)
            attr_id = lang_base + offset
            value = self.test_attribute_sssar(self.FAKE_SERVICE_CLASS_ID,
                                              attr_id)
            if value != self.FAKE_ATTRIBUTE_VALUE:
                self.fail_test('fake service attributes', value)


    def test_continuation_state_sssar(self):
        """Implementation of test TP/SERVER/SSA/BV-06-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        for size in 16, 32, 128:
            # This request should generate a long response, which will be
            # split into 98 chunks.
            value = json.loads(self.tester.service_search_attribute_request(
                        [self.PUBLIC_BROWSE_GROUP_CLASS_ID],
                        self.MIN_ATTR_BYTE_CNT,
                        [[0, 0xFFFF]], {'preferred_size':size}))
            if not isinstance(value, list) or value == []:
                self.fail_test('TP/SERVER/SSA/BV-06-C', value)


    def test_invalid_request_syntax_sssar(self):
        """Implementation of test TP/SERVER/SSA/BI-01-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        for size in 16, 32, 128:
            value = json.loads(self.tester.service_search_attribute_request(
                        [self.SDP_SERVER_CLASS_ID],
                        self.MAX_ATTR_BYTE_CNT,
                        [self.SERVICE_CLASS_ID_ATTRIBUTE_ID],
                        {'preferred_size':size,
                        'invalid_request':'9875'}))
            if value != self.ERROR_CODE_INVALID_SYNTAX:
                self.fail_test('TP/SERVER/SSA/BI-01-C', value)


    def test_invalid_pdu_size_sssar(self):
        """Implementation of test TP/SERVER/SSA/BI-02-C from SDP Specification.

        @raises error.TestFail if test fails

        """
        for size in 16, 32, 128:
            value = json.loads(self.tester.service_search_attribute_request(
                        [self.SDP_SERVER_CLASS_ID],
                        self.MAX_ATTR_BYTE_CNT,
                        [self.SERVICE_CLASS_ID_ATTRIBUTE_ID],
                        {'preferred_size':size,
                        'forced_pdu_size':100}))
            if value != self.ERROR_CODE_INVALID_PDU_SIZE:
                self.fail_test('TP/SERVER/SSA/BI-02-C', value)


    def correct_request_search_att_test(self):
        """Run tests for Service Search Attribute request.

        @raises error.TestFail if any test fails

        """
        # connect to the DUT via L2CAP using SDP socket
        self.tester.connect(self.adapter['Address'])

        self.test_non_existing_service()
        self.test_non_existing_attribute_sssar()
        self.test_non_existing_service_attribute()
        #self.test_existing_service_attribute()
        self.test_service_database_state_attribute_sssar()
        self.test_protocol_descriptor_list_attribute_sssar()
        self.test_browse_group_attribute_sssar()
        self.test_icon_url_attribute_sssar()
        self.test_version_list_attribute_sssar()
        self.test_profile_descriptor_list_attribute_sssar()
        self.test_documentation_url_attribute_sssar()
        self.test_client_executable_url_attribute_sssar()
        self.test_additional_protocol_descriptor_list_attribute_sssar()
        self.test_fake_attributes_sssar()
        self.test_continuation_state_sssar()
        self.test_invalid_request_syntax_sssar()
        self.test_invalid_pdu_size_sssar()
        logging.info('correct_request finished successfully!')


    def sdp_service_search_attribute_request_test(self, device):
        """Runs service search attribute request test"""

        if self.host.btpeer.get_platform() != 'RASPI':
            raise error.TestNAError('Test only runs on Raspi')

        self.tester = device
        # Reset the adapter to the powered on, discoverable state.
        if not self.bluetooth_facade.reset_on():
            raise error.TestFail('DUT adapter could not be powered on')
        if not self.bluetooth_facade.set_discoverable(True):
            raise error.TestFail('DUT could not be set as discoverable')

        self.adapter = self.bluetooth_facade.get_adapter_properties()

        # Create a fake service record in order to test attributes,
        # that are not present in any of existing services.
        uuid128 = ((self.FAKE_SERVICE_CLASS_ID << 96) +
                   self.BLUETOOTH_BASE_UUID)
        uuid_str = str(uuid.UUID(int=uuid128))
        sdp_record = self.build_service_record()
        self.bluetooth_facade.register_profile(self.FAKE_SERVICE_PATH,
                                     uuid_str,
                                     {"ServiceRecord": sdp_record})

        # Setup the tester as a generic computer.
        if not self.tester.setup('computer'):
            raise error.TestFail('Tester could not be initialized')

        # Since radio is involved, this test is not 100% reliable; instead we
        # repeat a few times until it succeeds.
        passing = False
        for failed_attempts in range(0, 4):
            try:
                self.correct_request_search_att_test()
                passing = True
            except error.TestFail as e:
                logging.warning('Ignoring error: %s', e)
            if passing:
                break
        else:
            self.correct_request_search_att_test()

        # Record how many attempts this took, hopefully we'll one day figure out
        # a way to reduce this to zero and then the loop above can go away.
        self.write_perf_keyval({'failed_attempts': failed_attempts})


class bluetooth_SDP_ServiceSearchRequestBasic(
    bluetooth_adapter_tests.BluetoothAdapterTests):
    """
    Verify the correct behaviour of the device when searching for services.
    """
    version = 1

    SDP_SERVER_CLASS_ID                = 0x1000
    NO_EXISTING_SERVICE_CLASS_ID       = 0x0001
    FAKE_SERVICES_CNT                  = 300
    FAKE_SERVICES_PATH                 = '/autotest/fake_service_'
    FAKE_SERVICES_CLASS_ID             = 0xABCD
    BLUETOOTH_BASE_UUID                = 0x0000000000001000800000805F9B34FB
    SSRB_INVALID_PDU_SIZE              = 9875
    ERROR_CODE_INVALID_REQUEST_SYNTAX  = 0x0003
    ERROR_CODE_INVALID_PDU_SIZE        = 0x0004


    def correct_request_basic_test(self):
        """Search the existing service on the DUT using the Tester.

        @return True if found, False if not found

        """
        # connect to the DUT via L2CAP using SDP socket
        self.tester.connect(self.adapter['Address'])

        for size in 16, 32, 128:
            # test case TP/SERVER/SS/BV-01-C:
            # at least the SDP server service exists
            resp = json.loads(self.tester.service_search_request(
                              [self.SDP_SERVER_CLASS_ID], 3,
                              {'preferred_size':size}))
            if resp != [0]:
                return False
            # test case TP/SERVER/SS/BV-04-C:
            # Service with Class ID = 0x0001 should never exist, as this UUID is
            # reserved as Bluetooth Core Specification UUID
            resp = json.loads(self.tester.service_search_request(
                              [self.NO_EXISTING_SERVICE_CLASS_ID], 3,
                              {'preferred_size':size}))
            if resp != []:
                return False
            # test case TP/SERVER/SS/BV-03-C:
            # request the fake services' Class ID to force SDP to use
            # continuation state
            resp = json.loads(self.tester.service_search_request(
                              [self.FAKE_SERVICES_CLASS_ID],
                              self.FAKE_SERVICES_CNT * 2,
                              {'preferred_size':size}))
            if len(resp) != self.FAKE_SERVICES_CNT:
                return False
            # test case TP/SERVER/SS/BI-01-C:
            # send a Service Search Request with intentionally invalid PDU size
            resp = json.loads(self.tester.service_search_request(
                              [self.SDP_SERVER_CLASS_ID], 3,
                              {'preferred_size':size,
                              'forced_pdu_size':self.SSRB_INVALID_PDU_SIZE}))
            if resp != self.ERROR_CODE_INVALID_PDU_SIZE:
                return False
            # test case TP/SERVER/SS/BI-02-C:
            # send a Service Search Request with invalid syntax
            resp = json.loads(self.tester.service_search_request(
                              [self.SDP_SERVER_CLASS_ID], 3,
                              {'preferred_size':size, 'invalid_request':True}))
            if resp != self.ERROR_CODE_INVALID_REQUEST_SYNTAX:
                return False

        return True


    def sdp_service_search_request_basic_test(self, device):
        """Runs service search request basic test"""

        if self.host.btpeer.get_platform() != 'RASPI':
            raise error.TestNAError('Test only runs on Raspi')

        self.tester = device
        # Reset the adapter to the powered on, discoverable state.
        if not (self.bluetooth_facade.reset_on() and
                self.bluetooth_facade.set_discoverable(True)):
            raise error.TestFail('DUT could not be reset to initial state')

        self.adapter = self.bluetooth_facade.get_adapter_properties()

        # Setup the tester as a generic computer.
        if not self.tester.setup('computer'):
            raise error.TestFail('Tester could not be initialized')

        # Create many fake services with the same Class ID
        for num in range(0, self.FAKE_SERVICES_CNT):
            path_str = self.FAKE_SERVICES_PATH + str(num)
            uuid128 = ((self.FAKE_SERVICES_CLASS_ID << 96) +
                      self.BLUETOOTH_BASE_UUID)
            uuid_str = str(uuid.UUID(int=uuid128))
            self.bluetooth_facade.register_profile(path_str, uuid_str, {})

        # Since radio is involved, this test is not 100% reliable; instead we
        # repeat a few times until it succeeds.
        for failed_attempts in range(0, 5):
            if self.correct_request_basic_test():
                break
        else:
            raise error.TestFail('Expected device was not found')

        # Record how many attempts this took, hopefully we'll one day figure out
        # a way to reduce this to zero and then the loop above can go away.
        self.write_perf_keyval({'failed_attempts': failed_attempts })


class BluetoothSDPTests(bluetooth_SDP_ServiceAttributeRequest,
                        bluetooth_SDP_ServiceBrowse,
                        bluetooth_SDP_ServiceSearchAttributeRequest,
                        bluetooth_SDP_ServiceSearchRequestBasic):
    """Master class that simplifies inheritance of sdp tests"""
    pass