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

How to use create_perf_message method in lisa

Best Python code snippet using lisa_python

iperf3.py

Source:iperf3.py

...308        for client_stream in client_json["end"]["streams"]:309            other_fields["retransmitted_segments"] = client_stream["sender"][310                "retransmits"311            ]312        return create_perf_message(313            NetworkTCPPerformanceMessage,314            self.node,315            test_result,316            test_case_name,317            other_fields,318        )319    def create_iperf_udp_performance_message(320        self,321        server_result_list: List[ExecutableResult],322        client_result_list: List[ExecutableResult],323        buffer_length: int,324        connections_num: int,325        test_case_name: str,326        test_result: "TestResult",327    ) -> NetworkUDPPerformanceMessage:328        client_udp_lost_list: List[Decimal] = []329        client_intervals_throughput_list: List[Decimal] = []330        client_throughput_list: List[Decimal] = []331        for client_result_raw in client_result_list:332            # remove warning which will bring exception when load json333            # warning: UDP block size 8192 exceeds TCP MSS 1406, may result in fragmentation / drops # noqa: E501334            client_result = json.loads(self._pre_handle(client_result_raw.stdout))335            if (336                "sum" in client_result["end"].keys()337                and "lost_percent" in client_result["end"]["sum"].keys()338            ):339                client_udp_lost_list.append(340                    Decimal(client_result["end"]["sum"]["lost_percent"])341                )342                for client_interval in client_result["intervals"]:343                    client_intervals_throughput_list.append(344                        client_interval["sum"]["bits_per_second"]345                    )346                client_throughput_list.append(347                    (348                        Decimal(349                            sum(client_intervals_throughput_list)350                            / len(client_intervals_throughput_list)351                        )352                        / 1000000000353                    )354                )355        server_udp_lost_list: List[Decimal] = []356        server_intervals_throughput_list: List[Decimal] = []357        server_throughput_list: List[Decimal] = []358        for server_result_raw in server_result_list:359            server_result = json.loads(self._pre_handle(server_result_raw.stdout))360            if (361                "sum" in server_result["end"].keys()362                and "lost_percent" in server_result["end"]["sum"].keys()363            ):364                server_udp_lost_list.append(365                    Decimal(server_result["end"]["sum"]["lost_percent"])366                )367                for server_interval in server_result["intervals"]:368                    server_intervals_throughput_list.append(369                        server_interval["sum"]["bits_per_second"]370                    )371                server_throughput_list.append(372                    (373                        Decimal(374                            sum(server_intervals_throughput_list)375                            / len(server_intervals_throughput_list)376                        )377                        / 1000000000378                    )379                )380        other_fields: Dict[str, Any] = {}381        other_fields["tool"] = constants.NETWORK_PERFORMANCE_TOOL_IPERF382        other_fields["tx_throughput_in_gbps"] = Decimal(383            sum(client_throughput_list) / len(client_throughput_list)384        )385        other_fields["data_loss"] = Decimal(386            sum(client_udp_lost_list) / len(client_udp_lost_list)387        )388        other_fields["rx_throughput_in_gbps"] = Decimal(389            sum(server_throughput_list) / len(server_throughput_list)390        )391        other_fields["send_buffer_size"] = Decimal(buffer_length)392        other_fields["connections_num"] = connections_num393        other_fields["protocol_type"] = TransportProtocol.Udp394        return create_perf_message(395            NetworkUDPPerformanceMessage,396            self.node,397            test_result,398            test_case_name,399            other_fields,400        )401    def get_sender_bandwidth(self, result: str) -> Decimal:402        return self._get_bandwidth(result, self._sender_pattern)403    def get_receiver_bandwidth(self, result: str) -> Decimal:404        return self._get_bandwidth(result, self._receiver_pattern)405    def _initialize(self, *args: Any, **kwargs: Any) -> None:406        firewall = self.node.tools[Firewall]407        firewall.stop()408    def _install_from_src(self) -> None:...

dpdkperf.py

Source:dpdkperf.py

...372        receiver_fields["role"] = "receiver/forwarder"373        receiver_fields["rx_pps_maximum"] = receiver.get_max_rx_pps()374        receiver_fields["rx_pps_average"] = receiver.get_mean_rx_pps()375        receiver_fields["rx_pps_minimum"] = receiver.get_min_rx_pps()376        send_results = create_perf_message(377            NetworkPPSPerformanceMessage,378            send_kit.node,379            test_result,380            test_case_name,381            sender_fields,382        )383        receive_results = create_perf_message(384            NetworkPPSPerformanceMessage,385            receive_kit.node,386            test_result,387            test_case_name,388            receiver_fields,389        )390        return send_results, receive_results391    def _validate_core_counts_are_equal(self, test_result: TestResult) -> None:392        environment = test_result.environment393        assert environment, "fail to get environment from testresult"394        core_counts = [395            n.tools[Lscpu].get_core_count() for n in environment.nodes.list()396        ]397        assert_that(core_counts).described_as(...

sar.py

Source:sar.py

...114        result_fields["tx_pps_minimum"] = min(tx_pps)115        result_fields["rx_tx_pps_maximum"] = max(tx_rx_pps)116        result_fields["rx_tx_pps_average"] = Decimal(sum(tx_rx_pps) / len(tx_rx_pps))117        result_fields["rx_tx_pps_minimum"] = min(tx_rx_pps)118        message = create_perf_message(119            NetworkPPSPerformanceMessage,120            self.node,121            test_result,122            test_case_name,123            result_fields,124        )125        return message126    def _initialize(self, *args: Any, **kwargs: Any) -> None:127        firewall = self.node.tools[Firewall]...

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.