path: root/gd/l2cap/le/cert/le_l2cap_test.py

#
#   Copyright 2020 - 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 cert.gd_base_test import GdBaseTestClass
from cert.truth import assertThat
from cert.py_l2cap import PyLeL2cap
from cert.matchers import L2capMatchers
from cert.metadata import metadata
from facade import common_pb2 as common
from google.protobuf import empty_pb2 as empty_proto
from hci.facade import le_acl_manager_facade_pb2 as le_acl_manager_facade
from hci.facade import le_advertising_manager_facade_pb2 as le_advertising_facade
from hci.facade import le_initiator_address_facade_pb2 as le_initiator_address_facade
import bluetooth_packets_python3 as bt_packets
from bluetooth_packets_python3 import hci_packets, l2cap_packets
from bluetooth_packets_python3.l2cap_packets import LeCreditBasedConnectionResponseResult
from l2cap.le.cert.cert_le_l2cap import CertLeL2cap
from l2cap.le.facade_pb2 import SecurityLevel

# Assemble a sample packet.
SAMPLE_PACKET = bt_packets.RawBuilder([0x19, 0x26, 0x08, 0x17])


class LeL2capTest(GdBaseTestClass):

    def setup_class(self):
        super().setup_class(dut_module='L2CAP', cert_module='HCI_INTERFACES')

    def setup_test(self):
        super().setup_test()

        self.dut_l2cap = PyLeL2cap(self.dut)
        self.cert_l2cap = CertLeL2cap(self.cert)
        self.dut_address = common.BluetoothAddressWithType(
            address=common.BluetoothAddress(address=bytes(b'D0:05:04:03:02:01')), type=common.RANDOM_DEVICE_ADDRESS)
        self.cert_address = common.BluetoothAddressWithType(
            address=common.BluetoothAddress(address=bytes(b'C0:11:FF:AA:33:22')), type=common.RANDOM_DEVICE_ADDRESS)
        dut_privacy_policy = le_initiator_address_facade.PrivacyPolicy(
            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
            address_with_type=self.dut_address,
            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
            minimum_rotation_time=0,
            maximum_rotation_time=0)
        self.dut_l2cap._device.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(dut_privacy_policy)
        privacy_policy = le_initiator_address_facade.PrivacyPolicy(
            address_policy=le_initiator_address_facade.AddressPolicy.USE_STATIC_ADDRESS,
            address_with_type=self.cert_address,
            rotation_irk=b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00',
            minimum_rotation_time=0,
            maximum_rotation_time=0)
        self.cert_l2cap._device.hci_le_initiator_address.SetPrivacyPolicyForInitiatorAddress(privacy_policy)

    def teardown_test(self):
        self.cert_l2cap.close()
        self.dut_l2cap.close()
        super().teardown_test()

    def _setup_link_from_cert(self):
        # DUT Advertises
        gap_name = hci_packets.GapData()
        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
        gap_name.data = list(bytes(b'Im_The_DUT'))
        gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
        config = le_advertising_facade.AdvertisingConfig(
            advertisement=[gap_data],
            interval_min=512,
            interval_max=768,
            advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
            own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
            channel_map=7,
            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
        request = le_advertising_facade.CreateAdvertiserRequest(config=config)
        create_response = self.dut.hci_le_advertising_manager.CreateAdvertiser(request)
        self.cert_l2cap.connect_le_acl(self.dut_address)

    def _set_link_from_dut_and_open_channel(self,
                                            signal_id=1,
                                            scid=0x0101,
                                            psm=0x33,
                                            mtu=1000,
                                            mps=100,
                                            initial_credit=6):
        # Cert Advertises
        gap_name = hci_packets.GapData()
        gap_name.data_type = hci_packets.GapDataType.COMPLETE_LOCAL_NAME
        gap_name.data = list(bytes(b'Im_The_DUT'))
        gap_data = le_advertising_facade.GapDataMsg(data=bytes(gap_name.Serialize()))
        config = le_advertising_facade.AdvertisingConfig(
            advertisement=[gap_data],
            interval_min=512,
            interval_max=768,
            advertising_type=le_advertising_facade.AdvertisingEventType.ADV_IND,
            own_address_type=common.USE_RANDOM_DEVICE_ADDRESS,
            channel_map=7,
            filter_policy=le_advertising_facade.AdvertisingFilterPolicy.ALL_DEVICES)
        request = le_advertising_facade.CreateAdvertiserRequest(config=config)
        create_response = self.cert.hci_le_advertising_manager.CreateAdvertiser(request)
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm)
        self.cert_l2cap.wait_for_connection()
        # TODO: Currently we can only connect by using Dynamic channel API. Use fixed channel instead.
        cert_channel = self.cert_l2cap.verify_and_respond_open_channel_from_remote(psm)
        dut_channel = response_future.get_channel()
        return (dut_channel, cert_channel)

    def _open_channel_from_cert(self, signal_id=1, scid=0x0101, psm=0x33, mtu=1000, mps=100, initial_credit=6):

        dut_channel = self.dut_l2cap.register_coc(self.cert_address, psm)
        cert_channel = self.cert_l2cap.open_channel(signal_id, psm, scid, mtu, mps, initial_credit)

        return (dut_channel, cert_channel)

    def _open_channel_from_dut(self, psm=0x33):
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm)
        cert_channel = self.cert_l2cap.verify_and_respond_open_channel_from_remote(psm)
        dut_channel = response_future.get_channel()
        return (dut_channel, cert_channel)

    def _open_fixed_channel(self, cid=4):
        dut_channel = self.dut_l2cap.get_fixed_channel(cid)
        cert_channel = self.cert_l2cap.open_fixed_channel(cid)
        return (dut_channel, cert_channel)

    def test_fixed_channel_send(self):
        self.dut_l2cap.enable_fixed_channel(4)
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_fixed_channel(4)
        dut_channel.send(b'hello' * 40)
        assertThat(cert_channel).emits(L2capMatchers.Data(b'hello' * 40))

    def test_fixed_channel_receive(self):
        self.dut_l2cap.enable_fixed_channel(4)
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_fixed_channel(4)
        cert_channel.send(SAMPLE_PACKET)
        assertThat(dut_channel).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'))

    def test_connect_from_dut_and_open_dynamic_channel(self):
        """
        Internal test for GD stack only
        """
        self._set_link_from_dut_and_open_channel()

    @metadata(pts_test_id="L2CAP/LE/CPU/BV-01-C", pts_test_name="Send Connection Parameter Update Request")
    def test_send_connection_parameter_update_request(self):
        """
        Verify that the IUT is able to send the connection parameter update Request to Lower Tester when acting as a peripheral device.
        NOTE: This is an optional feature. Also if both LL central and peripheral supports 4.1+ connection parameter update, this should happen in LL only, not L2CAP
        NOTE: Currently we need to establish at least one dynamic channel to allow update.
        """
        self._setup_link_from_cert()
        self._open_channel_from_dut()
        self.dut_l2cap.update_connection_parameter()
        assertThat(self.cert_l2cap.get_control_channel()).emits(L2capMatchers.LeConnectionParameterUpdateRequest())

    @metadata(pts_test_id="L2CAP/LE/CPU/BV-02-C", pts_test_name="Accept Connection Parameter Update Request")
    def test_accept_connection_parameter_update_request(self):
        """
        Verify that the IUT is able to receive and handle a request for connection parameter update when acting as a central device.
        NOTE: Currently we need to establish at least one dynamic channel to allow update.
        """
        self._set_link_from_dut_and_open_channel()
        self.cert_l2cap.get_control_channel().send(
            l2cap_packets.ConnectionParameterUpdateRequestBuilder(2, 0x10, 0x10, 0x0a, 0x64))
        assertThat(self.cert_l2cap.get_control_channel()).emits(
            L2capMatchers.LeConnectionParameterUpdateResponse(
                l2cap_packets.ConnectionParameterUpdateResponseResult.ACCEPTED))

    @metadata(pts_test_id="L2CAP/LE/CPU/BI-01-C", pts_test_name="Reject Connection Parameter Update Parameters")
    def test_reject_connection_parameter_update_parameters(self):
        """
        Verify that the IUT is able to reject a request for connection parameter update with illegal parameters.
        NOTE: Currently we need to establish at least one dynamic channel to allow update.
        """
        self._set_link_from_dut_and_open_channel()
        self.cert_l2cap.get_control_channel().send(
            l2cap_packets.ConnectionParameterUpdateRequestBuilder(2, 0x10, 0x10, 512, 0x64))
        assertThat(self.cert_l2cap.get_control_channel()).emits(
            L2capMatchers.LeConnectionParameterUpdateResponse(
                l2cap_packets.ConnectionParameterUpdateResponseResult.REJECTED))

    @metadata(pts_test_id="L2CAP/LE/CPU/BI-02-C", pts_test_name="Reject Connection Parameter Update Request")
    def test_reject_connection_parameter_update_request(self):
        """
        Verify that the IUT is able to reject a request for connection parameter update in peripheral mode.
        """
        self._setup_link_from_cert()
        self.cert_l2cap.get_control_channel().send(
            l2cap_packets.ConnectionParameterUpdateRequestBuilder(2, 0x10, 0x10, 0x0a, 0x64))
        assertThat(self.cert_l2cap.get_control_channel()).emits(L2capMatchers.LeCommandReject())

    @metadata(pts_test_id="L2CAP/COS/CFC/BV-01-C", pts_test_name="Segmentation")
    def test_segmentation(self):
        """
        Verify that the IUT can send data segments which are larger than the LE frame size.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert(mtu=1000, mps=102)
        dut_channel.send(b'hello' * 20 + b'world')
        # The first LeInformation packet contains 2 bytes of SDU size.
        # The packet is divided into first 100 bytes from 'hellohello....'
        # and remaining 5 bytes 'world'
        assertThat(cert_channel).emits(
            L2capMatchers.FirstLeIFrame(b'hello' * 20, sdu_size=105), L2capMatchers.Data(b'world')).inOrder()

    @metadata(pts_test_id="L2CAP/COS/CFC/BV-02-C", pts_test_name="No Segmentation")
    def test_no_segmentation(self):
        """
        Verify that the IUT can send data segments which do not require segmentation.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert(mtu=1000, mps=202)
        dut_channel.send(b'hello' * 40)
        assertThat(cert_channel).emits(L2capMatchers.FirstLeIFrame(b'hello' * 40, sdu_size=200))

    def test_no_segmentation_dut_is_central(self):
        """
        L2CAP/COS/CFC/BV-02-C
        """
        (dut_channel, cert_channel) = self._set_link_from_dut_and_open_channel()
        dut_channel.send(b'hello' * 40)
        assertThat(cert_channel).emits(L2capMatchers.FirstLeIFrame(b'hello' * 40, sdu_size=200))

    @metadata(pts_test_id="L2CAP/COS/CFC/BV-03-C", pts_test_name="Reassembling")
    def test_reassembling(self):
        """
        Verify that the IUT can correctly reassemble data received from the Lower Tester which is greater than the IUT LE-frame size.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        sdu_size_for_two_sample_packet = 8
        cert_channel.send_first_le_i_frame(sdu_size_for_two_sample_packet, SAMPLE_PACKET)
        cert_channel.send(SAMPLE_PACKET)
        assertThat(dut_channel).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17' * 2))

    @metadata(pts_test_id="L2CAP/COS/CFC/BV-04-C", pts_test_name="Data Receiving")
    def test_data_receiving(self):
        """
        Verify that the IUT can receive unsegmented data correctly.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        cert_channel.send_first_le_i_frame(4, SAMPLE_PACKET)
        assertThat(dut_channel).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'))

    def test_data_receiving_dut_is_central(self):
        """
        L2CAP/COS/CFC/BV-04-C
        """
        (dut_channel, cert_channel) = self._set_link_from_dut_and_open_channel()
        cert_channel.send_first_le_i_frame(4, SAMPLE_PACKET)
        assertThat(dut_channel).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'))

    @metadata(pts_test_id="L2CAP/COS/CFC/BV-05-C", pts_test_name="Multiple Channels with Interleaved Data Streams")
    def test_multiple_channels_with_interleaved_data_streams(self):
        """
        Verify that an IUT can create multiple channels and receives data streams on the channels when the streams are interleaved.
        """
        self._setup_link_from_cert()
        (dut_channel_x, cert_channel_x) = self._open_channel_from_cert(signal_id=1, scid=0x0103, psm=0x33)
        (dut_channel_y, cert_channel_y) = self._open_channel_from_cert(signal_id=2, scid=0x0105, psm=0x35)
        (dut_channel_z, cert_channel_z) = self._open_channel_from_cert(signal_id=3, scid=0x0107, psm=0x37)
        cert_channel_y.send_first_le_i_frame(4, SAMPLE_PACKET)
        cert_channel_z.send_first_le_i_frame(4, SAMPLE_PACKET)
        cert_channel_y.send_first_le_i_frame(4, SAMPLE_PACKET)
        cert_channel_z.send_first_le_i_frame(4, SAMPLE_PACKET)
        cert_channel_y.send_first_le_i_frame(4, SAMPLE_PACKET)
        # TODO: We should assert two events in order, but it got stuck
        assertThat(dut_channel_y).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'), at_least_times=3)
        assertThat(dut_channel_z).emits(
            L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'),
            L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17')).inOrder()
        cert_channel_z.send_first_le_i_frame(4, SAMPLE_PACKET)
        assertThat(dut_channel_z).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'))

    @metadata(pts_test_id="L2CAP/LE/REJ/BI-01-C", pts_test_name="Reject Unknown Command in LE Signaling Channel")
    def test_reject_unknown_command_in_le_sigling_channel(self):
        """
        Verify that the IUT is able to reject unknown command.
        """
        self._setup_link_from_cert()
        self.cert_l2cap.get_control_channel().send(
            l2cap_packets.InformationRequestBuilder(
                2, l2cap_packets.InformationRequestInfoType.EXTENDED_FEATURES_SUPPORTED))
        assertThat(self.cert_l2cap.get_control_channel()).emits(L2capMatchers.LeCommandReject())

    @metadata(pts_test_id="L2CAP/LE/REJ/BI-02-C", pts_test_name="Command Reject – Reserved PDU Codes")
    def test_command_reject_reserved_pdu_codes(self):
        """
        Verify that an IUT receiving a PDU with a reserved command code sends a command reject.
        """
        self._setup_link_from_cert()
        self.cert_l2cap.get_control_channel().send(l2cap_packets.MoveChannelRequestBuilder(2, 0, 0))
        assertThat(self.cert_l2cap.get_control_channel()).emits(L2capMatchers.LeCommandReject())

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-01-C", pts_test_name="LE Credit Based Connection Request - Legacy Peer")
    def test_le_credit_based_connection_request_legacy_peer(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request to a legacy peer and receiving a Command Reject does not establish the channel.
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_reject_open_channel_from_remote(psm=0x33)
        assertThat(response_future.get_status()).isNotEqualTo(LeCreditBasedConnectionResponseResult.SUCCESS)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-02-C", pts_test_name="LE Credit Based Connection Request on Supported LE_PSM")
    def test_le_credit_based_connection_request_on_supported_le_psm(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request to a peer will establish the channel upon receiving the LE Credit Based Connection Response.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_dut()
        cert_channel.send_first_le_i_frame(4, SAMPLE_PACKET)
        assertThat(dut_channel).emits(L2capMatchers.PacketPayloadRawData(b'\x19\x26\x08\x17'))

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-03-C", pts_test_name="LE Credit Based Connection Response on Supported LE_PSM")
    def test_credit_based_connection_response_on_supported_le_psm(self):
        """
        Verify that an IUT receiving a valid LE Credit Based Connection Request from a peer will send an LE Credit Based Connection Response and establish the channel.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        dut_channel.send(b'hello')
        assertThat(cert_channel).emits(L2capMatchers.FirstLeIFrame(b'hello', sdu_size=5))

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-04-C", pts_test_name="LE Credit Based Connection Request on an Unsupported LE_PSM")
    def test_credit_based_connection_request_on_an_unsupported_le_psm(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request on an unsupported LE_PSM will not establish a channel upon receiving an LE Credit Based Connection Response refusing the connection.
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.LE_PSM_NOT_SUPPORTED)
        assertThat(response_future.get_status()).isEqualTo(LeCreditBasedConnectionResponseResult.LE_PSM_NOT_SUPPORTED)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-05-C", pts_test_name="LE Credit Based Connection Request - unsupported LE_PSM")
    def test_credit_based_connection_request_unsupported_le_psm(self):
        """
        Verify that an IUT receiving an LE Credit Based Connection Request on an unsupported LE_PSM will respond with an LE Credit Based Connection Response refusing the connection.
        """
        self._setup_link_from_cert()
        self.cert_l2cap.get_control_channel().send(
            l2cap_packets.LeCreditBasedConnectionRequestBuilder(1, 0x34, 0x0101, 2000, 1000, 1000))
        assertThat(self.cert_l2cap.get_control_channel()).emits(
            L2capMatchers.CreditBasedConnectionResponse(
                result=LeCreditBasedConnectionResponseResult.LE_PSM_NOT_SUPPORTED))

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-06-C", pts_test_name="Credit Exchange – Receiving Incremental Credits")
    def test_credit_exchange_receiving_incremental_credits(self):
        """
        Verify the IUT handles flow control correctly, by handling the LE Flow Control Credit sent by the peer.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert(initial_credit=0)
        for _ in range(4):
            dut_channel.send(b'hello')
        cert_channel.send_credits(1)
        assertThat(cert_channel).emits(L2capMatchers.FirstLeIFrame(b'hello', sdu_size=5))
        cert_channel.send_credits(1)
        assertThat(cert_channel).emits(L2capMatchers.FirstLeIFrame(b'hello', sdu_size=5))
        cert_channel.send_credits(2)
        assertThat(cert_channel).emits(
            L2capMatchers.FirstLeIFrame(b'hello', sdu_size=5), L2capMatchers.FirstLeIFrame(b'hello', sdu_size=5))

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-07-C", pts_test_name="Credit Exchange – Sending Credits")
    def test_credit_exchange_sending_credits(self):
        """
        Verify that the IUT sends LE Flow Control Credit to the peer.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        credits = cert_channel.credits_left()
        # Note: DUT only needs to send credit when ALL credits are consumed.
        # Here we enforce that DUT sends credit after receiving 3 packets, to
        # test without sending too many packets (may take too long).
        # This behavior is not expected for all Bluetooth stacks.
        for _ in range(min(credits + 1, 3)):
            cert_channel.send_first_le_i_frame(4, SAMPLE_PACKET)
        self.cert_l2cap.verify_le_flow_control_credit(cert_channel)

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-08-C", pts_test_name="Disconnection Request")
    def test_disconnection_request(self):
        """
        Verify that the IUT can disconnect the channel.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        dut_channel.close_channel()
        cert_channel.verify_disconnect_request()

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-09-C", pts_test_name="Disconnection Response")
    def test_disconnection_response(self):
        """
        Verify that the IUT responds correctly to reception of a Disconnection Request.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        cert_channel.disconnect_and_verify()

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-10-C", pts_test_name="Security - Insufficient Authentication – Initiator")
    def test_security_insufficient_authentication_initiator(self):
        """
        Verify that the IUT does not establish the channel upon receipt of an LE Credit Based Connection Response indicating the connection was refused with Result “0x0005 – Connection Refused – Insufficient Authentication".
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHENTICATION)
        assertThat(response_future.get_status()).isEqualTo(
            LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHENTICATION)

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-11-C", pts_test_name="Security - Insufficient Authentication – Responder")
    def test_security_insufficient_authentication_responder(self):
        """
        Verify that an IUT refuses to create a connection upon reception of an LE Credit Based Connection
Request which fails to satisfy authentication requirements.
        """
        self._setup_link_from_cert()
        psm = 0x33
        self.dut_l2cap.register_coc(self.cert_address, psm, SecurityLevel.AUTHENTICATED_PAIRING_WITH_ENCRYPTION)
        self.cert_l2cap.open_channel_with_expected_result(
            psm, LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHENTICATION)

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-12-C", pts_test_name="Security - Insufficient Authorization – Initiator")
    def test_security_insufficient_authorization_initiator(self):
        """
        Verify that the IUT does not establish the channel upon receipt of an LE Credit Based Connection Response indicating the connection was refused with Result “0x0006 – Connection Refused – Insufficient Authorization”.
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHORIZATION)
        assertThat(response_future.get_status()).isEqualTo(
            LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHORIZATION)

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-13-C", pts_test_name="Security - Insufficient Authorization – Responder")
    def test_security_insufficient_authorization_responder(self):
        """
        Verify that an IUT refuses to create a connection upon reception of an LE Credit Based Connection
        Request which fails to satisfy authentication requirements.
        """
        self._setup_link_from_cert()
        psm = 0x33
        self.dut_l2cap.register_coc(self.cert_address, psm, SecurityLevel.AUTHORIZATION)
        self.cert_l2cap.open_channel_with_expected_result(
            psm, LeCreditBasedConnectionResponseResult.INSUFFICIENT_AUTHORIZATION)

    @metadata(pts_test_id="L2CAP/LE/CFC/BV-14-C", pts_test_name="Security - Insufficient Key Size – Initiator")
    def test_security_insufficient_key_size_initiator(self):
        """
        Verify that the IUT does not establish the channel upon receipt of an
        LE Credit Based Connection Response indicating the connection was
        refused with Result "0x0007 – Connection Refused – Insufficient
        Encryption Key Size".
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.INSUFFICIENT_ENCRYPTION_KEY_SIZE)
        assertThat(response_future.get_status()).isEqualTo(
            LeCreditBasedConnectionResponseResult.INSUFFICIENT_ENCRYPTION_KEY_SIZE)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-15-C", pts_test_name="Security - Insufficient Encryption Key Size – Responder")
    def test_security_insufficient_encryption_key_size_responder(self):
        """
        Verify that an IUT refuses to create a connection upon receipt of an LE Credit Based Connection
        Request which fails to satisfy Encryption Key Size requirements.
        """
        self._setup_link_from_cert()
        psm = 0x33
        self.dut_l2cap.register_coc(self.cert_address, psm, SecurityLevel.AUTHENTICATED_PAIRING_WITH_128_BIT_KEY)
        self.cert_l2cap.open_channel_with_expected_result(
            psm, LeCreditBasedConnectionResponseResult.INSUFFICIENT_ENCRYPTION_KEY_SIZE)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-16-C",
        pts_test_name="LE Credit Based Connection Request - refuse due to insufficient resources - Initiator")
    def test_le_connection_request_insufficient_resources_initiator(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request does
        not establish the channel upon receiving an LE Credit Based Connection
        Response refusing the connection with result "0x0004 – Connection
        refused – no resources available".
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.NO_RESOURCES_AVAILABLE)
        assertThat(response_future.get_status()).isEqualTo(LeCreditBasedConnectionResponseResult.NO_RESOURCES_AVAILABLE)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-18-C",
        pts_test_name="LE Credit Based Connection Request - refused due to Invalid Source CID - Initiator")
    def test_request_refused_due_to_invalid_source_cid_initiator(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x0009 – Connection refused – Invalid Source CID".
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.INVALID_SOURCE_CID)
        assertThat(response_future.get_status()).isEqualTo(LeCreditBasedConnectionResponseResult.INVALID_SOURCE_CID)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-19-C",
        pts_test_name="LE Credit Based Connection Request - refused due to source CID already allocated - Initiator")
    def test_request_refused_due_to_source_cid_already_allocated_initiator(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x000A – Connection refused – Source CID already allocated".
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.SOURCE_CID_ALREADY_ALLOCATED)
        assertThat(response_future.get_status()).isEqualTo(
            LeCreditBasedConnectionResponseResult.SOURCE_CID_ALREADY_ALLOCATED)

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-20-C",
        pts_test_name="LE Credit Based Connection Response - refused due to Source CID already allocated - Responder")
    def test_request_refused_due_to_source_cid_already_allocated_responder(self):
        """
        Verify that an IUT receiving an LE Credit Based Connection Request for a second channel will refuse the connection with result "0x000A - Connection refused – Source CID already allocated" if it receives a Source CID which is already in use.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert(psm=0x33, scid=0x0101)
        self.dut_l2cap.register_coc(self.cert_address, psm=0x35)
        self.cert_l2cap.get_control_channel().send(
            l2cap_packets.LeCreditBasedConnectionRequestBuilder(2, 0x35, 0x0101, 1000, 1000, 1000))
        assertThat(self.cert_l2cap.get_control_channel()).emits(L2capMatchers.CreditBasedConnectionResponseUsedCid())

    @metadata(
        pts_test_id="L2CAP/LE/CFC/BV-21-C",
        pts_test_name="LE Credit Based Connection Request - refused due to Unacceptable Parameters - Initiator")
    def test_request_refused_due_to_unacceptable_parameters_initiator(self):
        """
        Verify that an IUT sending an LE Credit Based Connection Request does not establish the channel upon receiving an LE Credit Based Connection Response refusing the connection with result "0x000B – Connection refused – Unacceptable Parameters".
        """
        self._setup_link_from_cert()
        response_future = self.dut_l2cap.connect_coc_to_cert(self.cert_address, psm=0x33)
        self.cert_l2cap.verify_and_respond_open_channel_from_remote(
            psm=0x33, result=LeCreditBasedConnectionResponseResult.UNACCEPTABLE_PARAMETERS)
        assertThat(response_future.get_status()).isEqualTo(
            LeCreditBasedConnectionResponseResult.UNACCEPTABLE_PARAMETERS)

    @metadata(pts_test_id="L2CAP/LE/CFC/BI-01-C", pts_test_name="Credit Exchange – Exceed Initial Credits")
    def test_credit_exchange_exceed_initial_credits(self):
        """
        Verify that the IUT disconnects the LE Data Channel when the credit count exceeds 65535.
        """
        self._setup_link_from_cert()
        (dut_channel, cert_channel) = self._open_channel_from_cert()
        cert_channel.send_credits(65535)
        cert_channel.verify_disconnect_request()