xref: /aosp_15_r20/external/autotest/server/cros/bluetooth/bluetooth_sdp_tests.py (revision 9c5db1993ded3edbeafc8092d69fe5de2ee02df7)
1# Lint as: python2, python3
2from __future__ import absolute_import
3from __future__ import division
4from __future__ import print_function
5
6import json
7import logging
8import uuid
9import xml.etree.ElementTree as ET
10
11import common
12from autotest_lib.client.common_lib import error
13from autotest_lib.server.cros.bluetooth import bluetooth_adapter_tests
14from six.moves import range
15
16
17class bluetooth_SDP_Test(object):
18    """Base class with Properties and methods common across SDP tests"""
19    version = 1
20
21    MIN_ATTR_BYTE_CNT                = 7
22    MAX_ATTR_BYTE_CNT                = 300
23
24    SDP_SERVER_CLASS_ID              = 0x1000
25    GAP_CLASS_ID                     = 0x1800
26    BROWSE_GROUP_LIST_ATTR_ID        = 0x0005
27    PUBLIC_BROWSE_ROOT               = 0x1002
28
29    DOCUMENTATION_URL_ATTR_ID        = 0x000A
30    CLIENT_EXECUTABLE_URL_ATTR_ID    = 0x000B
31    ICON_URL_ATTR_ID                 = 0x000C
32    PROTOCOL_DESCRIPTOR_LIST_ATTR_ID = 0x0004
33    L2CAP_UUID                       = 0x0100
34    ATT_UUID                         = 0x0007
35    ATT_PSM                          = 0x001F
36    PNP_INFORMATION_CLASS_ID         = 0x1200
37    VERSION_NUMBER_LIST_ATTR_ID      = 0x0200
38    SERVICE_DATABASE_STATE_ATTR_ID   = 0x0201
39    AVRCP_TG_CLASS_ID                = 0x110C
40    PROFILE_DESCRIPTOR_LIST_ATTR_ID  = 0x0009
41    ADDITIONAL_PROTOCOLLIST_ATTR_ID  = 0x000D
42
43    FAKE_SERVICE_PATH                = '/autotest/fake_service'
44    BLUEZ_URL                        = 'http://www.bluez.org/'
45
46    FAKE_SERVICE_CLASS_ID            = 0xCDEF
47    FAKE_ATTRIBUTE_VALUE             = 42
48    LANGUAGE_BASE_ATTRIBUTE_ID       = 0x0006
49
50    FAKE_GENERAL_ATTRIBUTE_IDS       = [
51                                        0x0003, # TP/SERVER/SA/BV-04-C
52                                        0x0002, # TP/SERVER/SA/BV-06-C
53                                        0x0007, # TP/SERVER/SA/BV-07-C
54                                        0x0008, # TP/SERVER/SA/BV-10-C
55                                        # TP/SERVER/SA/BV-09-C:
56                                        LANGUAGE_BASE_ATTRIBUTE_ID
57                                       ]
58
59    FAKE_LANGUAGE_ATTRIBUTE_OFFSETS  = [
60                                        0x0000, # TP/SERVER/SA/BV-12-C
61                                        0x0001, # TP/SERVER/SA/BV-13-C
62                                        0x0002  # TP/SERVER/SA/BV-14-C
63                                       ]
64
65    BLUETOOTH_BASE_UUID              = 0x0000000000001000800000805F9B34FB
66    SERVICE_CLASS_ID_ATTR_ID         = 0x0001
67    ERROR_CODE_INVALID_RECORD_HANDLE = 0x0002
68    ERROR_CODE_INVALID_SYNTAX        = 0x0003
69    ERROR_CODE_INVALID_PDU_SIZE      = 0x0004
70    INVALID_RECORD_HANDLE            = 0xFEEE
71    INVALID_SYNTAX_REQUEST           = '123'
72    INVALID_PDU_SIZE                 = 11
73
74
75    def build_service_record(self):
76        """Build SDP record manually for the fake service.
77
78        @return resulting record as string
79
80        """
81        value = ET.Element('uint16', {'value': str(self.FAKE_ATTRIBUTE_VALUE)})
82
83        sdp_record = ET.Element('record')
84
85        service_id_attr = ET.Element(
86            'attribute', {'id': str(self.SERVICE_CLASS_ID_ATTR_ID)})
87        sequence = ET.Element('sequence')
88        sequence.append(
89                ET.Element('uuid',
90                           {'value': '0x%X' % self.FAKE_SERVICE_CLASS_ID}))
91        service_id_attr.append(sequence)
92        sdp_record.append(service_id_attr)
93
94        for attr_id in self.FAKE_GENERAL_ATTRIBUTE_IDS:
95            attr = ET.Element('attribute', {'id': str(attr_id)})
96            attr.append(value)
97            sdp_record.append(attr)
98
99        for offset in self.FAKE_LANGUAGE_ATTRIBUTE_OFFSETS:
100            attr_id = self.FAKE_ATTRIBUTE_VALUE + offset
101            attr = ET.Element('attribute', {'id': str(attr_id)})
102            attr.append(value)
103            sdp_record.append(attr)
104
105        sdp_record_str = ('<?xml version="1.0" encoding="UTF-8"?>' +
106                          ET.tostring(sdp_record).decode('utf-8'))
107        return sdp_record_str
108
109
110class bluetooth_SDP_ServiceAttributeRequest(bluetooth_SDP_Test,
111    bluetooth_adapter_tests.BluetoothAdapterTests):
112    """
113    Verify the correct behaviour of the device when searching for attributes of
114    services.
115    """
116    version = 1
117
118    MAX_REC_CNT                      = 3
119
120    SERVICE_RECORD_HANDLE_ATTR_ID    = 0x0000
121
122    NON_EXISTING_ATTRIBUTE_ID        = 0xFEDC
123
124    @staticmethod
125    def assert_equal(actual, expected):
126        """Verify that |actual| is equal to |expected|.
127
128        @param actual: The value we got.
129        @param expected: The value we expected.
130        @raise error.TestFail: If the values are unequal.
131        """
132        if actual != expected:
133            raise error.TestFail(
134                'Expected |%s|, got |%s|' % (expected, actual))
135
136
137    @staticmethod
138    def assert_nonempty_list(value):
139        """Verify that |value| is a list, and that the list is non-empty.
140
141        @param value: The value to check.
142        @raise error.TestFail: If the value is not a list, or is empty.
143        """
144        if not isinstance(value, list):
145            raise error.TestFail('Value is not a list. Got |%s|.' % value)
146
147        if value == []:
148            raise error.TestFail('List is empty')
149
150
151    def get_single_handle(self, class_id):
152        """Send a Service Search Request to get a handle for specific class ID.
153
154        @param class_id: The class that we want a handle for.
155        @return The record handle, as an int.
156        @raise error.TestFail: If we failed to retrieve a handle.
157        """
158        res = json.loads(self.tester.service_search_request([class_id],
159                         self.MAX_REC_CNT, dict()))
160        if not (isinstance(res, list) and len(res) > 0):
161            raise error.TestFail(
162                    'Failed to retrieve handle for 0x%x' % class_id)
163        return res[0]
164
165
166    # TODO(quiche): Place this after get_attribute(), so all the tests are
167    # grouped together.
168    def test_record_handle_attribute(self):
169        """Implementation of test TP/SERVER/SA/BV-01-C from SDP Specification.
170
171        @raise error.TestFail: If the DUT failed the test.
172        """
173        # Send Service Search Request to find out record handle for
174        # SDP Server service.
175        record_handle = self.get_single_handle(self.SDP_SERVER_CLASS_ID)
176
177        # Send Service Attribute Request for Service Record Handle Attribute.
178        res = json.loads(self.tester.service_attribute_request(
179                  record_handle,
180                  self.MAX_ATTR_BYTE_CNT,
181                  [self.SERVICE_RECORD_HANDLE_ATTR_ID], {}))
182
183        # Ensure that returned attribute is correct.
184        self.assert_equal(res,
185                          [self.SERVICE_RECORD_HANDLE_ATTR_ID, record_handle])
186
187
188    def get_attribute(self, class_id, attr_id):
189        """Get a single attribute of a single service
190
191        @param class_id: Class ID of service to check.
192        @param attr_id: ID of attribute to check.
193        @return attribute value if attribute exists, None otherwise
194
195        """
196        record_handle = self.get_single_handle(class_id)
197        res = json.loads(self.tester.service_attribute_request(
198                  record_handle, self.MAX_ATTR_BYTE_CNT, [attr_id], {}))
199        if isinstance(res, list) and len(res) == 2 and res[0] == attr_id:
200            return res[1]
201        return None
202
203
204    # TODO(quiche): Move this up, to be grouped with the other |assert|
205    # methods.
206    def assert_attribute_equals(self, class_id, attr_id, expected_value):
207        """Verify that |attr_id| of service with |class_id| has |expected_value|
208
209        @param class_id: Class ID of service to check.
210        @param attr_id: ID of attribute to check.
211        @param expected_value: The expected value for the attribute.
212        @raise error.TestFail: If the actual value differs from |expected_value|
213        """
214        self.assert_equal(self.get_attribute(class_id, attr_id),
215                          expected_value)
216
217
218    def test_browse_group_attribute(self):
219        """Implementation of test TP/SERVER/SA/BV-08-C from SDP Specification.
220
221        @raise error.TestFail: If the DUT failed the test.
222        """
223        self.assert_attribute_equals(self.GAP_CLASS_ID,
224                                     self.BROWSE_GROUP_LIST_ATTR_ID,
225                                     [self.PUBLIC_BROWSE_ROOT])
226
227
228    def test_icon_url_attribute(self):
229        """Implementation of test TP/SERVER/SA/BV-11-C from SDP Specification.
230
231        @raise error.TestFail: If the DUT failed the test.
232        """
233        self.assert_attribute_equals(self.GAP_CLASS_ID,
234                                     self.ICON_URL_ATTR_ID,
235                                     self.BLUEZ_URL)
236
237
238    def test_documentation_url_attribute(self):
239        """Implementation of test TP/SERVER/SA/BV-18-C from SDP Specification.
240
241        @raise error.TestFail: If the DUT failed the test.
242        """
243        self.assert_attribute_equals(self.GAP_CLASS_ID,
244                                     self.DOCUMENTATION_URL_ATTR_ID,
245                                     self.BLUEZ_URL)
246
247
248    def test_client_executable_url_attribute(self):
249        """Implementation of test TP/SERVER/SA/BV-19-C from SDP Specification.
250
251        @raise error.TestFail: If the DUT failed the test.
252        """
253        self.assert_attribute_equals(self.GAP_CLASS_ID,
254                                     self.CLIENT_EXECUTABLE_URL_ATTR_ID,
255                                     self.BLUEZ_URL)
256
257
258    def test_protocol_descriptor_list_attribute(self):
259        """Implementation of test TP/SERVER/SA/BV-05-C from SDP Specification.
260
261        @raise error.TestFail: If the DUT failed the test.
262        """
263        value = self.get_attribute(self.GAP_CLASS_ID,
264                                   self.PROTOCOL_DESCRIPTOR_LIST_ATTR_ID)
265
266        # The first-layer protocol is L2CAP, using the PSM for ATT protocol.
267        self.assert_equal(value[0], [self.L2CAP_UUID, self.ATT_PSM])
268
269        # The second-layer protocol is ATT. The additional parameters are
270        # ignored, since they may reasonably vary between implementations.
271        self.assert_equal(value[1][0], self.ATT_UUID)
272
273
274    def test_continuation_state(self):
275        """Implementation of test TP/SERVER/SA/BV-03-C from SDP Specification.
276
277        @raise error.TestFail: If the DUT failed the test.
278        """
279        record_handle = self.get_single_handle(self.PNP_INFORMATION_CLASS_ID)
280        self.assert_nonempty_list(
281            json.loads(self.tester.service_attribute_request(
282                record_handle, self.MIN_ATTR_BYTE_CNT, [[0, 0xFFFF]], {})))
283
284
285    def test_version_list_attribute(self):
286        """Implementation of test TP/SERVER/SA/BV-15-C from SDP Specification.
287
288        @raise error.TestFail: If the DUT failed the test.
289        """
290        self.assert_nonempty_list(
291            self.get_attribute(self.SDP_SERVER_CLASS_ID,
292                self.VERSION_NUMBER_LIST_ATTR_ID))
293
294
295    def test_service_database_state_attribute(self):
296        """Implementation of test TP/SERVER/SA/BV-16-C from SDP Specification.
297
298        @raise error.TestFail: If the DUT failed the test.
299        """
300        state = self.get_attribute(self.SDP_SERVER_CLASS_ID,
301                                   self.SERVICE_DATABASE_STATE_ATTR_ID)
302        if not isinstance(state, int):
303            raise error.TestFail('State is not an int: %s' % state)
304
305
306    def test_profile_descriptor_list_attribute(self):
307        """Implementation of test TP/SERVER/SA/BV-17-C from SDP Specification.
308
309        @raise error.TestFail: If list attribute not correct form.
310
311        """
312        profile_list = self.get_attribute(self.PNP_INFORMATION_CLASS_ID,
313                                          self.PROFILE_DESCRIPTOR_LIST_ATTR_ID)
314
315        if not isinstance(profile_list, list):
316            raise error.TestFail('Value is not a list')
317        self.assert_equal(len(profile_list), 1)
318
319        if not isinstance(profile_list[0], list):
320            raise error.TestFail('Item is not a list')
321        self.assert_equal(len(profile_list[0]), 2)
322
323        self.assert_equal(profile_list[0][0], self.PNP_INFORMATION_CLASS_ID)
324
325
326    def test_additional_protocol_descriptor_list_attribute(self):
327        """Implementation of test TP/SERVER/SA/BV-21-C from SDP Specification.
328
329        @raise error.TestFail: If the DUT failed the test.
330
331        """
332
333        """AVRCP is not supported by Chromebook and no need to run this test
334        self.assert_nonempty_list(
335            self.get_attribute(self.AVRCP_TG_CLASS_ID,
336                self.ADDITIONAL_PROTOCOLLIST_ATTR_ID))
337        """
338
339    def test_non_existing_attribute(self):
340        """Implementation of test TP/SERVER/SA/BV-20-C from SDP Specification.
341
342        @raise error.TestFail: If the DUT failed the test.
343        """
344        record_handle = self.get_single_handle(self.FAKE_SERVICE_CLASS_ID)
345        res = json.loads(self.tester.service_attribute_request(
346                  record_handle, self.MAX_ATTR_BYTE_CNT,
347                  [self.NON_EXISTING_ATTRIBUTE_ID], {}))
348        self.assert_equal(res, [])
349
350
351    def test_fake_attributes(self):
352        """Test values of attributes of the fake service record.
353
354        @raise error.TestFail: If the DUT failed the test.
355        """
356        for attr_id in self.FAKE_GENERAL_ATTRIBUTE_IDS:
357            self.assert_attribute_equals(self.FAKE_SERVICE_CLASS_ID,
358                                         attr_id, self.FAKE_ATTRIBUTE_VALUE)
359
360        for offset in self.FAKE_LANGUAGE_ATTRIBUTE_OFFSETS:
361            record_handle = self.get_single_handle(self.FAKE_SERVICE_CLASS_ID)
362
363            lang_base = json.loads(self.tester.service_attribute_request(
364                            record_handle, self.MAX_ATTR_BYTE_CNT,
365                            [self.LANGUAGE_BASE_ATTRIBUTE_ID], {}))
366            attr_id = lang_base[1] + offset
367
368            response = json.loads(self.tester.service_attribute_request(
369                record_handle, self.MAX_ATTR_BYTE_CNT, [attr_id], {}))
370            self.assert_equal(response, [attr_id, self.FAKE_ATTRIBUTE_VALUE])
371
372
373    def test_invalid_record_handle(self):
374        """Implementation of test TP/SERVER/SA/BI-01-C from SDP Specification.
375
376        @raise error.TestFail: If the DUT failed the test.
377        """
378        res = json.loads(self.tester.service_attribute_request(
379                          self.INVALID_RECORD_HANDLE, self.MAX_ATTR_BYTE_CNT,
380                          [self.NON_EXISTING_ATTRIBUTE_ID], {}))
381        self.assert_equal(res, self.ERROR_CODE_INVALID_RECORD_HANDLE)
382
383
384    def test_invalid_request_syntax(self):
385        """Implementation of test TP/SERVER/SA/BI-02-C from SDP Specification.
386
387        @raise error.TestFail: If the DUT failed the test.
388        """
389        record_handle = self.get_single_handle(self.SDP_SERVER_CLASS_ID)
390        res = json.loads(self.tester.service_attribute_request(
391                          record_handle,
392                          self.MAX_ATTR_BYTE_CNT,
393                          [self.SERVICE_RECORD_HANDLE_ATTR_ID],
394                          {'invalid_request':self.INVALID_SYNTAX_REQUEST}))
395        self.assert_equal(res, self.ERROR_CODE_INVALID_SYNTAX)
396
397
398    def test_invalid_pdu_size(self):
399        """Implementation of test TP/SERVER/SA/BI-03-C from SDP Specification.
400
401        @raise error.TestFail: If the DUT failed the test.
402        """
403        opts = dict({'forced_pdu_size':self.INVALID_PDU_SIZE})
404        record_handle = self.get_single_handle(self.SDP_SERVER_CLASS_ID)
405        res = json.loads(self.tester.service_attribute_request(record_handle,
406            self.MAX_ATTR_BYTE_CNT, [self.SERVICE_RECORD_HANDLE_ATTR_ID], opts))
407        self.assert_equal(res, self.ERROR_CODE_INVALID_PDU_SIZE)
408
409
410    def correct_request_att_request_test(self):
411        """Run basic tests for Service Attribute Request."""
412        # Connect to the DUT via L2CAP using SDP socket.
413        self.tester.connect(self.adapter['Address'])
414
415        self.test_record_handle_attribute()
416        self.test_browse_group_attribute()
417        self.test_icon_url_attribute()
418        self.test_documentation_url_attribute()
419        self.test_client_executable_url_attribute()
420        self.test_protocol_descriptor_list_attribute()
421        self.test_continuation_state()
422        self.test_version_list_attribute()
423        self.test_service_database_state_attribute()
424        self.test_profile_descriptor_list_attribute()
425        self.test_additional_protocol_descriptor_list_attribute()
426        self.test_fake_attributes()
427        self.test_non_existing_attribute()
428        self.test_invalid_record_handle()
429        self.test_invalid_pdu_size()
430        self.test_invalid_request_syntax()
431
432
433    def sdp_service_attribute_request_test(self, device):
434        """Runs service attribute request test"""
435
436        if self.host.btpeer.get_platform() != 'RASPI':
437            raise error.TestNAError('Test only runs on Raspi')
438
439        self.tester = device
440        # Reset the adapter to the powered on, discoverable state.
441        if not self.bluetooth_facade.reset_on():
442            raise error.TestFail('DUT adapter could not be powered on')
443        if not self.bluetooth_facade.set_discoverable(True):
444            raise error.TestFail('DUT could not be set as discoverable')
445
446        self.adapter = self.bluetooth_facade.get_adapter_properties()
447
448        # Create a fake service record in order to test attributes,
449        # that are not present in any of existing services.
450        uuid128 = ((self.FAKE_SERVICE_CLASS_ID << 96) +
451                   self.BLUETOOTH_BASE_UUID)
452        uuid_str = str(uuid.UUID(int=uuid128))
453        sdp_record = self.build_service_record()
454        self.bluetooth_facade.register_profile(self.FAKE_SERVICE_PATH,
455                                     uuid_str,
456                                     {"ServiceRecord": sdp_record})
457
458        # Setup the tester as a generic computer.
459        if not self.tester.setup('computer'):
460            raise error.TestNAError('Tester could not be initialized')
461
462        self.correct_request_att_request_test()
463
464
465
466class bluetooth_SDP_ServiceBrowse(bluetooth_SDP_Test,
467    bluetooth_adapter_tests.BluetoothAdapterTests):
468    """
469    Verify that the IUT behave correct during Service Browse procedure.
470    """
471    version = 1
472
473    MAX_BROWSE_REC_CNT             = 100
474    MAX_ATTR_BYTE_CNT       = 300
475    SERVICE_CLASS_ID_LIST   = 0x0001
476    BROWSE_GROUP_DESCRIPTOR = 0x1001
477    GROUP_ID                = 0x0200
478
479
480    def get_attribute_ssr_sar(self, class_id, attr_id, size):
481        """Get service attributes using Service Search Request and Service
482        Attribute Request.
483
484        @param class_id: Class ID of service to check.
485        @param attr_id: ID of attribute to check.
486        @param size: Preferred size of UUID.
487
488        @return attribute value if attribute exists, None otherwise
489
490        """
491        handles = json.loads(self.tester.service_search_request(
492                      [class_id], self.MAX_BROWSE_REC_CNT,
493                      {'preferred_size':size}))
494
495        if not (isinstance(handles, list) and len(handles) > 0):
496            return None
497
498        res = []
499        for record_handle in handles:
500            value = json.loads(self.tester.service_attribute_request(
501                        record_handle, self.MAX_ATTR_BYTE_CNT, [attr_id], {}))
502            if not (isinstance(value, list) and len(value) == 2 and
503                value[0] == attr_id):
504                return None
505            res.append(value[1])
506
507        return res
508
509
510    def get_attribute_ssar(self, class_id, attr_id, size):
511        """Get service attributes using Service Search Attribute Request.
512
513        @param class_id: Class ID of service to check.
514        @param attr_id: ID of attribute to check.
515        @param size: Preferred size of UUID.
516
517        @return attribute value if attribute exists, None otherwise
518
519        """
520        response = json.loads(self.tester.service_search_attribute_request(
521                       [class_id], self.MAX_ATTR_BYTE_CNT, [attr_id],
522                       {'preferred_size':size}))
523
524        if not isinstance(response, list):
525            return None
526
527        res = []
528        for elem in response:
529            if not (isinstance(elem, list) and len(elem) == 2 and
530                    elem[0] == attr_id):
531                return None
532            res.append(elem[1])
533
534        return res
535
536
537    def test_attribute(self, class_id, attr_id, get_attribute):
538        """Test service attributes using 16-bit, 32-bit and 128-bit
539        size of UUID.
540
541        @param class_id: Class ID of service to check.
542        @param attr_id: ID of attribute to check.
543        @param get_attribute: Method to use to get an attribute value.
544
545        @return attribute value if attribute exists and values from three tests
546        are equal, None otherwise
547
548        """
549        result_16 = get_attribute(class_id, attr_id, 16)
550
551        for size in 32, 128:
552            result_cur = get_attribute(class_id, attr_id, size)
553            if result_16 != result_cur:
554                return None
555
556        return result_16
557
558
559    def service_browse(self, get_attribute):
560        """Execute a Service Browse procedure.
561
562        @param get_attribute: Method to use to get an attribute value.
563
564        @return sorted list of unique services on the DUT, or False if browse
565        did not finish correctly
566
567        """
568        # Find services on top of hierarchy.
569        root_services = self.test_attribute(self.PUBLIC_BROWSE_ROOT,
570                                            self.SERVICE_CLASS_ID_LIST,
571                                            get_attribute)
572
573        if not root_services:
574            return False
575
576        # Find additional browse groups.
577        group_ids = self.test_attribute(self.BROWSE_GROUP_DESCRIPTOR,
578                                        self.GROUP_ID,
579                                        get_attribute)
580        if not group_ids:
581            return False
582
583        # Find services from all browse groups.
584        all_services = []
585        for group_id in group_ids:
586            services = self.test_attribute(group_id,
587                                           self.SERVICE_CLASS_ID_LIST,
588                                           get_attribute)
589            if not services:
590                return False
591            all_services.extend(services)
592
593        # Ensure that root services are among all services.
594        for service in root_services:
595            if service not in all_services:
596                return False
597
598        # Sort all services and remove duplicates.
599        all_services.sort()
600        last = 0
601        for service in all_services[1:]:
602            if all_services[last] != service:
603                last += 1
604                all_services[last] = service
605
606        return all_services[:last + 1]
607
608
609    def correct_request_browse_test(self):
610        """Run basic tests for Service Browse procedure.
611
612        @return True if all tests finishes correctly, False otherwise
613
614        """
615
616        # Connect to the DUT via L2CAP using SDP socket.
617        self.tester.connect(self.adapter['Address'])
618
619        browse_ssar = self.service_browse(self.get_attribute_ssar)
620        if not browse_ssar:
621            return False
622
623        browse_ssr_sar = self.service_browse(self.get_attribute_ssr_sar)
624
625
626        # Ensure that two different browse methods return the same results.
627        return browse_ssar == browse_ssr_sar
628
629
630    def sdp_service_browse_test(self, device):
631        """Runs service browse test"""
632
633        if self.host.btpeer.get_platform() != 'RASPI':
634            raise error.TestNAError('Test only runs on Raspi')
635
636        self.tester = device
637        # Reset the adapter to the powered on, discoverable state.
638        if not (self.bluetooth_facade.reset_on() and
639                self.bluetooth_facade.set_discoverable(True)):
640            raise error.TestFail('DUT could not be reset to initial state')
641
642        self.adapter = self.bluetooth_facade.get_adapter_properties()
643
644        # Setup the tester as a generic computer.
645        if not self.tester.setup('computer'):
646            raise error.TestNAError('Tester could not be initialized')
647
648        # Since radio is involved, this test is not 100% reliable; instead we
649        # repeat a few times until it succeeds.
650        for failed_attempts in range(0, 5):
651            if self.correct_request_browse_test():
652                break
653        else:
654            raise error.TestFail('Expected device was not found')
655
656        # Record how many attempts this took, hopefully we'll one day figure out
657        # a way to reduce this to zero and then the loop above can go away.
658        self.write_perf_keyval({'failed_attempts': failed_attempts })
659
660
661class bluetooth_SDP_ServiceSearchAttributeRequest(bluetooth_SDP_Test,
662    bluetooth_adapter_tests.BluetoothAdapterTests):
663    """
664    Verify the correct behaviour of the device when searching for services and
665    attributes.
666    """
667
668
669    NON_EXISTING_SERVICE_CLASS_ID    = 0x9875
670    PUBLIC_BROWSE_GROUP_CLASS_ID     = 0x1002
671
672    NON_EXISTING_ATTRIBUTE_ID        = 0xABCD
673    SERVICE_CLASS_ID_ATTRIBUTE_ID    = 0x0001
674    VERSION_NUMBER_LIST_ATTR_ID      = 0x0200
675
676
677    def fail_test(self, testname, value):
678        """Raise an error for a particular SDP test.
679
680        @param testname: a string representation of the test name.
681        @param value: the value that did not pass muster.
682
683        """
684        raise error.TestFail('SDP test %s failed: got %s.' % (testname, value))
685
686
687    def test_non_existing(self, class_id, attr_id):
688        """Check that a single attribute of a single service does not exist
689
690        @param class_id: Class ID of service to check.
691        @param attr_id: ID of attribute to check.
692
693        @raises error.TestFail if service or attribute does exists.
694
695        """
696        for size in 16, 32, 128:
697            result = json.loads(self.tester.service_search_attribute_request(
698                         [class_id],
699                         self.MAX_ATTR_BYTE_CNT,
700                         [attr_id],
701                         {'preferred_size':size}))
702            if result != []:
703                raise error.TestFail('Attribute %s of class %s exists when it '
704                                     'should not!' % (class_id, attr_id))
705
706
707    def get_attribute_sssar(self, class_id, attr_id, size):
708        """Get a single attribute of a single service using Service Search
709        Attribute Request.
710
711        @param class_id: Class ID of service to check.
712        @param attr_id: ID of attribute to check.
713        @param size: Preferred size of UUID.
714
715        @return attribute value if attribute exists
716
717        @raises error.TestFail if attribute does not exist
718
719        """
720        res = json.loads(self.tester.service_search_attribute_request(
721                  [class_id], self.MAX_ATTR_BYTE_CNT, [attr_id],
722                  {'preferred_size':size}))
723
724        if (isinstance(res, list) and len(res) == 1 and
725            isinstance(res[0], list) and len(res[0]) == 2 and
726            res[0][0] == attr_id):
727            return res[0][1]
728
729        raise error.TestFail('Attribute %s of class %s does not exist! (size '
730                             '%s)' % (class_id, attr_id, size))
731
732
733    def test_attribute_sssar(self, class_id, attr_id):
734        """Test a single attribute of a single service using 16-bit, 32-bit and
735        128-bit size of UUID.
736
737        @param class_id: Class ID of service to check.
738        @param attr_id: ID of attribute to check.
739
740        @return attribute value if attribute exists and values from three tests
741        are equal
742
743        @raises error.TestFail if attribute doesn't exist or values not equal
744
745        """
746        result_16 = self.get_attribute_sssar(class_id, attr_id, 16)
747        for size in 32, 128:
748            result_cur = self.get_attribute_sssar(class_id, attr_id, size)
749            if result_16 != result_cur:
750                raise error.TestFail('Attribute test failed %s: expected %s, '
751                                     'got %s' % (size, result_16, result_cur))
752
753        return result_16
754
755
756    def test_non_existing_service(self):
757        """Implementation of test TP/SERVER/SSA/BV-01-C from SDP Specification.
758
759        @raises error.TestFail if test fails
760
761        """
762        self.test_non_existing(self.NON_EXISTING_SERVICE_CLASS_ID,
763                               self.SERVICE_CLASS_ID_ATTRIBUTE_ID)
764
765
766    def test_non_existing_attribute_sssar(self):
767        """Implementation of test TP/SERVER/SSA/BV-02-C from SDP Specification.
768
769        @raises error.TestFail if test fails
770
771        """
772        self.test_non_existing(self.PUBLIC_BROWSE_GROUP_CLASS_ID,
773                               self.NON_EXISTING_ATTRIBUTE_ID)
774
775
776    def test_non_existing_service_attribute(self):
777        """Implementation of test TP/SERVER/SSA/BV-03-C from SDP Specification.
778
779        @raises error.TestFail if test fails
780
781        """
782        self.test_non_existing(self.NON_EXISTING_SERVICE_CLASS_ID,
783                               self.NON_EXISTING_ATTRIBUTE_ID)
784
785
786    def test_existing_service_attribute(self):
787        """Implementation of test TP/SERVER/SSA/BV-04-C from SDP Specification.
788
789        @raises error.TestFail if test fails
790
791        """
792        value = self.test_attribute_sssar(self.SDP_SERVER_CLASS_ID,
793                                    self.SERVICE_CLASS_ID_ATTRIBUTE_ID)
794        if not value == [self.SDP_SERVER_CLASS_ID]:
795            self.fail_test('TP/SERVER/SSA/BV-04-C', value)
796
797
798    def test_service_database_state_attribute_sssar(self):
799        """Implementation of test TP/SERVER/SSA/BV-08-C from SDP Specification.
800
801        @raises error.TestFail if test fails
802
803        """
804        value = self.test_attribute_sssar(self.SDP_SERVER_CLASS_ID,
805                                    self.SERVICE_DATABASE_STATE_ATTR_ID)
806        if not isinstance(value, int):
807            self.fail_test('TP/SERVER/SSA/BV-08-C', value)
808
809
810    def test_protocol_descriptor_list_attribute_sssar(self):
811        """Implementation of test TP/SERVER/SSA/BV-11-C from SDP Specification.
812
813        @raises error.TestFail if test fails
814
815        """
816        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
817                                    self.PROTOCOL_DESCRIPTOR_LIST_ATTR_ID)
818
819        # The first-layer protocol is L2CAP, using the PSM for ATT protocol.
820        if value[0] != [self.L2CAP_UUID, self.ATT_PSM]:
821            self.fail_test('TP/SERVER/SSA/BV-11-C', value)
822
823        # The second-layer protocol is ATT. The additional parameters are
824        # ignored, since they may reasonably vary between implementations.
825        if value[1][0] != self.ATT_UUID:
826            self.fail_test('TP/SERVER/SSA/BV-11-C', value)
827
828
829
830    def test_browse_group_attribute_sssar(self):
831        """Implementation of test TP/SERVER/SSA/BV-12-C from SDP Specification.
832
833        @raises error.TestFail if test fails
834
835        """
836        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
837                                    self.BROWSE_GROUP_LIST_ATTR_ID)
838        if not value == [self.PUBLIC_BROWSE_ROOT]:
839            self.fail_test('TP/SERVER/SSA/BV-12-C', value)
840
841
842    def test_icon_url_attribute_sssar(self):
843        """Implementation of test TP/SERVER/SSA/BV-15-C from SDP Specification.
844
845        @raises error.TestFail if test fails
846
847        """
848        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
849                                    self.ICON_URL_ATTR_ID)
850        if not value == self.BLUEZ_URL:
851            self.fail_test('TP/SERVER/SSA/BV-15-C', value)
852
853
854    def test_version_list_attribute_sssar(self):
855        """Implementation of test TP/SERVER/SSA/BV-19-C from SDP Specification.
856
857        @raises error.TestFail if test fails
858
859        """
860        value = self.test_attribute_sssar(self.SDP_SERVER_CLASS_ID,
861                                    self.VERSION_NUMBER_LIST_ATTR_ID)
862        if not isinstance(value, list) and value != []:
863            self.fail_test('TP/SERVER/SSA/BV-19-C', value)
864
865
866    def test_profile_descriptor_list_attribute_sssar(self):
867        """Implementation of test TP/SERVER/SSA/BV-20-C from SDP Specification.
868
869        @raises error.TestFail if test fails
870
871        """
872        value = self.test_attribute_sssar(self.PNP_INFORMATION_CLASS_ID,
873                                    self.PROFILE_DESCRIPTOR_LIST_ATTR_ID)
874        if not (isinstance(value, list) and len(value) == 1 and
875                isinstance(value[0], list) and len(value[0]) == 2 and
876                value[0][0] == self.PNP_INFORMATION_CLASS_ID):
877            self.fail_test('TP/SERVER/SSA/BV-20-C', value)
878
879
880    def test_documentation_url_attribute_sssar(self):
881        """Implementation of test TP/SERVER/SSA/BV-21-C from SDP Specification.
882
883        @raises error.TestFail if test fails
884
885        """
886        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
887                                    self.DOCUMENTATION_URL_ATTR_ID)
888        if not value == self.BLUEZ_URL:
889            self.fail_test('TP/SERVER/SSA/BV-21-C', value)
890
891
892    def test_client_executable_url_attribute_sssar(self):
893        """Implementation of test TP/SERVER/SSA/BV-22-C from SDP Specification.
894
895        @raises error.TestFail if test fails
896
897        """
898        value = self.test_attribute_sssar(self.GAP_CLASS_ID,
899                                    self.CLIENT_EXECUTABLE_URL_ATTR_ID)
900        if not value == self.BLUEZ_URL:
901            self.fail_test('TP/SERVER/SSA/BV-22-C', value)
902
903
904    def test_additional_protocol_descriptor_list_attribute_sssar(self):
905        """Implementation of test TP/SERVER/SSA/BV-23-C from SDP Specification.
906
907        @raises error.TestFail if test fails
908
909        """
910
911        """AVRCP is not supported by Chromebook and no need to run this test
912        value = self.test_attribute_sssar(self.AVRCP_TG_CLASS_ID,
913                                    self.ADDITIONAL_PROTOCOLLIST_ATTR_ID)
914        if not isinstance(value, list) and value != []:
915            self.fail_test('TP/SERVER/SSA/BV-23-C', value)
916        """
917
918    def test_fake_attributes_sssar(self):
919        """Test values of attributes of the fake service record.
920
921        @raises error.TestFail if test fails
922
923        """
924        for attr_id in self.FAKE_GENERAL_ATTRIBUTE_IDS:
925            value = self.test_attribute_sssar(self.FAKE_SERVICE_CLASS_ID,
926                                              attr_id)
927            if value != self.FAKE_ATTRIBUTE_VALUE:
928                self.fail_test('fake service attributes', value)
929
930        for offset in self.FAKE_LANGUAGE_ATTRIBUTE_OFFSETS:
931            lang_base = self.test_attribute_sssar(self.FAKE_SERVICE_CLASS_ID,
932                                            self.LANGUAGE_BASE_ATTRIBUTE_ID)
933            attr_id = lang_base + offset
934            value = self.test_attribute_sssar(self.FAKE_SERVICE_CLASS_ID,
935                                              attr_id)
936            if value != self.FAKE_ATTRIBUTE_VALUE:
937                self.fail_test('fake service attributes', value)
938
939
940    def test_continuation_state_sssar(self):
941        """Implementation of test TP/SERVER/SSA/BV-06-C from SDP Specification.
942
943        @raises error.TestFail if test fails
944
945        """
946        for size in 16, 32, 128:
947            # This request should generate a long response, which will be
948            # split into 98 chunks.
949            value = json.loads(self.tester.service_search_attribute_request(
950                        [self.PUBLIC_BROWSE_GROUP_CLASS_ID],
951                        self.MIN_ATTR_BYTE_CNT,
952                        [[0, 0xFFFF]], {'preferred_size':size}))
953            if not isinstance(value, list) or value == []:
954                self.fail_test('TP/SERVER/SSA/BV-06-C', value)
955
956
957    def test_invalid_request_syntax_sssar(self):
958        """Implementation of test TP/SERVER/SSA/BI-01-C from SDP Specification.
959
960        @raises error.TestFail if test fails
961
962        """
963        for size in 16, 32, 128:
964            value = json.loads(self.tester.service_search_attribute_request(
965                        [self.SDP_SERVER_CLASS_ID],
966                        self.MAX_ATTR_BYTE_CNT,
967                        [self.SERVICE_CLASS_ID_ATTRIBUTE_ID],
968                        {'preferred_size':size,
969                        'invalid_request':'9875'}))
970            if value != self.ERROR_CODE_INVALID_SYNTAX:
971                self.fail_test('TP/SERVER/SSA/BI-01-C', value)
972
973
974    def test_invalid_pdu_size_sssar(self):
975        """Implementation of test TP/SERVER/SSA/BI-02-C from SDP Specification.
976
977        @raises error.TestFail if test fails
978
979        """
980        for size in 16, 32, 128:
981            value = json.loads(self.tester.service_search_attribute_request(
982                        [self.SDP_SERVER_CLASS_ID],
983                        self.MAX_ATTR_BYTE_CNT,
984                        [self.SERVICE_CLASS_ID_ATTRIBUTE_ID],
985                        {'preferred_size':size,
986                        'forced_pdu_size':100}))
987            if value != self.ERROR_CODE_INVALID_PDU_SIZE:
988                self.fail_test('TP/SERVER/SSA/BI-02-C', value)
989
990
991    def correct_request_search_att_test(self):
992        """Run tests for Service Search Attribute request.
993
994        @raises error.TestFail if any test fails
995
996        """
997        # connect to the DUT via L2CAP using SDP socket
998        self.tester.connect(self.adapter['Address'])
999
1000        self.test_non_existing_service()
1001        self.test_non_existing_attribute_sssar()
1002        self.test_non_existing_service_attribute()
1003        #self.test_existing_service_attribute()
1004        self.test_service_database_state_attribute_sssar()
1005        self.test_protocol_descriptor_list_attribute_sssar()
1006        self.test_browse_group_attribute_sssar()
1007        self.test_icon_url_attribute_sssar()
1008        self.test_version_list_attribute_sssar()
1009        self.test_profile_descriptor_list_attribute_sssar()
1010        self.test_documentation_url_attribute_sssar()
1011        self.test_client_executable_url_attribute_sssar()
1012        self.test_additional_protocol_descriptor_list_attribute_sssar()
1013        self.test_fake_attributes_sssar()
1014        self.test_continuation_state_sssar()
1015        self.test_invalid_request_syntax_sssar()
1016        self.test_invalid_pdu_size_sssar()
1017        logging.info('correct_request finished successfully!')
1018
1019
1020    def sdp_service_search_attribute_request_test(self, device):
1021        """Runs service search attribute request test"""
1022
1023        if self.host.btpeer.get_platform() != 'RASPI':
1024            raise error.TestNAError('Test only runs on Raspi')
1025
1026        self.tester = device
1027        # Reset the adapter to the powered on, discoverable state.
1028        if not self.bluetooth_facade.reset_on():
1029            raise error.TestFail('DUT adapter could not be powered on')
1030        if not self.bluetooth_facade.set_discoverable(True):
1031            raise error.TestFail('DUT could not be set as discoverable')
1032
1033        self.adapter = self.bluetooth_facade.get_adapter_properties()
1034
1035        # Create a fake service record in order to test attributes,
1036        # that are not present in any of existing services.
1037        uuid128 = ((self.FAKE_SERVICE_CLASS_ID << 96) +
1038                   self.BLUETOOTH_BASE_UUID)
1039        uuid_str = str(uuid.UUID(int=uuid128))
1040        sdp_record = self.build_service_record()
1041        self.bluetooth_facade.register_profile(self.FAKE_SERVICE_PATH,
1042                                     uuid_str,
1043                                     {"ServiceRecord": sdp_record})
1044
1045        # Setup the tester as a generic computer.
1046        if not self.tester.setup('computer'):
1047            raise error.TestNAError('Tester could not be initialized')
1048
1049        # Since radio is involved, this test is not 100% reliable; instead we
1050        # repeat a few times until it succeeds.
1051        passing = False
1052        for failed_attempts in range(0, 4):
1053            try:
1054                self.correct_request_search_att_test()
1055                passing = True
1056            except error.TestFail as e:
1057                logging.warning('Ignoring error: %s', e)
1058            if passing:
1059                break
1060        else:
1061            self.correct_request_search_att_test()
1062
1063        # Record how many attempts this took, hopefully we'll one day figure out
1064        # a way to reduce this to zero and then the loop above can go away.
1065        self.write_perf_keyval({'failed_attempts': failed_attempts})
1066
1067
1068class bluetooth_SDP_ServiceSearchRequestBasic(
1069    bluetooth_adapter_tests.BluetoothAdapterTests):
1070    """
1071    Verify the correct behaviour of the device when searching for services.
1072    """
1073    version = 1
1074
1075    SDP_SERVER_CLASS_ID                = 0x1000
1076    NO_EXISTING_SERVICE_CLASS_ID       = 0x0001
1077    FAKE_SERVICES_CNT                  = 300
1078    FAKE_SERVICES_PATH                 = '/autotest/fake_service_'
1079    FAKE_SERVICES_CLASS_ID             = 0xABCD
1080    BLUETOOTH_BASE_UUID                = 0x0000000000001000800000805F9B34FB
1081    SSRB_INVALID_PDU_SIZE              = 9875
1082    ERROR_CODE_INVALID_REQUEST_SYNTAX  = 0x0003
1083    ERROR_CODE_INVALID_PDU_SIZE        = 0x0004
1084
1085
1086    def correct_request_basic_test(self):
1087        """Search the existing service on the DUT using the Tester.
1088
1089        @return True if found, False if not found
1090
1091        """
1092        # connect to the DUT via L2CAP using SDP socket
1093        self.tester.connect(self.adapter['Address'])
1094
1095        for size in 16, 32, 128:
1096            # test case TP/SERVER/SS/BV-01-C:
1097            # at least the SDP server service exists
1098            resp = json.loads(self.tester.service_search_request(
1099                              [self.SDP_SERVER_CLASS_ID], 3,
1100                              {'preferred_size':size}))
1101            if resp != [0]:
1102                return False
1103            # test case TP/SERVER/SS/BV-04-C:
1104            # Service with Class ID = 0x0001 should never exist, as this UUID is
1105            # reserved as Bluetooth Core Specification UUID
1106            resp = json.loads(self.tester.service_search_request(
1107                              [self.NO_EXISTING_SERVICE_CLASS_ID], 3,
1108                              {'preferred_size':size}))
1109            if resp != []:
1110                return False
1111            # test case TP/SERVER/SS/BV-03-C:
1112            # request the fake services' Class ID to force SDP to use
1113            # continuation state
1114            resp = json.loads(self.tester.service_search_request(
1115                              [self.FAKE_SERVICES_CLASS_ID],
1116                              self.FAKE_SERVICES_CNT * 2,
1117                              {'preferred_size':size}))
1118            if len(resp) != self.FAKE_SERVICES_CNT:
1119                return False
1120            # test case TP/SERVER/SS/BI-01-C:
1121            # send a Service Search Request with intentionally invalid PDU size
1122            resp = json.loads(self.tester.service_search_request(
1123                              [self.SDP_SERVER_CLASS_ID], 3,
1124                              {'preferred_size':size,
1125                              'forced_pdu_size':self.SSRB_INVALID_PDU_SIZE}))
1126            if resp != self.ERROR_CODE_INVALID_PDU_SIZE:
1127                return False
1128            # test case TP/SERVER/SS/BI-02-C:
1129            # send a Service Search Request with invalid syntax
1130            resp = json.loads(self.tester.service_search_request(
1131                              [self.SDP_SERVER_CLASS_ID], 3,
1132                              {'preferred_size':size, 'invalid_request':True}))
1133            if resp != self.ERROR_CODE_INVALID_REQUEST_SYNTAX:
1134                return False
1135
1136        return True
1137
1138
1139    def sdp_service_search_request_basic_test(self, device):
1140        """Runs service search request basic test"""
1141
1142        if self.host.btpeer.get_platform() != 'RASPI':
1143            raise error.TestNAError('Test only runs on Raspi')
1144
1145        self.tester = device
1146        # Reset the adapter to the powered on, discoverable state.
1147        if not (self.bluetooth_facade.reset_on() and
1148                self.bluetooth_facade.set_discoverable(True)):
1149            raise error.TestFail('DUT could not be reset to initial state')
1150
1151        self.adapter = self.bluetooth_facade.get_adapter_properties()
1152
1153        # Setup the tester as a generic computer.
1154        if not self.tester.setup('computer'):
1155            raise error.TestNAError('Tester could not be initialized')
1156
1157        # Create many fake services with the same Class ID
1158        for num in range(0, self.FAKE_SERVICES_CNT):
1159            path_str = self.FAKE_SERVICES_PATH + str(num)
1160            uuid128 = ((self.FAKE_SERVICES_CLASS_ID << 96) +
1161                      self.BLUETOOTH_BASE_UUID)
1162            uuid_str = str(uuid.UUID(int=uuid128))
1163            self.bluetooth_facade.register_profile(path_str, uuid_str, {})
1164
1165        # Since radio is involved, this test is not 100% reliable; instead we
1166        # repeat a few times until it succeeds.
1167        for failed_attempts in range(0, 5):
1168            if self.correct_request_basic_test():
1169                break
1170        else:
1171            raise error.TestFail('Expected device was not found')
1172
1173        # Record how many attempts this took, hopefully we'll one day figure out
1174        # a way to reduce this to zero and then the loop above can go away.
1175        self.write_perf_keyval({'failed_attempts': failed_attempts })
1176
1177
1178class BluetoothSDPTests(bluetooth_SDP_ServiceAttributeRequest,
1179                        bluetooth_SDP_ServiceBrowse,
1180                        bluetooth_SDP_ServiceSearchAttributeRequest,
1181                        bluetooth_SDP_ServiceSearchRequestBasic):
1182    """Derived class that simplifies inheritance of sdp tests"""
1183    pass
1184