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

How to use make_by_build_type method in lisa

Best Python code snippet using lisa_python

performance.py

Source:performance.py

...218        assert environment, "fail to get environment from testresult"219        server = environment.nodes[0]220        client = environment.nodes[1]221        server_xdpdump = get_xdpdump(server)222        server_xdpdump.make_by_build_type(BuildType.PERF)223        server_nic = server.nics.get_nic_by_index(1)224        # the latency is not stable in cloud environment, test multiple times225        # and aggregate the result.226        tested_runs = 5227        latency_without_xdp: List[float] = []228        latency_with_xdp: List[float] = []229        for _ in range(tested_runs):230            latency_without_xdp.append(231                self._send_packets_for_latency(server, client, test_result, tool_type)232            )233            try:234                server_xdpdump.start_async(nic_name=server_nic.upper, timeout=0)235                latency_with_xdp.append(236                    self._send_packets_for_latency(237                        server, client, test_result, tool_type238                    )239                )240            finally:241                server_kill = server.tools[Kill]242                server_kill.by_name("xdpdump")243        final_without_xdp = calculate_middle_average(latency_without_xdp)244        final_with_xdp = calculate_middle_average(latency_with_xdp)245        log.info(246            f"Latency with XDP: {final_with_xdp}us, "247            f"without XDP: {final_without_xdp}us. "248            f"Raw with XDP: {latency_with_xdp}, "249            f"without XDP: {latency_without_xdp}. "250        )251        assert_that(final_with_xdp / final_without_xdp).described_as(252            f"The XDP latency: {final_with_xdp}us shouldn't slower 40% than "253            f"the normal latency: {final_without_xdp}us."254        ).is_less_than_or_equal_to(_default_latency_threshold)255    def _send_packets_for_latency(256        self,257        server: Node,258        client: Node,259        test_result: TestResult,260        tool_type: Type[Tool],261    ) -> float:262        assert_that(tool_type).described_as("the tool is not supported").is_in(263            Lagscope, Ntttcp264        )265        # mypy doesn't work with generic type method "get". So use a266        # intermidiate variable tools to store it.267        tools: List[Any] = run_in_parallel(268            [269                partial(server.tools.get, tool_type),270                partial(client.tools.get, tool_type),271            ]272        )273        server_nic = server.nics.get_nic_by_index(1)274        if tool_type is Lagscope:275            server_lagscope: Lagscope = tools[0]276            client_lagscope: Lagscope = tools[1]277            try:278                run_in_parallel(279                    [server_lagscope.set_busy_poll, client_lagscope.set_busy_poll]280                )281                server_lagscope.run_as_server(ip=server_nic.ip_addr)282                result = client_lagscope.run_as_client(server_ip=server_nic.ip_addr)283                lagscope_messages = client_lagscope.create_latency_performance_messages(284                    result=result,285                    test_case_name=inspect.stack()[2].function,286                    test_result=test_result,287                )288                assert lagscope_messages289                assert_that(len(lagscope_messages)).described_as(290                    "at least one message is necessary"291                ).is_greater_than(0)292                return float(293                    sum(x.average_latency_us for x in lagscope_messages)294                    / len(lagscope_messages)295                )296            finally:297                for lagscope in [server_lagscope, client_lagscope]:298                    lagscope.kill()299                    lagscope.restore_busy_poll()300        else:301            ntttcp_messages = perf_ntttcp(302                test_result=test_result,303                udp_mode=False,304                connections=[1],305                test_case_name=inspect.stack()[2].function,306            )307            return float(308                # The type is always TCP message, because the above line set udp309                # to False. Ignore type error here, because UDP message has no310                # latency metrics.311                sum(x.latency_us for x in ntttcp_messages)  # type: ignore312                / len(ntttcp_messages)313            )314    def _execute_rx_drop_test(315        self,316        environment: Environment,317        is_multi_thread: bool,318        log: Logger,319        threshold: float = _default_received_threshold,320    ) -> None:321        sender = environment.nodes[0]322        receiver = environment.nodes[1]323        # install pktgen on sender, and xdpdump on receiver.324        try:325            tools: List[Any] = run_in_parallel(326                [partial(sender.tools.get, Pktgen), partial(get_xdpdump, receiver)]327            )328        except UnsupportedKernelException as identifier:329            raise SkippedException(identifier)330        # type annotations331        pktgen: Pktgen = tools[0]332        xdpdump: XdpDump = tools[1]333        sender_nic = sender.nics.get_nic_by_index(1)334        receiver_nic = receiver.nics.get_nic_by_index(1)335        xdpdump.make_by_build_type(build_type=BuildType.PERF_DROP)336        original_dropped_count = get_dropped_count(337            node=receiver,338            nic=receiver_nic,339            previous_count=0,340            log=log,341        )342        try:343            xdpdump.start_async(nic_name=receiver_nic.upper, timeout=0)344            pktgen_result = self._send_packets(345                is_multi_thread, sender, pktgen, sender_nic, receiver_nic346            )347            self._wait_packets_proceeded(348                log, receiver, receiver_nic, original_dropped_count349            )...

