Quelle  BV_10_C.py

# Copyright 2023 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


class Test(ControllerTest):

    SDU_Interval_C_TO_P = 10000  # 10ms
    SDU_Interval_P_TO_C = 10000  # 10ms
    ISO_Interval = 16  # 20ms
    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

    # LL/CIS/CEN/BV-01-C [CIS Setup Procedure, Central Initiated]
    async def test(self):
        # Test parameters.
        cig_id = 0x12
        cis_id_1 = 0x42
        cis_id_2 = 0x43
        cis_connection_handle_1 = 0xe00
        cis_connection_handle_2 = 0xe01
        peer_address_1 = Address('aa:bb:cc:dd:ee:ff')
        peer_address_2 = Address('aa:bb:cc:dd:ee:fe')
        controller = self.controller

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

        # Prelude: Establish ACL connections as central with the IUT.
        acl_connection_handle_1 = await self.establish_le_connection_central(peer_address_1)
        acl_connection_handle_2 = await self.establish_le_connection_central(peer_address_2)

        # Prelude: Establish CIS(1) and CIS(2) with connected peers.
        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_1,
                                                  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),
                    hci.LeCisParametersTestConfig(cis_id=cis_id_2,
                                                  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_1, cis_connection_handle_2]))

        controller.send_cmd(
            hci.LeCreateCis(cis_config=[
                hci.LeCreateCisConfig(cis_connection_handle=cis_connection_handle_1,
                                      acl_connection_handle=acl_connection_handle_1),
                hci.LeCreateCisConfig(cis_connection_handle=cis_connection_handle_2,
                                      acl_connection_handle=acl_connection_handle_2)
            ]))

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

        cis_req_1 = await self.expect_llcp(source_address=controller.address,
                                           destination_address=peer_address_1,
                                           expected_pdu=llcp.CisReq(
                                               cig_id=cig_id,
                                               cis_id=cis_id_1,
                                               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))

        controller.send_llcp(source_address=peer_address_1,
                             destination_address=controller.address,
                             pdu=llcp.CisRsp(cis_offset_min=cis_req_1.cis_offset_min,
                                             cis_offset_max=cis_req_1.cis_offset_max,
                                             conn_event_count=0))

        cis_ind_1 = await self.expect_llcp(source_address=controller.address,
                                           destination_address=peer_address_1,
                                           expected_pdu=llcp.CisInd(aa=0,
                                                                    cis_offset=self.Any,
                                                                    cig_sync_delay=self.Any,
                                                                    cis_sync_delay=self.Any,
                                                                    conn_event_count=0))

        await self.expect_evt(
            hci.LeCisEstablishedV1(status=ErrorCode.SUCCESS,
                                   connection_handle=cis_connection_handle_1,
                                   cig_sync_delay=cis_ind_1.cig_sync_delay,
                                   cis_sync_delay=cis_ind_1.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))

        cis_req_2 = await self.expect_llcp(source_address=controller.address,
                                           destination_address=peer_address_2,
                                           expected_pdu=llcp.CisReq(
                                               cig_id=cig_id,
                                               cis_id=cis_id_2,
                                               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))

        controller.send_llcp(source_address=peer_address_2,
                             destination_address=controller.address,
                             pdu=llcp.CisRsp(cis_offset_min=cis_req_2.cis_offset_min,
                                             cis_offset_max=cis_req_2.cis_offset_max,
                                             conn_event_count=0))

        cis_ind_2 = await self.expect_llcp(source_address=controller.address,
                                           destination_address=peer_address_2,
                                           expected_pdu=llcp.CisInd(aa=0,
                                                                    cis_offset=self.Any,
                                                                    cig_sync_delay=self.Any,
                                                                    cis_sync_delay=self.Any,
                                                                    conn_event_count=0))

        await self.expect_evt(
            hci.LeCisEstablishedV1(status=ErrorCode.SUCCESS,
                                   connection_handle=cis_connection_handle_2,
                                   cig_sync_delay=cis_ind_2.cig_sync_delay,
                                   cis_sync_delay=cis_ind_2.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))

        # 1. The Upper Tester orders the IUT to send a payload of the specified length to the Lower Testers.
        iso_sdu = [random.randint(1, 251) for n in range(self.Max_SDU_C_TO_P)]

        # 2. Lower Tester 1 receives the payload PDU in the first subevent on CIS(1).
        # 3. Lower Tester 1 sends an Ack T_IFS after receiving the payload PDU.
        controller.send_iso(
            hci.IsoWithoutTimestamp(
                connection_handle=cis_connection_handle_1,
                pb_flag=hci.IsoPacketBoundaryFlag.COMPLETE_SDU,
                packet_sequence_number=42,
                payload=iso_sdu,
            ))

        await self.expect_ll(
            ll.LeConnectedIsochronousPdu(source_address=controller.address,
                                         destination_address=peer_address_1,
                                         cig_id=cig_id,
                                         cis_id=cis_id_1,
                                         sequence_number=42,
                                         data=iso_sdu))

        await self.expect_evt(
            hci.NumberOfCompletedPackets(completed_packets=[
                hci.CompletedPackets(connection_handle=cis_connection_handle_1,
                                     host_num_of_completed_packets=1)
            ]))

        # 4. Lower Tester 2 receives the payload PDU in the first subevent on CIS(2).
        # 5. Lower Tester 2 sends an Ack T_IFS after receiving the payload PDU.
        controller.send_iso(
            hci.IsoWithoutTimestamp(
                connection_handle=cis_connection_handle_2,
                pb_flag=hci.IsoPacketBoundaryFlag.COMPLETE_SDU,
                packet_sequence_number=42,
                payload=iso_sdu,
            ))

        await self.expect_ll(
            ll.LeConnectedIsochronousPdu(source_address=controller.address,
                                         destination_address=peer_address_2,
                                         cig_id=cig_id,
                                         cis_id=cis_id_2,
                                         sequence_number=42,
                                         data=iso_sdu))

        await self.expect_evt(
            hci.NumberOfCompletedPackets(completed_packets=[
                hci.CompletedPackets(connection_handle=cis_connection_handle_2,
                                     host_num_of_completed_packets=1)
            ]))

        # 6. If Table 4.139 specifies a BN of 2 or 3, when CIS(1) subevent interval ends, repeat steps 1–3 in
        # the next subevent.
        # 7. If Table 4.139 specifies a BN of 2 or 3, when CIS(2) subevent interval ends, repeat steps 4 and 5
        # in the next subevent.
        # 8. If Table 4.139 specifies a BN of 3, when CIS(1) subevent interval ends, repeat steps 1–3 in the
        # next subevent.
        # 9. If Table 4.139 specifies a BN of 3, when CIS(2) subevent interval ends, repeat steps 4 and 5 in
        # the next subevent.
        # 10. The time between the ACL connection event for the first received CIS and the start of the first
        # subevent on the same CIS is the observed CIS_Offset(a). The Lower Tester validates that
        # CIS_Offset(a) = CIS_Offset in the LL_CIS_IND associated with CIS sent during setup of the
        # CISes.
        # 11. The time between the ACL connection event for the second received CIS and the start of the first
        # subevent on the same CIS is the observed CIS_Offset(b). The Lower Tester validates that the
        # CIS_Offset(b) = CIS_Offset in the LL_CIS_IND associated with CIS sent during setup of the
        # CISes.