def verify_sriov_ethtool_offload_setting()

in microsoft/testsuites/network/sriov.py [0:0]


    def verify_sriov_ethtool_offload_setting(self, environment: Environment) -> None:
        client_iperf3_log = "iperfResults.log"
        server_node = cast(RemoteNode, environment.nodes[0])
        client_node = cast(RemoteNode, environment.nodes[1])
        client_ethtool = client_node.tools[Ethtool]
        vm_nics = initialize_nic_info(environment)
        # skip test if scatter-gather can't be updated
        for _, client_nic_info in vm_nics[client_node.name].items():
            device_sg_settings = client_ethtool.get_device_sg_settings(
                client_nic_info.upper, True
            )
            if device_sg_settings.sg_fixed:
                raise SkippedException(
                    "scatter-gather is fixed can cannot be changed for device"
                    f" {client_nic_info.upper}. Skipping test."
                )
            else:
                break
        # save original enabled features
        device_enabled_features_origin = client_ethtool.get_all_device_enabled_features(
            True
        )

        # preparation work before launch iperf3
        stop_firewall(environment)

        # run iperf3 on server side and client side
        # iperfResults.log stored client side log
        source_iperf3 = server_node.tools[Iperf3]
        dest_iperf3 = client_node.tools[Iperf3]
        source_iperf3.run_as_server()
        dest_iperf3.run_as_client_async(
            server_node.internal_address, client_iperf3_log, 1800
        )

        # wait for a while then check any error shown up in iperfResults.log
        dest_cat = client_node.tools[Cat]
        iperf_log = dest_cat.read(client_iperf3_log, sudo=True, force_run=True)
        assert_that(iperf_log).does_not_contain("error")

        # disable and enable VF in pci level
        for node in environment.nodes.list():
            lspci = node.tools[Lspci]
            lspci.disable_devices(constants.DEVICE_TYPE_SRIOV)
            lspci.enable_devices()
        # check VF still paired with synthetic nic
        vm_nics = initialize_nic_info(environment)

        # get the enabled features after disable and enable VF
        # make sure there is not any change
        device_enabled_features_after = client_ethtool.get_all_device_enabled_features(
            True
        )
        assert_that(device_enabled_features_origin[0].enabled_features).is_equal_to(
            device_enabled_features_after[0].enabled_features
        )

        # set on for scatter-gather feature for synthetic nic
        # verify vf scatter-gather feature has value 'on'
        for _, client_nic_info in vm_nics[client_node.name].items():
            new_settings = client_ethtool.change_device_sg_settings(
                client_nic_info.upper, True
            )
            device_vf_sg_settings = client_ethtool.get_device_sg_settings(
                client_nic_info.lower, True
            )
            assert_that(
                new_settings.sg_setting,
                "sg setting is not sync into VF.",
            ).is_equal_to(device_vf_sg_settings.sg_setting)

        # set off for scatter-gather feature for synthetic nic
        # verify vf scatter-gather feature has value 'off'
        for _, client_nic_info in vm_nics[client_node.name].items():
            new_settings = client_ethtool.change_device_sg_settings(
                client_nic_info.upper, False
            )
            device_vf_sg_settings = client_ethtool.get_device_sg_settings(
                client_nic_info.lower, True
            )
            assert_that(
                new_settings.sg_setting,
                "sg setting is not sync into VF.",
            ).is_equal_to(device_vf_sg_settings.sg_setting)

        #  disable and enable VF in pci level
        for node in environment.nodes.list():
            lspci = node.tools[Lspci]
            lspci.disable_devices(constants.DEVICE_TYPE_SRIOV)
            lspci.enable_devices()

        # check VF still paired with synthetic nic
        vm_nics = initialize_nic_info(environment)

        # check VF's scatter-gather feature keep consistent with previous status
        for _, client_nic_info in vm_nics[client_node.name].items():
            device_vf_sg_settings = client_ethtool.get_device_sg_settings(
                client_nic_info.lower, True
            )
            assert_that(
                device_vf_sg_settings.sg_setting,
                "sg setting is not sync into VF.",
            ).is_equal_to(False)

        # disable and enable sriov in network interface level
        network_interface_feature = client_node.features[NetworkInterface]
        for _ in range(3):
            sriov_is_enabled = network_interface_feature.is_enabled_sriov()
            network_interface_feature.switch_sriov(enable=(not sriov_is_enabled))
        network_interface_feature.switch_sriov(enable=True)

        # check VF still paired with synthetic nic
        vm_nics = initialize_nic_info(environment)

        # check VF's scatter-gather feature keep consistent with previous status
        for _, client_nic_info in vm_nics[client_node.name].items():
            device_vf_sg_settings = client_ethtool.get_device_sg_settings(
                client_nic_info.lower, True
            )
            assert_that(
                device_vf_sg_settings.sg_setting,
                "sg setting is not sync into VF.",
            ).is_equal_to(False)

        # reload sriov modules
        module_in_used = ""
        for node in environment.nodes.list():
            module_in_used = remove_module(node)
        for node in environment.nodes.list():
            load_module(node, module_in_used)

        # check VF still paired with synthetic nic
        vm_nics = initialize_nic_info(environment)

        # check VF's scatter-gather feature keep consistent with previous status
        for _, client_nic_info in vm_nics[client_node.name].items():
            device_vf_sg_settings = client_ethtool.get_device_sg_settings(
                client_nic_info.lower, True
            )
            assert_that(
                device_vf_sg_settings.sg_setting,
                "sg setting is not sync into VF.",
            ).is_equal_to(False)

        # check there is no error happen in iperf3 log
        # after above operations
        dest_cat = client_node.tools[Cat]
        iperf_log = dest_cat.read(client_iperf3_log, sudo=True, force_run=True)
        assert_that(iperf_log).does_not_contain("error")