xdpdump.py

Source:xdpdump.py

...126        if not nic_name:127            nic_name = self.node.nics.default_nic128        if not ping_source_node:129            ping_source_node = self.node130        self.make_by_build_type(build_type=build_type)131        # if there is an remote address defined, test it in async mode, and132        # check the ping result.133        if remote_address:134            ping = ping_source_node.tools[Ping]135        xdpdump_process = self.start_async(nic_name=nic_name, timeout=timeout)136        if remote_address:137            is_success = ping.ping(138                remote_address,139                nic_name=nic_name,140                ignore_error=True,141                package_size=ping_package_size,142            )143            assert_that(is_success).described_as(144                "ping result is not expected."145            ).is_equal_to(expected_ping_success)146        result = self.wait_result(nic_name=nic_name, process=xdpdump_process)147        return result.stdout148    def make_by_build_type(self, build_type: Optional[BuildType] = None) -> None:149        env_variables: Dict[str, str] = {}150        # if no build type specified, rebuild it with default behavior.151        if build_type:152            cflags = f"-D __{build_type.name}__ -I../libbpf/src/root/usr/include"153            # no output log to improve perf with high volume data.154            if build_type in [BuildType.PERF_DROP, BuildType.TX_FWD, BuildType.PERF]:155                cflags = f"{cflags} -D __PERF__"156            env_variables["CFLAGS"] = cflags157        make = self.node.tools[Make]158        make.make(159            arguments="",160            cwd=self._code_path,161            is_clean=True,162            update_envs=env_variables,163        )164        # discard local changes after built, it's used to cleanup changes from165        # the forwarder role.166        git = self.node.tools[Git]167        git.discard_local_changes(cwd=self._code_path)168    def make_on_forwarder_role(169        self,170        forwarder_nic: NicInfo,171        receiver_nic: NicInfo,172    ) -> None:173        sed = self.node.tools[Sed]174        # replace hard code mac and ip addresses in code, the changes will be175        # reset after built.176        forwarder_mac = self._convert_mac_to_c_style(forwarder_nic.mac_addr)177        forwarder_ip = self._convert_ip_to_c_style(forwarder_nic.ip_addr)178        receiver_mac = self._convert_mac_to_c_style(receiver_nic.mac_addr)179        receiver_ip = self._convert_ip_to_c_style(receiver_nic.ip_addr)180        sed.substitute(181            regexp=r"unsigned char newethsrc \[\] = "182            "{ 0x00, 0x22, 0x48, 0x4c, 0xc4, 0x4d };",183            replacement=r"unsigned char newethsrc \[\] = {" + forwarder_mac + "};",184            file=f"{self._code_path}/xdpdump_kern.c",185        )186        sed.substitute(187            regexp=r"unsigned char newethdest \[\] = "188            "{ 0x00, 0x22, 0x48, 0x4c, 0xc0, 0xfd };",189            replacement=r"unsigned char newethdest \[\] = {" + receiver_mac + "};",190            file=f"{self._code_path}/xdpdump_kern.c",191        )192        sed.substitute(193            regexp=r"__u8 newsrc \[\] = { 10, 0, 1, 5 };",194            replacement=r"__u8 newsrc \[\] = {" + forwarder_ip + "};",195            file=f"{self._code_path}/xdpdump_kern.c",196        )197        sed.substitute(198            regexp=r"__u8 newdest \[\] = { 10, 0, 1, 4 };",199            replacement=r"__u8 newdest \[\] = {" + receiver_ip + "};",200            file=f"{self._code_path}/xdpdump_kern.c",201        )202        self.make_by_build_type(build_type=BuildType.TX_FWD)203    def _convert_mac_to_c_style(self, mac: str) -> str:204        """205        convert 00:22:48:7a:ed:28 to 0x00, 0x22, 0x48, 0x7a, 0xed, 0x28206        """207        bytes_list = [f"0x{x}" for x in mac.split(":")]208        return ", ".join(bytes_list)209    def _convert_ip_to_c_style(self, ip: str) -> str:210        """211        convert 10.0.0.1 to 10, 0, 0, 1212        """213        return ", ".join(ip.split("."))214    def _disable_lro(self, nic_name: str) -> None:215        ethtool = self.node.tools[Ethtool]216        gro_lro_settings = self._get_gro_lro_settings(nic_name)...

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.