gd/hal/cert/simple_hal_test_lib.py

#!/usr/bin/env python3
#
#   Copyright 2019 - The Android Open Source Project
#
#   Licensed under the Apache License, Version 2.0 (the "License");
#   you may not use this file except in compliance with the License.
#   You may obtain a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#   Unless required by applicable law or agreed to in writing, software
#   distributed under the License is distributed on an "AS IS" BASIS,
#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#   See the License for the specific language governing permissions and
#   limitations under the License.

from datetime import timedelta

from cert.event_stream import EventStream
from cert.truth import assertThat
from cert.py_hal import PyHal
from cert.matchers import HciMatchers
from cert.captures import HciCaptures
from google.protobuf import empty_pb2
from facade import rootservice_pb2 as facade_rootservice_pb2
from hal import hal_facade_pb2 as hal_facade_pb2
from bluetooth_packets_python3 import hci_packets
import bluetooth_packets_python3 as bt_packets
from bluetooth_packets_python3.hci_packets import AclBuilder
from bluetooth_packets_python3 import RawBuilder

_GRPC_TIMEOUT = 10


class SimpleHalTestBase():

    def setup_test(self, dut, cert):
        self.dut_hal = PyHal(dut)
        self.cert_hal = PyHal(cert)

        self.dut_hal.reset()
        self.cert_hal.reset()

    def teardown_test(self):
        self.dut_hal.close()
        self.cert_hal.close()

    def test_stream_events(self):
        self.dut_hal.send_hci_command(
            hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM, '0C:05:04:03:02:01'))

        assertThat(self.dut_hal.get_hci_event_stream()).emits(
            HciMatchers.Exactly(hci_packets.LeAddDeviceToConnectListCompleteBuilder(1, hci_packets.ErrorCode.SUCCESS)))

    def test_loopback_hci_command(self):
        self.dut_hal.send_hci_command(hci_packets.WriteLoopbackModeBuilder(hci_packets.LoopbackMode.ENABLE_LOCAL))

        command = hci_packets.LeAddDeviceToConnectListBuilder(hci_packets.ConnectListAddressType.RANDOM,
                                                              '0C:05:04:03:02:01')
        self.dut_hal.send_hci_command(command)

        assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LoopbackOf(command))

    def test_inquiry_from_dut(self):
        self.cert_hal.send_hci_command(hci_packets.WriteScanEnableBuilder(hci_packets.ScanEnable.INQUIRY_AND_PAGE_SCAN))

        lap = hci_packets.Lap()
        lap.lap = 0x33
        self.dut_hal.send_hci_command(hci_packets.InquiryBuilder(lap, 0x30, 0xff))

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'\x02\x0f' in packet.payload
                                                              # Expecting an HCI Event (code 0x02, length 0x0f)
                                                             )

    def test_le_ad_scan_cert_advertises(self):
        self.dut_hal.unmask_event(hci_packets.EventCode.LE_META_EVENT)
        self.dut_hal.set_random_le_address('0D:05:04:03:02:01')

        self.dut_hal.set_scan_parameters()
        self.dut_hal.start_scanning()

        advertisement = self.cert_hal.create_advertisement(
            0,
            '0C:05:04:03:02:01',
            min_interval=512,
            max_interval=768,
            peer_address='A6:A5:A4:A3:A2:A1',
            tx_power=0x7f,
            sid=1)
        advertisement.set_data(b'Im_A_Cert')
        advertisement.start()

        assertThat(self.dut_hal.get_hci_event_stream()).emits(lambda packet: b'Im_A_Cert' in packet.payload)

        advertisement.stop()

        self.dut_hal.stop_scanning()

    def test_le_connection_dut_advertises(self):
        self.cert_hal.unmask_event(hci_packets.EventCode.LE_META_EVENT)
        self.cert_hal.set_random_le_address('0C:05:04:03:02:01')
        self.cert_hal.initiate_le_connection('0D:05:04:03:02:01')

        # DUT Advertises
        self.dut_hal.unmask_event(hci_packets.EventCode.LE_META_EVENT)
        advertisement = self.dut_hal.create_advertisement(0, '0D:05:04:03:02:01')
        advertisement.set_data(b'Im_The_DUT')
        advertisement.set_scan_response(b'Im_The_D')
        advertisement.start()

        cert_acl = self.cert_hal.complete_le_connection()
        dut_acl = self.dut_hal.complete_le_connection()

        dut_acl.send_first(b'Just SomeAclData')
        cert_acl.send_first(b'Just SomeMoreAclData')

        assertThat(self.cert_hal.get_acl_stream()).emits(lambda packet: b'SomeAclData' in packet.payload)
        assertThat(self.dut_hal.get_acl_stream()).emits(lambda packet: b'SomeMoreAclData' in packet.payload)

    def test_le_connect_list_connection_cert_advertises(self):
        self.dut_hal.unmask_event(hci_packets.EventCode.LE_META_EVENT)
        self.dut_hal.set_random_le_address('0D:05:04:03:02:01')
        self.dut_hal.add_to_connect_list('0C:05:04:03:02:01')
        self.dut_hal.initiate_le_connection_by_connect_list('BA:D5:A4:A3:A2:A1')

        self.cert_hal.unmask_event(hci_packets.EventCode.LE_META_EVENT)
        advertisement = self.cert_hal.create_advertisement(
            1,
            '0C:05:04:03:02:01',
            min_interval=512,
            max_interval=768,
            peer_address='A6:A5:A4:A3:A2:A1',
            tx_power=0x7F,
            sid=0)
        advertisement.set_data(b'Im_A_Cert')
        advertisement.start()

        assertThat(self.cert_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())
        assertThat(self.dut_hal.get_hci_event_stream()).emits(HciMatchers.LeConnectionComplete())