r'PDR_UPPER:\s(\d+.\d+)'
)
REGEX_PERF_MSG_INFO = re.compile(
- r'NDR_LOWER:\s(\d+.\d+)\s([a-zA-Z]*).*\s(\d+.\d+)\s([a-zA-Z]*).*\n'
- r'LATENCY.*\[\'(.*)\', \'(.*)\'\].*\n'
- r'NDR_UPPER:\s(\d+.\d+)\s([a-zA-Z]*).*\s(\d+.\d+)\s([a-zA-Z]*).*\n'
- r'PDR_LOWER:\s(\d+.\d+)\s([a-zA-Z]*).*\s(\d+.\d+)\s([a-zA-Z]*).*\n'
- r'LATENCY.*\[\'(.*)\', \'(.*)\'\].*\n'
- r'PDR_UPPER:\s(\d+.\d+)\s([a-zA-Z]*).*\s(\d+.\d+)\s([a-zA-Z]*)'
+ r'NDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n'
+ r'PDR_LOWER:\s(\d+.\d+)\s.*\s(\d+.\d+)\s.*\n.*\n.*\n'
+ r'Latency at 90% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n'
+ r'Latency at 50% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n'
+ r'Latency at 10% PDR:.*\[\'(.*)\', \'(.*)\'\].*\n'
)
+ REGEX_MRR_MSG_INFO = re.compile(r'.*\[(.*)\]')
+
# TODO: Remove when not needed
REGEX_NDRPDR_LAT_BASE = re.compile(
r'LATENCY.*\[\'(.*)\', \'(.*)\'\]\s\n.*\n.*\n'
"""
return self._data
+ 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 u"Test Failed."
+
+ try:
+ data = groups.group(1).split(u", ")
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return u"Test Failed."
+
+ out_str = u"["
+ try:
+ for item in data:
+ out_str += f"{(float(item) / 1e6):.2f}, "
+ return out_str[:-2] + u"]"
+ except (AttributeError, IndexError, ValueError, KeyError):
+ return u"Test Failed."
+
def _get_data_from_perf_test_msg(self, msg):
- """Get
- - NDR_LOWER
- - LATENCY
- - NDR_UPPER
- - PDR_LOWER
- - LATENCY
- - PDR_UPPER
- from message of NDRPDR performance tests.
+ """Get info from message of NDRPDR performance tests.
:param msg: Message to be processed.
:type msg: str
"""
groups = re.search(self.REGEX_PERF_MSG_INFO, msg)
- if not groups or groups.lastindex != 20:
- return msg
+ if not groups or groups.lastindex != 10:
+ return u"Test Failed."
try:
data = {
u"ndr_low": float(groups.group(1)),
- u"ndr_low_unit": groups.group(2),
- u"ndr_low_b": float(groups.group(3)),
- u"ndr_low_b_unit": groups.group(4),
- u"ndr_lat_1": groups.group(5),
- u"ndr_lat_2": groups.group(6),
- u"ndr_up": float(groups.group(7)),
- u"ndr_up_unit": groups.group(8),
- u"ndr_up_b": float(groups.group(9)),
- u"ndr_up_b_unit": groups.group(10),
- u"pdr_low": float(groups.group(11)),
- u"pdr_low_unit": groups.group(12),
- u"pdr_low_b": float(groups.group(13)),
- u"pdr_low_b_unit": groups.group(14),
- u"pdr_lat_1": groups.group(15),
- u"pdr_lat_2": groups.group(16),
- u"pdr_up": float(groups.group(17)),
- u"pdr_up_unit": groups.group(18),
- u"pdr_up_b": float(groups.group(19)),
- u"pdr_up_b_unit": groups.group(20)
+ u"ndr_low_b": float(groups.group(2)),
+ u"pdr_low": float(groups.group(3)),
+ u"pdr_low_b": float(groups.group(4)),
+ u"pdr_lat_90_1": groups.group(5),
+ u"pdr_lat_90_2": groups.group(6),
+ u"pdr_lat_50_1": groups.group(7),
+ u"pdr_lat_50_2": groups.group(8),
+ u"pdr_lat_10_1": groups.group(9),
+ u"pdr_lat_10_2": groups.group(10),
}
except (AttributeError, IndexError, ValueError, KeyError):
- return msg
+ return u"Test Failed."
def _process_lat(in_str_1, in_str_2):
"""Extract min, avg, max values from latency string.
robot framework.
:type in_str_1: str
:type in_str_2: str
- :returns: Processed latency string or original string if a problem
- occurs.
- :rtype: tuple(str, str)
+ :returns: Processed latency string or None if a problem occurs.
+ :rtype: tuple
"""
in_list_1 = in_str_1.split('/', 3)
- if len(in_list_1) < 3:
- return u"Not Measured.", u"Not Measured."
-
in_list_2 = in_str_2.split('/', 3)
- if len(in_list_2) < 3:
- return u"Not Measured.", u"Not Measured."
- hdr_lat_1 = u""
- if len(in_list_1) == 4:
- in_list_1[3] += u"=" * (len(in_list_1[3]) % 4)
- try:
- hdr_lat_1 = hdrh.histogram.HdrHistogram.decode(in_list_1[3])
- except hdrh.codec.HdrLengthException:
- pass
- hdr_lat_2 = u""
- if len(in_list_2) == 4:
- in_list_2[3] += u"=" * (len(in_list_2[3]) % 4)
- try:
- hdr_lat_2 = hdrh.histogram.HdrHistogram.decode(in_list_2[3])
- except hdrh.codec.HdrLengthException:
- pass
+ if len(in_list_1) != 4 and len(in_list_2) != 4:
+ return None
+
+ in_list_1[3] += u"=" * (len(in_list_1[3]) % 4)
+ try:
+ hdr_lat_1 = hdrh.histogram.HdrHistogram.decode(in_list_1[3])
+ except hdrh.codec.HdrLengthException:
+ return None
+
+ in_list_2[3] += u"=" * (len(in_list_2[3]) % 4)
+ try:
+ hdr_lat_2 = hdrh.histogram.HdrHistogram.decode(in_list_2[3])
+ except hdrh.codec.HdrLengthException:
+ return None
- hdr_lat = u"Not Measured."
if hdr_lat_1 and hdr_lat_2:
hdr_lat = (
- f"50%/90%/99%/99.9%, "
- f"{hdr_lat_1.get_value_at_percentile(50.0)}/"
- f"{hdr_lat_1.get_value_at_percentile(90.0)}/"
- f"{hdr_lat_1.get_value_at_percentile(99.0)}/"
- f"{hdr_lat_1.get_value_at_percentile(99.9)}, "
- f"{hdr_lat_2.get_value_at_percentile(50.0)}/"
- f"{hdr_lat_2.get_value_at_percentile(90.0)}/"
- f"{hdr_lat_2.get_value_at_percentile(99.0)}/"
- f"{hdr_lat_2.get_value_at_percentile(99.9)} "
- f"uSec."
+ 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)
)
- return (
- f"Min/Avg/Max, "
- f"{in_list_1[0]}/{in_list_1[1]}/{in_list_1[2]}, "
- f"{in_list_2[0]}/{in_list_2[1]}/{in_list_2[2]} uSec.",
- hdr_lat
- )
+ if all(hdr_lat):
+ return hdr_lat
+
+ return None
try:
- pdr_lat = _process_lat(data[u'pdr_lat_1'], data[u'pdr_lat_2'])
- ndr_lat = _process_lat(data[u'ndr_lat_1'], data[u'ndr_lat_2'])
- return (
- f"NDR Throughput: {(data[u'ndr_low'] / 1e6):.2f} "
- f"M{data[u'ndr_low_unit']}, "
- f"{data[u'ndr_low_b']:.2f} {data[u'ndr_low_b_unit']}.\n"
- f"One-Way Latency at NDR: {ndr_lat[0]}\n"
- f"One-Way Latency at NDR by percentiles: {ndr_lat[1]}\n"
- f"PDR Throughput: {(data[u'pdr_low'] / 1e6):.2f} "
- f"M{data[u'pdr_low_unit']}, "
- f"{data[u'pdr_low_b']:.2f} {data[u'pdr_low_b_unit']}.\n"
- f"One-Way Latency at PDR: {pdr_lat[0]}\n"
- f"One-Way Latency at PDR by percentiles: {pdr_lat[1]}"
+ out_msg = (
+ f"1. {(data[u'ndr_low'] / 1e6):.2f} "
+ f"{data[u'ndr_low_b']:.2f}"
+ f"\n2. {(data[u'pdr_low'] / 1e6):.2f} "
+ f"{data[u'pdr_low_b']:.2f}"
)
+ latency = (
+ _process_lat(data[u'pdr_lat_10_1'], data[u'pdr_lat_10_2']),
+ _process_lat(data[u'pdr_lat_50_1'], data[u'pdr_lat_50_2']),
+ _process_lat(data[u'pdr_lat_90_1'], data[u'pdr_lat_90_2'])
+ )
+ if all(latency):
+ max_len = len(str(max((max(item) for item in latency))))
+
+ for idx, lat in enumerate(latency):
+ if not idx:
+ out_msg += u"\n"
+ out_msg += f"\n{idx + 3}. "
+ for count, itm in enumerate(lat):
+ if count == 3:
+ out_msg += u" " * 6
+ out_msg += u" " * (max_len - len(str(itm)) + 1)
+ out_msg += str(itm)
+
+ return out_msg
+
except (AttributeError, IndexError, ValueError, KeyError):
- return msg
+ return u"Test Failed."
def _get_testbed(self, msg):
"""Called when extraction of testbed IP is required.
replace(u'\r', u'').\
replace(u'[', u' |br| [').\
replace(u' |br| [', u'[', 1)
- test_result[u"msg"] = self._get_data_from_perf_test_msg(test.message).\
+ test_result[u"msg"] = test.message.\
replace(u'\n', u' |br| ').\
replace(u'\r', u'').\
replace(u'"', u"'")
if test.status == u"PASS":
if u"NDRPDR" in tags:
+ test_result[u"msg"] = self._get_data_from_perf_test_msg(
+ test.message). \
+ replace(u'\n', u' |br| '). \
+ replace(u'\r', u''). \
+ replace(u'"', u"'")
test_result[u"type"] = u"NDRPDR"
test_result[u"throughput"], test_result[u"status"] = \
self._get_ndrpdr_throughput(test.message)
groups = re.search(self.REGEX_TCP, test.message)
test_result[u"result"] = int(groups.group(2))
elif u"MRR" in tags or u"FRMOBL" in tags or u"BMRR" in tags:
+ test_result[u"msg"] = self._get_data_from_mrr_test_msg(
+ test.message). \
+ replace(u'\n', u' |br| '). \
+ replace(u'\r', u''). \
+ replace(u'"', u"'")
if u"MRR" in tags:
test_result[u"type"] = u"MRR"
else:
}
except (AttributeError, IndexError, ValueError, TypeError):
test_result[u"status"] = u"FAIL"
+ elif u"DEVICETEST" in tags:
+ test_result[u"type"] = u"DEVICETEST"
else:
test_result[u"status"] = u"FAIL"
self._data[u"tests"][self._test_id] = test_result
Code Review / csit.git / blobdiff