KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use generate_testpmd_command method in lisa

Best Python code snippet using lisa_python

dpdktestpmd.py

Source:dpdktestpmd.py

...226        return min(227            cores_available - 1,228            (nics_available * (txq + rxq)) + (nics_available * service_cores),229        )230    def generate_testpmd_command(231        self,232        nic_to_include: NicInfo,233        vdev_id: int,234        mode: str,235        pmd: str,236        extra_args: str = "",237        txq: int = 0,238        rxq: int = 0,239        service_cores: int = 1,240        use_max_nics: bool = False,241    ) -> str:242        #   testpmd \243        #   -l <core-list> \244        #   -n <num of mem channels> \...

dpdksuite.py

Source:dpdksuite.py

...296    ) -> None:297        test_kit = initialize_node_resources(node, log, variables, "failsafe")298        testpmd = test_kit.testpmd299        test_nic = node.nics.get_nic_by_index()300        testpmd_cmd = testpmd.generate_testpmd_command(301            test_nic, 0, "txonly", "failsafe"302        )303        kit_cmd_pairs = {304            test_kit: testpmd_cmd,305        }306        run_testpmd_concurrent(307            kit_cmd_pairs, DPDK_VF_REMOVAL_MAX_TEST_TIME, log, rescind_sriov=True308        )309        rescind_tx_pps_set = testpmd.get_mean_tx_pps_sriov_rescind()310        self._check_rx_or_tx_pps_sriov_rescind("TX", rescind_tx_pps_set)311    def _check_rx_or_tx_pps_sriov_rescind(312        self, tx_or_rx: str, pps: Tuple[int, int, int]313    ) -> None:314        before_rescind, during_rescind, after_reenable = pps...

dpdkutil.py

Source:dpdkutil.py

...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]218    log.info(f"Node[{node.name}] LSPCI Info:\n{lspci.run().stdout}\n")219    # check compatibility first.220    try:221        check_dpdk_support(node)222    except UnsupportedDistroException as err:223        # forward message from distro exception224        raise SkippedException(err)225    # verify SRIOV is setup as-expected on the node after compat check226    node.nics.wait_for_sriov_enabled()227    # create tool, initialize testpmd tool (installs dpdk)228    testpmd: DpdkTestpmd = node.tools.get(229        DpdkTestpmd,230        dpdk_source=dpdk_source,231        dpdk_branch=dpdk_branch,232        sample_apps=sample_apps,233    )234    # init and enable hugepages (required by dpdk)235    init_hugepages(node)236    assert_that(len(node.nics)).described_as(237        "Test needs at least 1 NIC on the test node."238    ).is_greater_than_or_equal_to(1)239    test_nic = node.nics.get_nic_by_index()240    # check an assumption that our nics are bound to hv_netvsc241    # at test start.242    assert_that(test_nic.bound_driver).described_as(243        f"Error: Expected test nic {test_nic.upper} to be "244        f"bound to hv_netvsc. Found {test_nic.bound_driver}."245    ).is_equal_to("hv_netvsc")246    # netvsc pmd requires uio_hv_generic to be loaded before use247    if pmd == "netvsc":248        # this code makes changes to interfaces that will cause later tests to fail.249        # Therefore we mark the node dirty to prevent future testing on this environment250        node.mark_dirty()251        enable_uio_hv_generic_for_nic(node, test_nic)252        # if this device is paired, set the upper device 'down'253        if test_nic.lower:254            node.nics.unbind(test_nic)255            node.nics.bind(test_nic, UIO_HV_GENERIC_SYSFS_PATH)256    return DpdkTestResources(node, testpmd)257def check_send_receive_compatibility(test_kits: List[DpdkTestResources]) -> None:258    for kit in test_kits:259        if not kit.testpmd.has_tx_ip_flag():260            raise UnsupportedPackageVersionException(261                kit.node.os,262                "dpdk",263                kit.testpmd.get_dpdk_version(),264                "-tx-ip flag for ip forwarding",265            )266def run_testpmd_concurrent(267    node_cmd_pairs: Dict[DpdkTestResources, str],268    seconds: int,269    log: Logger,270    rescind_sriov: bool = False,271) -> Dict[DpdkTestResources, str]:272    output: Dict[DpdkTestResources, str] = dict()273    task_manager = start_testpmd_concurrent(node_cmd_pairs, seconds, log, output)274    if rescind_sriov:275        time.sleep(10)  # run testpmd for a bit before disabling sriov276        test_kits = node_cmd_pairs.keys()277        # disable sriov (and wait for change to apply)278        for node_resources in [x for x in test_kits if x.switch_sriov]:279            node_resources.nic_controller.switch_sriov(280                enable=False, wait=True, reset_connections=False281            )282        # let run for a bit with SRIOV disabled283        time.sleep(10)284        # re-enable sriov285        for node_resources in [x for x in test_kits if x.switch_sriov]:286            node_resources.nic_controller.switch_sriov(287                enable=True, wait=True, reset_connections=False288            )289        # run for a bit with SRIOV re-enabled290        time.sleep(10)291        # kill the commands to collect the output early and terminate before timeout292        for node_resources in test_kits:293            node_resources.testpmd.kill_previous_testpmd_command()294    task_manager.wait_for_all_workers()295    return output296def start_testpmd_concurrent(297    node_cmd_pairs: Dict[DpdkTestResources, str],298    seconds: int,299    log: Logger,300    output: Dict[DpdkTestResources, str],301) -> TaskManager[Tuple[DpdkTestResources, str]]:302    cmd_pairs_as_tuples = deque(node_cmd_pairs.items())303    def _collect_dict_result(result: Tuple[DpdkTestResources, str]) -> None:304        output[result[0]] = result[1]305    def _run_command_with_testkit(306        run_kit: Tuple[DpdkTestResources, str]307    ) -> Tuple[DpdkTestResources, str]:308        testkit, cmd = run_kit309        return (testkit, testkit.testpmd.run_for_n_seconds(cmd, seconds))310    task_manager = run_in_parallel_async(311        [partial(_run_command_with_testkit, x) for x in cmd_pairs_as_tuples],312        _collect_dict_result,313    )314    return task_manager315def init_nodes_concurrent(316    environment: Environment, log: Logger, variables: Dict[str, Any], pmd: str317) -> List[DpdkTestResources]:318    # Use threading module to parallelize the IO-bound node init.319    test_kits = run_in_parallel(320        [321            partial(initialize_node_resources, node, log, variables, pmd)322            for node in environment.nodes.list()323        ],324        log,325    )326    return test_kits327def verify_dpdk_build(328    node: Node,329    log: Logger,330    variables: Dict[str, Any],331    pmd: str,332) -> None:333    # setup and unwrap the resources for this test334    test_kit = initialize_node_resources(node, log, variables, pmd)335    testpmd = test_kit.testpmd336    # grab a nic and run testpmd337    test_nic = node.nics.get_nic_by_index()338    testpmd_cmd = testpmd.generate_testpmd_command(339        test_nic,340        0,341        "txonly",342        pmd,343    )344    testpmd.run_for_n_seconds(testpmd_cmd, 10)345    tx_pps = testpmd.get_mean_tx_pps()346    log.info(347        f"TX-PPS:{tx_pps} from {test_nic.upper}/{test_nic.lower}:"348        + f"{test_nic.pci_slot}"349    )350    assert_that(tx_pps).described_as(351        f"TX-PPS ({tx_pps}) should have been greater than 2^20 (~1m) PPS."352    ).is_greater_than(2**20)...

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.