How to use _set_forced_source_by_distro method in lisa

Best Python code snippet using lisa_python

dpdkutil.py

Source:dpdkutil.py

...114                "hugepages-1048576kB/nr_hugepages"115            ),116            sudo=True,117        )118def _set_forced_source_by_distro(node: Node, variables: Dict[str, Any]) -> None:119    # DPDK packages 17.11 which is EOL and doesn't have the120    # net_vdev_netvsc pmd used for simple handling of hyper-v121    # guests. Force stable source build on this platform.122    # Default to 20.11 unless another version is provided by the123    # user. 20.11 is the latest dpdk version for 18.04.124    if isinstance(node.os, Ubuntu) and node.os.information.version < "20.4.0":125        variables["dpdk_source"] = variables.get("dpdk_source", DPDK_STABLE_GIT_REPO)126        variables["dpdk_branch"] = variables.get("dpdk_branch", "v20.11")127def get_random_nic_with_ip(test_kit: DpdkTestResources) -> NicInfo:128    nics = [129        test_kit.node.nics.get_nic(nic)130        for nic in test_kit.node.nics.get_upper_nics()131        if nic != test_kit.node.nics.get_nic_by_index(0).upper132        and test_kit.node.nics.get_nic(nic).ip_addr133    ]134    if not nics:135        raise LisaException(136            f"Node has no secondary nics with ip addresses! {test_kit.node.name}"137        )138    return random.choice(nics)139def generate_send_receive_run_info(140    pmd: str,141    sender: DpdkTestResources,142    receiver: DpdkTestResources,143    txq: int = 0,144    rxq: int = 0,145    use_max_nics: bool = False,146    use_service_cores: int = 1,147) -> Dict[DpdkTestResources, str]:148    snd_nic, rcv_nic = [get_random_nic_with_ip(x) for x in [sender, receiver]]149    snd_cmd = sender.testpmd.generate_testpmd_command(150        snd_nic,151        0,152        "txonly",153        pmd,154        extra_args=f"--tx-ip={snd_nic.ip_addr},{rcv_nic.ip_addr}",155        txq=txq,156        rxq=rxq,157        service_cores=use_service_cores,158        use_max_nics=use_max_nics,159    )160    rcv_cmd = receiver.testpmd.generate_testpmd_command(161        rcv_nic,162        0,163        "rxonly",164        pmd,165        txq=txq,166        rxq=rxq,167        service_cores=use_service_cores,168        use_max_nics=use_max_nics,169    )170    kit_cmd_pairs = {171        sender: snd_cmd,172        receiver: rcv_cmd,173    }174    return kit_cmd_pairs175UIO_HV_GENERIC_SYSFS_PATH = "/sys/bus/vmbus/drivers/uio_hv_generic"176HV_NETVSC_SYSFS_PATH = "/sys/bus/vmbus/drivers/hv_netvsc"177def enable_uio_hv_generic_for_nic(node: Node, nic: NicInfo) -> None:178    # hv_uio_generic driver uuid, a constant value used by vmbus.179    # https://doc.dpdk.org/guides/nics/netvsc.html#installation180    hv_uio_generic_uuid = "f8615163-df3e-46c5-913f-f2d2f965ed0e"181    # using netvsc pmd directly for dpdk on hv counterintuitively requires182    # you to enable to uio_hv_generic driver, steps are found:183    # https://doc.dpdk.org/guides/nics/netvsc.html#installation184    echo = node.tools[Echo]185    lsmod = node.tools[Lsmod]186    modprobe = node.tools[Modprobe]187    # enable if it is not already enabled188    if not lsmod.module_exists("uio_hv_generic", force_run=True):189        modprobe.load("uio_hv_generic")190        # vmbus magic to enable uio_hv_generic191        echo.write_to_file(192            hv_uio_generic_uuid,193            node.get_pure_path("/sys/bus/vmbus/drivers/uio_hv_generic/new_id"),194            sudo=True,195        )196def initialize_node_resources(197    node: Node,198    log: Logger,199    variables: Dict[str, Any],200    pmd: str,201    sample_apps: Union[List[str], None] = None,202) -> DpdkTestResources:203    _set_forced_source_by_distro(node, variables)204    dpdk_source = variables.get("dpdk_source", PACKAGE_MANAGER_SOURCE)205    dpdk_branch = variables.get("dpdk_branch", "")206    log.info(207        "Dpdk initialize_node_resources running"208        f"found dpdk_source '{dpdk_source}' and dpdk_branch '{dpdk_branch}'"209    )210    network_interface_feature = node.features[NetworkInterface]211    sriov_is_enabled = network_interface_feature.is_enabled_sriov()212    log.info(f"Node[{node.name}] Verify SRIOV is enabled: {sriov_is_enabled}")213    assert_that(sriov_is_enabled).described_as(214        f"SRIOV was not enabled for this test node ({node.name})"215    ).is_true()216    # dump some info about the pci devices before we start217    lspci = node.tools[Lspci]...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.