+ def _get_data_from_mrr_test_msg(self, msg):
+ """Get info from message of MRR performance tests.
+
+ :param msg: Message to be processed.
+ :type msg: str
+ :returns: Processed message or original message if a problem occurs.
+ :rtype: str
+ """
+
+ groups = re.search(self.REGEX_MRR_MSG_INFO, msg)
+ if not groups or groups.lastindex != 1:
+ return "Test Failed."
+
+ try:
+ data = groups.group(1).split(", ")
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return "Test Failed."
+
+ out_str = "["
+ try:
+ for item in data:
+ out_str += f"{(float(item) / 1e6):.2f}, "
+ return out_str[:-2] + "]"
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return "Test Failed."
+
+ def _get_data_from_cps_test_msg(self, msg):
+ """Get info from message of NDRPDR CPS tests.
+
+ :param msg: Message to be processed.
+ :type msg: str
+ :returns: Processed message or "Test Failed." if a problem occurs.
+ :rtype: str
+ """
+
+ groups = re.search(self.REGEX_CPS_MSG_INFO, msg)
+ if not groups or groups.lastindex != 2:
+ return "Test Failed."
+
+ try:
+ return (
+ f"1. {(float(groups.group(1)) / 1e6):5.2f}\n"
+ f"2. {(float(groups.group(2)) / 1e6):5.2f}"
+ )
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return "Test Failed."
+
+ def _get_data_from_pps_test_msg(self, msg):
+ """Get info from message of NDRPDR PPS tests.
+
+ :param msg: Message to be processed.
+ :type msg: str
+ :returns: Processed message or "Test Failed." if a problem occurs.
+ :rtype: str
+ """
+
+ groups = re.search(self.REGEX_PPS_MSG_INFO, msg)
+ if not groups or groups.lastindex != 4:
+ return "Test Failed."
+
+ try:
+ return (
+ f"1. {(float(groups.group(1)) / 1e6):5.2f} "
+ f"{float(groups.group(2)):5.2f}\n"
+ f"2. {(float(groups.group(3)) / 1e6):5.2f} "
+ f"{float(groups.group(4)):5.2f}"
+ )
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return "Test Failed."
+
+ def _get_data_from_perf_test_msg(self, msg):
+ """Get info from message of NDRPDR performance tests.
+
+ :param msg: Message to be processed.
+ :type msg: str
+ :returns: Processed message or "Test Failed." if a problem occurs.
+ :rtype: str
+ """
+
+ groups = re.search(self.REGEX_PERF_MSG_INFO, msg)
+ if not groups or groups.lastindex != 10:
+ return "Test Failed."
+
+ try:
+ data = {
+ "ndr_low": float(groups.group(1)),
+ "ndr_low_b": float(groups.group(2)),
+ "pdr_low": float(groups.group(3)),
+ "pdr_low_b": float(groups.group(4)),
+ "pdr_lat_90_1": groups.group(5),
+ "pdr_lat_90_2": groups.group(6),
+ "pdr_lat_50_1": groups.group(7),
+ "pdr_lat_50_2": groups.group(8),
+ "pdr_lat_10_1": groups.group(9),
+ "pdr_lat_10_2": groups.group(10),
+ }
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return "Test Failed."
+
+ def _process_lat(in_str_1, in_str_2):
+ """Extract P50, P90 and P99 latencies or min, avg, max values from
+ latency string.
+
+ :param in_str_1: Latency string for one direction produced by robot
+ framework.
+ :param in_str_2: Latency string for second direction produced by
+ robot framework.
+ :type in_str_1: str
+ :type in_str_2: str
+ :returns: Processed latency string or None if a problem occurs.
+ :rtype: tuple
+ """
+ in_list_1 = in_str_1.split('/', 3)
+ in_list_2 = in_str_2.split('/', 3)
+
+ if len(in_list_1) != 4 and len(in_list_2) != 4:
+ return None
+
+ in_list_1[3] += "=" * (len(in_list_1[3]) % 4)
+ try:
+ hdr_lat_1 = hdrh.histogram.HdrHistogram.decode(in_list_1[3])
+ except hdrh.codec.HdrLengthException:
+ hdr_lat_1 = None
+
+ in_list_2[3] += "=" * (len(in_list_2[3]) % 4)
+ try:
+ hdr_lat_2 = hdrh.histogram.HdrHistogram.decode(in_list_2[3])
+ except hdrh.codec.HdrLengthException:
+ hdr_lat_2 = None
+
+ if hdr_lat_1 and hdr_lat_2:
+ hdr_lat = (
+ hdr_lat_1.get_value_at_percentile(50.0),
+ hdr_lat_1.get_value_at_percentile(90.0),
+ hdr_lat_1.get_value_at_percentile(99.0),
+ hdr_lat_2.get_value_at_percentile(50.0),
+ hdr_lat_2.get_value_at_percentile(90.0),
+ hdr_lat_2.get_value_at_percentile(99.0)
+ )
+ if all(hdr_lat):
+ return hdr_lat
+
+ hdr_lat = (
+ int(in_list_1[0]), int(in_list_1[1]), int(in_list_1[2]),
+ int(in_list_2[0]), int(in_list_2[1]), int(in_list_2[2])
+ )
+ for item in hdr_lat:
+ if item in (-1, 4294967295, 0):
+ return None
+ return hdr_lat
+
+ try:
+ out_msg = (
+ f"1. {(data['ndr_low'] / 1e6):5.2f} "
+ f"{data['ndr_low_b']:5.2f}"
+ f"\n2. {(data['pdr_low'] / 1e6):5.2f} "
+ f"{data['pdr_low_b']:5.2f}"
+ )
+ latency = (
+ _process_lat(data['pdr_lat_10_1'], data['pdr_lat_10_2']),
+ _process_lat(data['pdr_lat_50_1'], data['pdr_lat_50_2']),
+ _process_lat(data['pdr_lat_90_1'], data['pdr_lat_90_2'])
+ )
+ if all(latency):
+ max_len = len(str(max((max(item) for item in latency))))
+ max_len = 4 if max_len < 4 else max_len
+
+ for idx, lat in enumerate(latency):
+ if not idx:
+ out_msg += "\n"
+ out_msg += (
+ f"\n{idx + 3}. "
+ f"{lat[0]:{max_len}d} "
+ f"{lat[1]:{max_len}d} "
+ f"{lat[2]:{max_len}d} "
+ f"{lat[3]:{max_len}d} "
+ f"{lat[4]:{max_len}d} "
+ f"{lat[5]:{max_len}d} "
+ )
+
+ return out_msg
+
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return "Test Failed."
+
Code Review / csit.git / blobdiff