How to use verify_cpu_hot_plug method in lisa

Best Python code snippet using lisa_python

cpusuite.py

Source:cpusuite.py

...63        requirement=simple_requirement(64            min_core_count=32,65        ),66    )67    def verify_cpu_hot_plug(self, log: Logger, node: Node) -> None:68        verify_cpu_hot_plug(log, node)69    @TestCaseMetadata(70        description="""71            This test will check cpu hotplug with storage workload.72            The cpu hotplug steps are same as `verify_cpu_hot_plug` test case.73            """,74        priority=4,75        requirement=simple_requirement(76            disk=schema.DiskOptionSettings(77                data_disk_count=search_space.IntRange(min=1),78                data_disk_size=search_space.IntRange(min=20),79            ),80            min_core_count=16,81        ),82    )83    def verify_cpu_offline_storage_workload(self, log: Logger, node: Node) -> None:84        # run fio process asynchronously on the node85        fio_data_size_in_gb = 186        try:87            image_folder_path = node.find_partition_with_freespace(fio_data_size_in_gb)88            # Each CPU takes ~5 seconds to toggle offline-online89            fio_run_time = 300 + (node.tools[Lscpu].get_core_count() * 5)90            fio_process = node.tools[Fio].launch_async(91                name="workload",92                filename=f"{image_folder_path}/fiodata",93                mode="readwrite",94                iodepth=128,95                numjob=10,96                time=fio_run_time,97                block_size="1M",98                size_gb=fio_data_size_in_gb,99                group_reporting=False,100                overwrite=True,101                time_based=True,102            )103            # Added to find an optional runtime for fio_run_time104            # Remove once test is stable105            hot_plug_start_time = time.time()106            # verify cpu hotplug functionality107            verify_cpu_hot_plug(log, node)108            log.debug(f"CPU Hotplug duration: {time.time() - hot_plug_start_time} s")109            # verify that the fio was running when hotplug was triggered110            assert_that(111                fio_process.is_running(),112                "Storage workload was not running during CPUhotplug",113            ).is_true()114        finally:115            # kill fio process116            node.tools[Kill].by_name("fio", ignore_not_exist=True)117    @TestCaseMetadata(118        description="""119            This test will check cpu hotplug with network workload.120            The cpu hotplug steps are same as `verify_cpu_hot_plug` test case.121            """,122        priority=4,123        requirement=simple_requirement(124            min_count=2,125            min_core_count=16,126        ),127    )128    def verify_cpu_offline_network_workload(129        self, log: Logger, environment: Environment130    ) -> None:131        server = cast(RemoteNode, environment.nodes[0])132        client = cast(RemoteNode, environment.nodes[1])133        try:134            # run iperf3 process asynchronously on the server and client135            server.tools[Iperf3].run_as_server_async()136            client_iperf_process = client.tools[Iperf3].run_as_client_async(137                server.internal_address138            )139            # verify cpu hotplug functionality140            verify_cpu_hot_plug(log, client)141            # verify that the iperf3 was running when hotplug was triggered142            assert_that(143                client_iperf_process.is_running(),144                "Network workload was not running during CPU hotplug",145            ).is_true()146        finally:147            # kill fio process148            server.tools[Kill].by_name("iperf3", ignore_not_exist=True)149            client.tools[Kill].by_name("iperf3", ignore_not_exist=True)150    @TestCaseMetadata(151        description="""152            This test will check that the added channels to synthetic network153            adapter do not handle interrupts on offline cpu.154            Steps:...

common.py

Source:common.py

...105            raise BadEnvironmentStateException(106                exception_message,107                f"the test failed leaving cpu{target_cpu} in a bad state.",108            )109def verify_cpu_hot_plug(log: Logger, node: Node, run_times: int = 1) -> None:110    check_runnable(node)111    file_path_list: Dict[str, str] = {}112    restore_state = False113    try:114        for iteration in range(1, run_times + 1):115            log.debug(f"start the {iteration} time(s) testing.")116            restore_state = False117            # set vmbus channels target cpu into 0 if kernel supports this feature.118            file_path_list = set_interrupts_assigned_cpu(log, node)119            # when kernel doesn't support above feature, we have to rely on current vm's120            # cpu usage. then collect the cpu not in used exclude cpu0.121            idle_cpu = get_idle_cpus(node)122            if 0 == len(idle_cpu):123                raise SkippedException(...

stress.py

Source:stress.py

...28            min_core_count=32,29        ),30    )31    def verify_cpu_hot_plug_stress(self, log: Logger, node: Node) -> None:...

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.