# Copyright 2025 Google LLC
#
# 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
#
#     https://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.

import hci_packets as hci
import link_layer_packets as ll
import llcp_packets as llcp
import random
import unittest
from hci_packets import ErrorCode
from py.bluetooth import Address
from py.controller import ControllerTest, generate_rpa


class Test(ControllerTest):

    SDU_Interval_C_TO_P = 10000  # 10ms
    SDU_Interval_P_TO_C = 10000  # 10ms
    ISO_Interval = 16  # 20ms
    Sub_Interval = 7500  # 7.5ms (approximation)
    CIG_Sync_Delay = 7500  # 7.5ms (approximation)
    CIS_Sync_Delay = 7500  # 7.5ms (approximation)
    Worst_Case_SCA = hci.ClockAccuracy.PPM_500
    Packing = hci.Packing.SEQUENTIAL
    Framing = hci.Enable.DISABLED
    NSE = 4
    Max_SDU_C_TO_P = 130
    Max_SDU_P_TO_C = 130
    Max_PDU_C_TO_P = 130
    Max_PDU_P_TO_C = 130
    PHY_C_TO_P = 0x1
    PHY_P_TO_C = 0x1
    FT_C_TO_P = 1
    FT_P_TO_C = 1
    BN_C_TO_P = 2
    BN_P_TO_C = 2
    Max_Transport_Latency_C_TO_P = 40000  # 40ms
    Max_Transport_Latency_P_TO_C = 40000  # 40ms
    RTN_C_TO_P = 3
    RTN_P_TO_C = 3

    # Test the behavior of the controller when disconnecting the
    # ACL connection for a peripheral CIS that is in accepting state.
    async def test_peripheral_accepting(self):
        # Test parameters.
        cig_id = 0x12
        cis_id = 0x42
        cis_connection_handle = 0xe00
        peer_address = Address('aa:bb:cc:dd:ee:ff')
        controller = self.controller

        # Enable Connected Isochronous Stream Host Support.
        await self.enable_connected_isochronous_stream_host_support()

        # Prelude: Establish an ACL connection as peripheral with the IUT.
        acl_connection_handle = await self.establish_le_connection_peripheral(peer_address)

        # The Lower Tester sends an LL_CIS_REQ to the IUT.
        controller.send_llcp(source_address=peer_address,
                             destination_address=controller.address,
                             pdu=llcp.CisReq(cig_id=cig_id,
                                             cis_id=cis_id,
                                             phy_c_to_p=hci.PhyType.LE_1M,
                                             phy_p_to_c=hci.PhyType.LE_1M,
                                             framed=self.Framing == hci.Enable.ENABLED,
                                             max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                             max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                             sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                             sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                             max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                             max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                             nse=self.NSE,
                                             sub_interval=self.Sub_Interval,
                                             bn_p_to_c=self.BN_C_TO_P,
                                             bn_c_to_p=self.BN_P_TO_C,
                                             ft_c_to_p=self.FT_C_TO_P,
                                             ft_p_to_c=self.FT_P_TO_C,
                                             iso_interval=self.ISO_Interval,
                                             cis_offset_min=0,
                                             cis_offset_max=0,
                                             conn_event_count=0))

        # The IUT sends an HCI_LE_CIS_Request event to the Upper Tester and the parameters include
        # CIS_Connection_Handle assigned by the IUT.
        await self.expect_evt(
            hci.LeCisRequest(acl_connection_handle=acl_connection_handle,
                             cis_connection_handle=cis_connection_handle,
                             cig_id=cig_id,
                             cis_id=cis_id))

        # The ACL connection is disconnected before the Host can accept the CIS request.
        controller.send_ll(
            ll.Disconnect(source_address=peer_address,
                          destination_address=controller.address,
                          reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller sends HCI Disconnection Complete event for the ACL
        # connection, but the pending CIS connection is silently dropped.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=acl_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

    # Test the behavior of the controller when disconnecting the
    # ACL connection for a peripheral CIS that is in accepted state, pending the
    # CIS connection indication.
    async def test_peripheral_accepted(self):
        # Test parameters.
        cig_id = 0x12
        cis_id = 0x42
        cis_connection_handle = 0xe00
        peer_address = Address('aa:bb:cc:dd:ee:ff')
        controller = self.controller

        # Enable Connected Isochronous Stream Host Support.
        await self.enable_connected_isochronous_stream_host_support()

        # Prelude: Establish an ACL connection as peripheral with the IUT.
        acl_connection_handle = await self.establish_le_connection_peripheral(peer_address)

        # The Lower Tester sends an LL_CIS_REQ to the IUT.
        controller.send_llcp(source_address=peer_address,
                             destination_address=controller.address,
                             pdu=llcp.CisReq(cig_id=cig_id,
                                             cis_id=cis_id,
                                             phy_c_to_p=hci.PhyType.LE_1M,
                                             phy_p_to_c=hci.PhyType.LE_1M,
                                             framed=self.Framing == hci.Enable.ENABLED,
                                             max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                             max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                             sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                             sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                             max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                             max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                             nse=self.NSE,
                                             sub_interval=self.Sub_Interval,
                                             bn_p_to_c=self.BN_C_TO_P,
                                             bn_c_to_p=self.BN_P_TO_C,
                                             ft_c_to_p=self.FT_C_TO_P,
                                             ft_p_to_c=self.FT_P_TO_C,
                                             iso_interval=self.ISO_Interval,
                                             cis_offset_min=0,
                                             cis_offset_max=0,
                                             conn_event_count=0))

        # The IUT sends an HCI_LE_CIS_Request event to the Upper Tester and the parameters include
        # CIS_Connection_Handle assigned by the IUT.
        await self.expect_evt(
            hci.LeCisRequest(acl_connection_handle=acl_connection_handle,
                             cis_connection_handle=cis_connection_handle,
                             cig_id=cig_id,
                             cis_id=cis_id))

        # The Upper Tester accepts the CIS request.
        controller.send_cmd(hci.LeAcceptCisRequest(connection_handle=cis_connection_handle))

        # The IUT sends a successful Command Status to the Upper Tester.
        await self.expect_evt(
            hci.LeAcceptCisRequestStatus(status=ErrorCode.SUCCESS, num_hci_command_packets=1))

        # The IUT sends an LL_CIS_RSP PDU to the Upper Tester. In the message, the CIS_Offset_Min
        # field and the CIS_Offset_Max field are equal to or a subset of the values received in the
        # LL_CIS_REQ sent in step 2.
        cis_rsp = await self.expect_llcp(source_address=controller.address,
                                         destination_address=peer_address,
                                         expected_pdu=llcp.CisRsp(cis_offset_min=self.Any,
                                                                  cis_offset_max=self.Any,
                                                                  conn_event_count=0))

        # The ACL connection is disconnected before the Lower Tester sends back LL_CIS_IND.
        controller.send_ll(
            ll.Disconnect(source_address=peer_address,
                          destination_address=controller.address,
                          reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller send HCI LE Cis Established for to notify of the status
        # of the CIS connection.
        await self.expect_evt(
            hci.LeCisEstablishedV1(status=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION,
                                   connection_handle=self.Any))

        # The controller sends HCI Disconnection Complete event for the ACL
        # connection, but the pending CIS connection is silently dropped.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=acl_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

    # Test the behavior of the controller when disconnecting the
    # ACL connection for a peripheral CIS that is in connected state.
    async def test_peripheral_connected(self):
        # Test parameters.
        cig_id = 0x12
        cis_id = 0x42
        cis_connection_handle = 0xe00
        peer_address = Address('aa:bb:cc:dd:ee:ff')
        controller = self.controller

        # Enable Connected Isochronous Stream Host Support.
        await self.enable_connected_isochronous_stream_host_support()

        # Prelude: Establish an ACL connection as peripheral with the IUT.
        acl_connection_handle = await self.establish_le_connection_peripheral(peer_address)

        # The Lower Tester sends an LL_CIS_REQ to the IUT.
        controller.send_llcp(source_address=peer_address,
                             destination_address=controller.address,
                             pdu=llcp.CisReq(cig_id=cig_id,
                                             cis_id=cis_id,
                                             phy_c_to_p=hci.PhyType.LE_1M,
                                             phy_p_to_c=hci.PhyType.LE_1M,
                                             framed=self.Framing == hci.Enable.ENABLED,
                                             max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                             max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                             sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                             sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                             max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                             max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                             nse=self.NSE,
                                             sub_interval=self.Sub_Interval,
                                             bn_p_to_c=self.BN_C_TO_P,
                                             bn_c_to_p=self.BN_P_TO_C,
                                             ft_c_to_p=self.FT_C_TO_P,
                                             ft_p_to_c=self.FT_P_TO_C,
                                             iso_interval=self.ISO_Interval,
                                             cis_offset_min=0,
                                             cis_offset_max=0,
                                             conn_event_count=0))

        # The IUT sends an HCI_LE_CIS_Request event to the Upper Tester and the parameters include
        # CIS_Connection_Handle assigned by the IUT.
        await self.expect_evt(
            hci.LeCisRequest(acl_connection_handle=acl_connection_handle,
                             cis_connection_handle=cis_connection_handle,
                             cig_id=cig_id,
                             cis_id=cis_id))

        # The Upper Tester accepts the CIS request.
        controller.send_cmd(hci.LeAcceptCisRequest(connection_handle=cis_connection_handle))

        # The IUT sends a successful Command Status to the Upper Tester.
        await self.expect_evt(
            hci.LeAcceptCisRequestStatus(status=ErrorCode.SUCCESS, num_hci_command_packets=1))

        # The IUT sends an LL_CIS_RSP PDU to the Upper Tester. In the message, the CIS_Offset_Min
        # field and the CIS_Offset_Max field are equal to or a subset of the values received in the
        # LL_CIS_REQ sent in step 2.
        cis_rsp = await self.expect_llcp(source_address=controller.address,
                                         destination_address=peer_address,
                                         expected_pdu=llcp.CisRsp(cis_offset_min=self.Any,
                                                                  cis_offset_max=self.Any,
                                                                  conn_event_count=0))

        # The Lower Tester sends an LL_CIS_IND where the CIS_Offset is the time (ms) from the start of
        # the ACL connection event in connEvent Count to the first CIS anchor point, the CIS_Sync_Delay
        # is CIG_Sync_Delay minus the offset from the CIG reference point to the CIS anchor point in s,
        # and the connEventCount is the CIS_Offset reference point.
        controller.send_llcp(source_address=peer_address,
                             destination_address=controller.address,
                             pdu=llcp.CisInd(aa=0,
                                             cis_offset=cis_rsp.cis_offset_max,
                                             cig_sync_delay=self.CIG_Sync_Delay,
                                             cis_sync_delay=self.CIS_Sync_Delay,
                                             conn_event_count=0))

        # The IUT sends a successful HCI_LE_CIS_Established event to the Upper Tester, after the first
        # CIS packet sent by the Lower Tester. The Connection_Handle parameter is the
        # CIS_Connection_Handle value provided in the HCI_LE_CIS_Request event.
        await self.expect_evt(
            hci.LeCisEstablishedV1(status=ErrorCode.SUCCESS,
                                   connection_handle=cis_connection_handle,
                                   cig_sync_delay=self.CIG_Sync_Delay,
                                   cis_sync_delay=self.CIS_Sync_Delay,
                                   transport_latency_c_to_p=self.Any,
                                   transport_latency_p_to_c=self.Any,
                                   phy_c_to_p=hci.SecondaryPhyType.LE_1M,
                                   phy_p_to_c=hci.SecondaryPhyType.LE_1M,
                                   nse=self.NSE,
                                   bn_c_to_p=self.BN_C_TO_P,
                                   bn_p_to_c=self.BN_P_TO_C,
                                   ft_c_to_p=self.FT_C_TO_P,
                                   ft_p_to_c=self.FT_P_TO_C,
                                   max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                   max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                   iso_interval=self.ISO_Interval))

        # The ACL connection is disconnected when the CIS connection is fully established.
        controller.send_ll(
            ll.Disconnect(source_address=peer_address,
                          destination_address=controller.address,
                          reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller send HCI LE Cis Established for to notify of the status
        # of the CIS connection.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=cis_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller sends HCI Disconnection Complete event for the ACL
        # connection, but the pending CIS connection is silently dropped.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=acl_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

    # Test the behavior of the controller when disconnecting the
    # ACL connection for a peripheral CIS that is in accepted state, pending the
    # CIS connection indication.
    async def test_central_initiating(self):
        # Test parameters.
        cig_id = 0x12
        cis_id = 0x42
        cis_connection_handle = 0xe00
        peer_address = Address('aa:bb:cc:dd:ee:ff')
        controller = self.controller

        # Enable Connected Isochronous Stream Host Support.
        await self.enable_connected_isochronous_stream_host_support()

        # Prelude: Establish an ACL connection as central with the IUT.
        acl_connection_handle = await self.establish_le_connection_central(peer_address)

        # The Upper Tester sends an HCI_LE_Set_CIG_Parameters_Test command to the IUT with
        # CIS_Count set to 1, BN, FT, NSE, PHY_C_TO_P[], PHY_P_TO_C[] and ISO_Interval to be set to
        # the values specified in Table 4.135 and Table 4.136. Any remaining values are assigned the
        # default values as specified in Section 4.10.1.3 Default Values for Set CIG Parameters
        # Commands. The Upper Tester receives a successful HCI_Command_Complete event with a
        # valid Connection_Handle from the IUT and CIS_Count = 1.
        controller.send_cmd(
            hci.LeSetCigParametersTest(cig_id=cig_id,
                                       sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                       sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                       ft_c_to_p=self.FT_C_TO_P,
                                       ft_p_to_c=self.FT_P_TO_C,
                                       iso_interval=self.ISO_Interval,
                                       worst_case_sca=self.Worst_Case_SCA,
                                       packing=self.Packing,
                                       framing=self.Framing,
                                       cis_config=[
                                           hci.LeCisParametersTestConfig(
                                               cis_id=cis_id,
                                               nse=self.NSE,
                                               max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                               max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                               max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                               max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                               phy_c_to_p=self.PHY_C_TO_P,
                                               phy_p_to_c=self.PHY_P_TO_C,
                                               bn_c_to_p=self.BN_C_TO_P,
                                               bn_p_to_c=self.BN_P_TO_C)
                                       ]))

        await self.expect_evt(
            hci.LeSetCigParametersTestComplete(status=ErrorCode.SUCCESS,
                                               num_hci_command_packets=1,
                                               cig_id=cig_id,
                                               connection_handle=[cis_connection_handle]))

        # The Upper Tester sends an HCI_LE_Create_CIS command to the IUT with the
        # ACL_Connection_Handle of the established ACL connection and CIS_Count set to 1. The Upper
        # Tester receives a Status of Success from the IUT.
        controller.send_cmd(
            hci.LeCreateCis(cis_config=[
                hci.LeCreateCisConfig(cis_connection_handle=cis_connection_handle,
                                      acl_connection_handle=acl_connection_handle)
            ]))

        await self.expect_evt(
            hci.LeCreateCisStatus(status=ErrorCode.SUCCESS, num_hci_command_packets=1))

        # The Lower Tester receives an LL_CIS_REQ PDU from the IUT with all fields set to valid values.
        # CIS_Offset_Min is a value between 500µs and TSPX_conn_interval, CIS_Offset_Max is a value
        # between CIS_Offset_Min and the CIS_Offset_Max value as calculated in [14] Section 2.4.2.29
        # using TSPX_conn_interval as the value of connInterval, and connEventCount is the reference
        # event anchor point for which the offsets applied.
        cis_req = await self.expect_llcp(source_address=controller.address,
                                         destination_address=peer_address,
                                         expected_pdu=llcp.CisReq(
                                             cig_id=cig_id,
                                             cis_id=cis_id,
                                             phy_c_to_p=hci.PhyType.LE_1M,
                                             phy_p_to_c=hci.PhyType.LE_1M,
                                             framed=self.Framing == hci.Enable.ENABLED,
                                             max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                             max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                             sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                             sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                             max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                             max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                             nse=self.NSE,
                                             sub_interval=self.Any,
                                             bn_p_to_c=self.BN_C_TO_P,
                                             bn_c_to_p=self.BN_P_TO_C,
                                             ft_c_to_p=self.FT_C_TO_P,
                                             ft_p_to_c=self.FT_P_TO_C,
                                             iso_interval=self.ISO_Interval,
                                             cis_offset_min=self.Any,
                                             cis_offset_max=self.Any,
                                             conn_event_count=0))

        # The ACL connection is disconnected before the Lower Tester sends back LL_CIS_RSP.
        controller.send_ll(
            ll.Disconnect(source_address=peer_address,
                          destination_address=controller.address,
                          reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller send HCI LE Cis Established for to notify of the status
        # of the CIS connection.
        await self.expect_evt(
            hci.LeCisEstablishedV1(status=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION,
                                   connection_handle=cis_connection_handle))

        # The controller sends HCI Disconnection Complete event for the ACL
        # connection.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=acl_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

    # Test the behavior of the controller when disconnecting the
    # ACL connection for a central CIS that is in connected state.
    async def test_central_connected(self):
        # Test parameters.
        cig_id = 0x12
        cis_id = 0x42
        cis_connection_handle = 0xe00
        peer_address = Address('aa:bb:cc:dd:ee:ff')
        controller = self.controller

        # Enable Connected Isochronous Stream Host Support.
        await self.enable_connected_isochronous_stream_host_support()

        # Prelude: Establish an ACL connection as central with the IUT.
        acl_connection_handle = await self.establish_le_connection_central(peer_address)

        # The Upper Tester sends an HCI_LE_Set_CIG_Parameters_Test command to the IUT with
        # CIS_Count set to 1, BN, FT, NSE, PHY_C_TO_P[], PHY_P_TO_C[] and ISO_Interval to be set to
        # the values specified in Table 4.135 and Table 4.136. Any remaining values are assigned the
        # default values as specified in Section 4.10.1.3 Default Values for Set CIG Parameters
        # Commands. The Upper Tester receives a successful HCI_Command_Complete event with a
        # valid Connection_Handle from the IUT and CIS_Count = 1.
        controller.send_cmd(
            hci.LeSetCigParametersTest(cig_id=cig_id,
                                       sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                       sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                       ft_c_to_p=self.FT_C_TO_P,
                                       ft_p_to_c=self.FT_P_TO_C,
                                       iso_interval=self.ISO_Interval,
                                       worst_case_sca=self.Worst_Case_SCA,
                                       packing=self.Packing,
                                       framing=self.Framing,
                                       cis_config=[
                                           hci.LeCisParametersTestConfig(
                                               cis_id=cis_id,
                                               nse=self.NSE,
                                               max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                               max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                               max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                               max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                               phy_c_to_p=self.PHY_C_TO_P,
                                               phy_p_to_c=self.PHY_P_TO_C,
                                               bn_c_to_p=self.BN_C_TO_P,
                                               bn_p_to_c=self.BN_P_TO_C)
                                       ]))

        await self.expect_evt(
            hci.LeSetCigParametersTestComplete(status=ErrorCode.SUCCESS,
                                               num_hci_command_packets=1,
                                               cig_id=cig_id,
                                               connection_handle=[cis_connection_handle]))

        # The Upper Tester sends an HCI_LE_Create_CIS command to the IUT with the
        # ACL_Connection_Handle of the established ACL connection and CIS_Count set to 1. The Upper
        # Tester receives a Status of Success from the IUT.
        controller.send_cmd(
            hci.LeCreateCis(cis_config=[
                hci.LeCreateCisConfig(cis_connection_handle=cis_connection_handle,
                                      acl_connection_handle=acl_connection_handle)
            ]))

        await self.expect_evt(
            hci.LeCreateCisStatus(status=ErrorCode.SUCCESS, num_hci_command_packets=1))

        # The Lower Tester receives an LL_CIS_REQ PDU from the IUT with all fields set to valid values.
        # CIS_Offset_Min is a value between 500µs and TSPX_conn_interval, CIS_Offset_Max is a value
        # between CIS_Offset_Min and the CIS_Offset_Max value as calculated in [14] Section 2.4.2.29
        # using TSPX_conn_interval as the value of connInterval, and connEventCount is the reference
        # event anchor point for which the offsets applied.
        cis_req = await self.expect_llcp(source_address=controller.address,
                                         destination_address=peer_address,
                                         expected_pdu=llcp.CisReq(
                                             cig_id=cig_id,
                                             cis_id=cis_id,
                                             phy_c_to_p=hci.PhyType.LE_1M,
                                             phy_p_to_c=hci.PhyType.LE_1M,
                                             framed=self.Framing == hci.Enable.ENABLED,
                                             max_sdu_c_to_p=self.Max_SDU_C_TO_P,
                                             max_sdu_p_to_c=self.Max_SDU_P_TO_C,
                                             sdu_interval_c_to_p=self.SDU_Interval_C_TO_P,
                                             sdu_interval_p_to_c=self.SDU_Interval_P_TO_C,
                                             max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                             max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                             nse=self.NSE,
                                             sub_interval=self.Any,
                                             bn_p_to_c=self.BN_C_TO_P,
                                             bn_c_to_p=self.BN_P_TO_C,
                                             ft_c_to_p=self.FT_C_TO_P,
                                             ft_p_to_c=self.FT_P_TO_C,
                                             iso_interval=self.ISO_Interval,
                                             cis_offset_min=self.Any,
                                             cis_offset_max=self.Any,
                                             conn_event_count=0))

        # The Lower Tester sends an LL_CIS_RSP PDU to the IUT.
        controller.send_llcp(source_address=peer_address,
                             destination_address=controller.address,
                             pdu=llcp.CisRsp(cis_offset_min=cis_req.cis_offset_min,
                                             cis_offset_max=cis_req.cis_offset_max,
                                             conn_event_count=0))

        # The Lower Tester receives an LL_CIS_IND from the IUT where the CIS_Offset is the time (ms)
        # from the start of the ACL connection event in connEvent Count to the first CIS anchor point, the
        # CIS_Sync_Delay is CIG_Sync_Delay minus the offset from the CIG reference point to the CIS
        # anchor point in s, and the connEventCount is the CIS_Offset reference point.
        cis_ind = await self.expect_llcp(source_address=controller.address,
                                         destination_address=peer_address,
                                         expected_pdu=llcp.CisInd(aa=0,
                                                                  cis_offset=self.Any,
                                                                  cig_sync_delay=self.Any,
                                                                  cis_sync_delay=self.Any,
                                                                  conn_event_count=0))

        # 7. The Upper Tester receives a successful HCI_LE_CIS_Established event with the NSE, BN, FT,
        # and Max_PDU parameters as set in step 1 from the IUT, after the first CIS packet sent by the LT.
        # The Connection_Handle parameter is set to the value provided in the HCI_LE_Create_CIS
        # command.
        await self.expect_evt(
            hci.LeCisEstablishedV1(status=ErrorCode.SUCCESS,
                                   connection_handle=cis_connection_handle,
                                   cig_sync_delay=cis_ind.cig_sync_delay,
                                   cis_sync_delay=cis_ind.cis_sync_delay,
                                   transport_latency_c_to_p=self.Any,
                                   transport_latency_p_to_c=self.Any,
                                   phy_c_to_p=hci.SecondaryPhyType.LE_1M,
                                   phy_p_to_c=hci.SecondaryPhyType.LE_1M,
                                   nse=self.NSE,
                                   bn_c_to_p=self.BN_C_TO_P,
                                   bn_p_to_c=self.BN_P_TO_C,
                                   ft_c_to_p=self.FT_C_TO_P,
                                   ft_p_to_c=self.FT_P_TO_C,
                                   max_pdu_c_to_p=self.Max_PDU_C_TO_P,
                                   max_pdu_p_to_c=self.Max_PDU_P_TO_C,
                                   iso_interval=self.ISO_Interval))

        # The ACL connection is disconnected after the CIS is established.
        controller.send_ll(
            ll.Disconnect(source_address=peer_address,
                          destination_address=controller.address,
                          reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller send HCI Disconnection Complete event to notify of the status
        # of the CIS connection.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=cis_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))

        # The controller sends HCI Disconnection Complete event for the ACL
        # connection.
        await self.expect_evt(
            hci.DisconnectionComplete(status=hci.ErrorCode.SUCCESS,
                                      connection_handle=acl_connection_handle,
                                      reason=hci.ErrorCode.REMOTE_USER_TERMINATED_CONNECTION))