Quelle  BV_40_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.

from dataclasses import dataclass
import hci_packets as hci
import link_layer_packets as ll
import unittest
from hci_packets import ErrorCode
from py.bluetooth import Address
from py.controller import ControllerTest


@dataclass
class TestRound:
    req_all_phys: int
    req_tx_phys: int
    req_rx_phys: int
    phy_ltpref_c_to_p: int
    phy_ltpref_p_to_c: int


class Test(ControllerTest):

    # LL/CON/PER/BV-40-C [Initiating PHY Update Procedure]
    async def test(self):
        # Test parameters.
        controller = self.controller
        acl_connection_handle = None
        peer_address = Address('11:22:33:44:55:66')

        # Prelude: Establish an ACL connection as central with the IUT.
        controller.send_cmd(
            hci.LeSetAdvertisingParameters(
                advertising_interval_min=0x200,
                advertising_interval_max=0x200,
                advertising_type=hci.AdvertisingType.ADV_IND,
                own_address_type=hci.OwnAddressType.PUBLIC_DEVICE_ADDRESS,
                advertising_channel_map=0x7,
                advertising_filter_policy=hci.AdvertisingFilterPolicy.ALL_DEVICES))

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

        controller.send_cmd(hci.LeSetAdvertisingEnable(advertising_enable=True))

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

        controller.send_ll(ll.LeConnect(source_address=peer_address,
                                        destination_address=controller.address,
                                        initiating_address_type=ll.AddressType.PUBLIC,
                                        advertising_address_type=ll.AddressType.PUBLIC,
                                        conn_interval=0x200,
                                        conn_peripheral_latency=0x200,
                                        conn_supervision_timeout=0x200),
                           rssi=-16)

        await self.expect_ll(
            ll.LeConnectComplete(source_address=controller.address,
                                 destination_address=peer_address,
                                 conn_interval=0x200,
                                 conn_peripheral_latency=0x200,
                                 conn_supervision_timeout=0x200))

        evt = await self.expect_evt(
            hci.LeEnhancedConnectionCompleteV1(
                status=ErrorCode.SUCCESS,
                connection_handle=self.Any,
                role=hci.Role.PERIPHERAL,
                peer_address_type=hci.AddressType.PUBLIC_DEVICE_ADDRESS,
                peer_address=peer_address,
                connection_interval=0x200,
                peripheral_latency=0x200,
                supervision_timeout=0x200,
                central_clock_accuracy=hci.ClockAccuracy.PPM_500))

        acl_connection_handle = evt.connection_handle

        test_rounds = [
            TestRound(0x00, 0x01, 0x03, 0x02, 0x00),
            TestRound(0x00, 0x02, 0x05, 0x01, 0x02),
            TestRound(0x00, 0x04, 0x02, 0x02, 0x04),
            TestRound(0x00, 0x02, 0x01, 0x00, 0x02),
            TestRound(0x00, 0x01, 0x04, 0x04, 0x00),
            TestRound(0x00, 0x06, 0x03, 0x02, 0x02),
            TestRound(0x00, 0x01, 0x01, 0x01, 0x01),
            TestRound(0x00, 0x03, 0x04, 0x00, 0x01),
            TestRound(0x00, 0x01, 0x05, 0x00, 0x00),
            TestRound(0x00, 0x02, 0x04, 0x04, 0x02),
            TestRound(0x00, 0x07, 0x03, 0x01, 0x00),
            TestRound(0x00, 0x06, 0x06, 0x02, 0x04),
            TestRound(0x00, 0x02, 0x03, 0x02, 0x02),
            TestRound(0x00, 0x06, 0x01, 0x01, 0x04),
            TestRound(0x00, 0x06, 0x05, 0x01, 0x04),
            TestRound(0x00, 0x05, 0x04, 0x04, 0x01),
            TestRound(0x00, 0x05, 0x01, 0x01, 0x00),
            TestRound(0x00, 0x03, 0x05, 0x01, 0x02),
            TestRound(0x00, 0x04, 0x06, 0x02, 0x04),
            TestRound(0x00, 0x07, 0x02, 0x00, 0x01),
            TestRound(0x00, 0x01, 0x06, 0x02, 0x00),
            TestRound(0x00, 0x02, 0x02, 0x02, 0x00),
            TestRound(0x00, 0x04, 0x03, 0x01, 0x04),
            TestRound(0x00, 0x03, 0x07, 0x04, 0x01),
            TestRound(0x00, 0x01, 0x02, 0x02, 0x01),
            TestRound(0x00, 0x03, 0x03, 0x01, 0x01),
            TestRound(0x00, 0x03, 0x02, 0x02, 0x01),
            TestRound(0x00, 0x07, 0x04, 0x00, 0x02),
            TestRound(0x00, 0x04, 0x07, 0x00, 0x00),
            TestRound(0x00, 0x01, 0x07, 0x04, 0x00),
            TestRound(0x00, 0x05, 0x06, 0x04, 0x01),
            TestRound(0x00, 0x06, 0x02, 0x00, 0x04),
            TestRound(0x00, 0x07, 0x07, 0x01, 0x01),
            TestRound(0x00, 0x06, 0x04, 0x04, 0x02),
            TestRound(0x00, 0x05, 0x02, 0x02, 0x01),
            TestRound(0x00, 0x02, 0x06, 0x04, 0x00),
            TestRound(0x00, 0x02, 0x07, 0x01, 0x00),
            TestRound(0x00, 0x06, 0x07, 0x04, 0x04),
            TestRound(0x00, 0x07, 0x06, 0x02, 0x02),
            TestRound(0x00, 0x03, 0x06, 0x00, 0x02),
            TestRound(0x00, 0x04, 0x05, 0x04, 0x04),
            TestRound(0x00, 0x05, 0x07, 0x04, 0x01),
            TestRound(0x00, 0x07, 0x01, 0x00, 0x04),
            TestRound(0x00, 0x06, 0x06, 0x02, 0x02),
            TestRound(0x00, 0x06, 0x06, 0x04, 0x04),
            TestRound(0x01, 0x00, 0x01, 0x01, 0x01),
            TestRound(0x02, 0x03, 0x00, 0x02, 0x02),
            TestRound(0x03, 0x00, 0x00, 0x01, 0x01),
            TestRound(0x02, 0x05, 0x00, 0x04, 0x04),
            TestRound(0x03, 0x00, 0x00, 0x04, 0x04),
            TestRound(0x03, 0x00, 0x00, 0x02, 0x02),
            TestRound(0x01, 0x00, 0x06, 0x04, 0x04),
            TestRound(0x01, 0x00, 0x06, 0x04, 0x02),
            TestRound(0x02, 0x07, 0x00, 0x02, 0x01),
            TestRound(0x02, 0x06, 0x00, 0x02, 0x02),
            # Disabled test cases:
            # the values are invalid as RFU, not certain
            # if that is intended by the TS or an
            # error.
            #TestRound(0x01, 0x00, 0x08, 0x02, 0x02),
            #TestRound(0x01, 0x00, 0x11, 0x01, 0x00),
            #TestRound(0x01, 0x00, 0x20, 0x02, 0x01),
            #TestRound(0x01, 0x00, 0x41, 0x01, 0x01),
            #TestRound(0x01, 0x00, 0x80, 0x01, 0x02),
            #TestRound(0x01, 0x00, 0xFF, 0x00, 0x01),
            #TestRound(0x02, 0x08, 0x00, 0x00, 0x00),
            #TestRound(0x02, 0x11, 0x00, 0x02, 0x00),
            #TestRound(0x02, 0x20, 0x00, 0x00, 0x02),
            #TestRound(0x02, 0x41, 0x00, 0x01, 0x01),
            #TestRound(0x02, 0x80, 0x00, 0x02, 0x02),
            #TestRound(0x02, 0xFF, 0x00, 0x01, 0x01),
        ]

        # Repeat steps 1-9 for each Round shown in Table 4.77.
        phy_c_to_p = 0x1
        phy_p_to_c = 0x1
        for test_round in test_rounds:
            (phy_c_to_p, phy_p_to_c) = await self.steps_1_9(peer_address, acl_connection_handle,
                                                            phy_c_to_p, phy_p_to_c,
                                                            **vars(test_round))

    async def steps_1_9(self, peer_address: Address, connection_handle: int, phy_c_to_p: int,
                        phy_p_to_c: int, req_all_phys: int, req_tx_phys: int, req_rx_phys: int,
                        phy_ltpref_c_to_p: int, phy_ltpref_p_to_c: int):
        controller = self.controller

        def phy_from_mask(mask: int):
            if mask & 0x4:
                return hci.PhyType.LE_CODED
            elif mask & 0x2:
                return hci.PhyType.LE_2M
            else:
                return hci.PhyType.LE_1M

        # 1. Upper Tester sends an HCI_LE_Set_PHY command to the IUT with the payload defined in the
        # HCI_LE_Set_PHY section of Table 4.66 and PHY_options set to 0x0000.
        controller.send_cmd(
            hci.LeSetPhy(connection_handle=connection_handle,
                         all_phys_no_transmit_preference=(req_all_phys & 0x1) != 0,
                         all_phys_no_receive_preference=(req_all_phys & 0x2) != 0,
                         tx_phys=req_tx_phys,
                         rx_phys=req_rx_phys,
                         phy_options=hci.PhyOptions.NO_PREFERENCE))

        # 2. The Upper Tester receives an HCI_Command_Status event from the IUT in response. If any bits
        # set in TX_PHYS or RX_PHYS correspond to unsupported PHYs, the Status shall be set to
        # “Unsupported Feature or Parameter Value (0x11)”. If the IUT does not support Asymmetric
        # Connections, when ALL_PHYS is 0x00 and TX_PHYS does not equal RX_PHYS, the Status
        # shall be set to “Unsupported Feature or Parameter Value (0x11)”. Otherwise. the Status shall be
        # set to zero.
        await self.expect_evt(
            hci.LeSetPhyStatus(status=ErrorCode.SUCCESS, num_hci_command_packets=1))

        # 3. If the IUT does NOT initiate a PHY change, proceed to step 9 if the Status in step 2 was set to
        # zero, or proceed to the next round if the Status in step 2 was set to a non-zero value.

        if (req_all_phys & 0x1) != 0:
            req_tx_phys = 0x7
        if (req_all_phys & 0x2) != 0:
            req_rx_phys = 0x7

        # 4. The Lower Tester receives an LL_PHY_REQ control PDU from the IUT with at least one bit in
        # each field (TX_PHYS, RX_PHYS) set. The Lower Tester responds with an
        # LL_PHY_UPDATE_IND PDU with the values defined in the “Lower Tester preference” section of
        # Table 4.66 bitwise ANDed against the value sent by the IUT in the LL_PHY_REQ PDU.
        await self.expect_ll(
            ll.LlPhyReq(source_address=controller.address,
                        destination_address=peer_address,
                        tx_phys=req_tx_phys,
                        rx_phys=req_rx_phys))

        next_phy_c_to_p = (req_rx_phys & phy_ltpref_c_to_p) or phy_c_to_p
        next_phy_p_to_c = (req_tx_phys & phy_ltpref_p_to_c) or phy_p_to_c
        controller.send_ll(
            ll.LlPhyUpdateInd(source_address=peer_address,
                              destination_address=controller.address,
                              phy_c_to_p=(0 if next_phy_c_to_p == phy_c_to_p else next_phy_c_to_p),
                              phy_p_to_c=(0 if next_phy_p_to_c == phy_p_to_c else next_phy_p_to_c)))

        # 5. Lower Tester receives a packet from the IUT acknowledging the LL_PHY_UPDATE_IND.

        # 6. Lower Tester sends empty DATA packets to the IUT, receiving acknowledgements until the event
        # count matches the Instant indicated in the LL_PHY_UPDATE_IND packet.

        # 7. At the Instant of the PHY change, start maintaining the connection with the new PHY(s) selected
        # by the LL_PHY_UPDATE_IND PDU (or no change, if no change was specified in the
        # LL_PHY_UPDATE_IND PDU).

        # 8. Lower Tester sends empty DATA packets to the IUT, receiving acknowledgements. If PHY has
        # changed, the Lower Tester shall use the new PHY.

        # 9. If the command was accepted in step 2 or at least one of the PHY fields in the
        # LL_PHY_UPDATE_IND PDU was non-zero, the Upper Tester receives an
        # LE_PHY_Update_Complete event from the IUT with a payload consistent with the PHY(s)
        # indicated in the LL_PHY_UPDATE_IND PDU (or the prior PHY, in cases where a field in the
        # LL_PHY_UPDATE_IND PDU was zero or the LL_PHY_UPDATE_IND PDU was not sent).
        # Otherwise the Upper Tester receives no event.
        await self.expect_evt(
            hci.LePhyUpdateComplete(connection_handle=connection_handle,
                                    status=ErrorCode.SUCCESS,
                                    tx_phy=phy_from_mask(next_phy_p_to_c),
                                    rx_phy=phy_from_mask(next_phy_c_to_p)))

        return (next_phy_c_to_p, next_phy_p_to_c)