How to use create_perf_message method in lisa

Best Python code snippet using lisa_python

iperf3.py

Source:iperf3.py Github

copy

Full Screen

...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:...

Full Screen

Full Screen

dpdkperf.py

Source:dpdkperf.py Github

copy

Full Screen

...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(...

Full Screen

Full Screen

sar.py

Source:sar.py Github

copy

Full Screen

...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]...

Full Screen

Full Screen

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.

LambdaTest Learning Hubs:

YouTube

You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.

Run lisa automation tests on LambdaTest cloud grid

Perform automation testing on 3000+ real desktop and mobile devices online.

Try LambdaTest Now !!

Get 100 minutes of automation test minutes FREE!!

Next-Gen App & Browser Testing Cloud

Was this article helpful?

Helpful

NotHelpful