+ def _get_ndrpdr_throughput(self, msg):
+ """Get NDR_LOWER, NDR_UPPER, PDR_LOWER and PDR_UPPER from the test
+ message.
+
+ :param msg: The test message to be parsed.
+ :type msg: str
+ :returns: Parsed data as a dict and the status (PASS/FAIL).
+ :rtype: tuple(dict, str)
+ """
+
+ throughput = {
+ "NDR": {"LOWER": -1.0, "UPPER": -1.0},
+ "PDR": {"LOWER": -1.0, "UPPER": -1.0}
+ }
+ status = "FAIL"
+ groups = re.search(self.REGEX_NDRPDR_RATE, msg)
+
+ if groups is not None:
+ try:
+ throughput["NDR"]["LOWER"] = float(groups.group(1))
+ throughput["NDR"]["UPPER"] = float(groups.group(2))
+ throughput["PDR"]["LOWER"] = float(groups.group(3))
+ throughput["PDR"]["UPPER"] = float(groups.group(4))
+ status = "PASS"
+ except (IndexError, ValueError):
+ pass
+
+ return throughput, status
+
+ def _get_ndrpdr_latency(self, msg):
+ """Get LATENCY from the test message.
+
+ :param msg: The test message to be parsed.
+ :type msg: str
+ :returns: Parsed data as a dict and the status (PASS/FAIL).
+ :rtype: tuple(dict, str)
+ """
+
+ latency = {
+ "NDR": {
+ "direction1": {"min": -1.0, "avg": -1.0, "max": -1.0},
+ "direction2": {"min": -1.0, "avg": -1.0, "max": -1.0}
+ },
+ "PDR": {
+ "direction1": {"min": -1.0, "avg": -1.0, "max": -1.0},
+ "direction2": {"min": -1.0, "avg": -1.0, "max": -1.0}
+ }
+ }
+ status = "FAIL"
+ groups = re.search(self.REGEX_NDRPDR_LAT, msg)
+
+ if groups is not None:
+ keys = ("min", "avg", "max")
+ try:
+ latency["NDR"]["direction1"] = dict(
+ zip(keys, [float(l) for l in groups.group(1).split('/')]))
+ latency["NDR"]["direction2"] = dict(
+ zip(keys, [float(l) for l in groups.group(2).split('/')]))
+ latency["PDR"]["direction1"] = dict(
+ zip(keys, [float(l) for l in groups.group(3).split('/')]))
+ latency["PDR"]["direction2"] = dict(
+ zip(keys, [float(l) for l in groups.group(4).split('/')]))
+ status = "PASS"
+ except (IndexError, ValueError):
+ pass
+
+ return latency, status
+