diff options
Diffstat (limited to 'resources/libraries/python/model/ExportJson.py')
| -rw-r--r-- | resources/libraries/python/model/ExportJson.py | 216 |
1 files changed, 109 insertions, 107 deletions
diff --git a/resources/libraries/python/model/ExportJson.py b/resources/libraries/python/model/ExportJson.py index 73e822491c..843949eb9f 100644 --- a/resources/libraries/python/model/ExportJson.py +++ b/resources/libraries/python/model/ExportJson.py @@ -43,14 +43,14 @@ from resources.libraries.python.model.validate import ( class ExportJson(): """Class handling the json data setting and export.""" - ROBOT_LIBRARY_SCOPE = u"GLOBAL" + ROBOT_LIBRARY_SCOPE = "GLOBAL" def __init__(self): """Declare required fields, cache output dir. Also memorize schema validator instances. """ - self.output_dir = BuiltIn().get_variable_value(u"\\${OUTPUT_DIR}", ".") + self.output_dir = BuiltIn().get_variable_value("\\${OUTPUT_DIR}", ".") self.file_path = None self.data = None self.validators = get_validators() @@ -62,25 +62,25 @@ class ExportJson(): :rtype: str :raises RuntimeError: If the test tags does not contain expected values. """ - tags = self.data[u"tags"] + tags = self.data["tags"] # First 5 options are specific for VPP tests. - if u"DEVICETEST" in tags: - test_type = u"device" - elif u"LDP_NGINX" in tags: - test_type = u"hoststack" - elif u"HOSTSTACK" in tags: - test_type = u"hoststack" - elif u"GSO_TRUE" in tags or u"GSO_FALSE" in tags: - test_type = u"mrr" - elif u"RECONF" in tags: - test_type = u"reconf" + if "DEVICETEST" in tags: + test_type = "device" + elif "LDP_NGINX" in tags: + test_type = "hoststack" + elif "HOSTSTACK" in tags: + test_type = "hoststack" + elif "GSO_TRUE" in tags or "GSO_FALSE" in tags: + test_type = "mrr" + elif "RECONF" in tags: + test_type = "reconf" # The remaining 3 options could also apply to DPDK and TRex tests. - elif u"SOAK" in tags: - test_type = u"soak" - elif u"NDRPDR" in tags: - test_type = u"ndrpdr" - elif u"MRR" in tags: - test_type = u"mrr" + elif "SOAK" in tags: + test_type = "soak" + elif "NDRPDR" in tags: + test_type = "ndrpdr" + elif "MRR" in tags: + test_type = "mrr" else: raise RuntimeError(f"Unable to infer test type from tags: {tags}") return test_type @@ -108,12 +108,12 @@ class ExportJson(): new_file_path = write_output(self.file_path, self.data) # Data is going to be cleared (as a sign that export succeeded), # so this is the last chance to detect if it was for a test case. - is_testcase = u"result" in self.data + is_testcase = "result" in self.data self.data = None # Validation for output goes here when ready. self.file_path = None if is_testcase: - validate(new_file_path, self.validators[u"tc_info"]) + validate(new_file_path, self.validators["tc_info"]) def warn_on_bad_export(self): """If bad state is detected, log a warning and clean up state.""" @@ -133,25 +133,25 @@ class ExportJson(): """ self.warn_on_bad_export() start_time = datetime.datetime.utcnow().strftime( - u"%Y-%m-%dT%H:%M:%S.%fZ" + "%Y-%m-%dT%H:%M:%S.%fZ" ) - suite_name = BuiltIn().get_variable_value(u"\\${SUITE_NAME}") - suite_id = suite_name.lower().replace(u" ", u"_") - suite_path_part = os.path.join(*suite_id.split(u".")) + suite_name = BuiltIn().get_variable_value("\\${SUITE_NAME}") + suite_id = suite_name.lower().replace(" ", "_") + suite_path_part = os.path.join(*suite_id.split(".")) output_dir = self.output_dir self.file_path = os.path.join( - output_dir, suite_path_part, u"setup.info.json" + output_dir, suite_path_part, "setup.info.json" ) self.data = dict() - self.data[u"version"] = Constants.MODEL_VERSION - self.data[u"start_time"] = start_time - self.data[u"suite_name"] = suite_name - self.data[u"suite_documentation"] = BuiltIn().get_variable_value( - u"\\${SUITE_DOCUMENTATION}" + self.data["version"] = Constants.MODEL_VERSION + self.data["start_time"] = start_time + self.data["suite_name"] = suite_name + self.data["suite_documentation"] = BuiltIn().get_variable_value( + "\\${SUITE_DOCUMENTATION}" ) # "end_time" and "duration" are added on flush. - self.data[u"hosts"] = set() - self.data[u"telemetry"] = list() + self.data["hosts"] = set() + self.data["telemetry"] = list() def start_test_export(self): """Set new file path, initialize data to minimal tree for the test case. @@ -167,30 +167,30 @@ class ExportJson(): """ self.warn_on_bad_export() start_time = datetime.datetime.utcnow().strftime( - u"%Y-%m-%dT%H:%M:%S.%fZ" + "%Y-%m-%dT%H:%M:%S.%fZ" ) - suite_name = BuiltIn().get_variable_value(u"\\${SUITE_NAME}") - suite_id = suite_name.lower().replace(u" ", u"_") - suite_path_part = os.path.join(*suite_id.split(u".")) - test_name = BuiltIn().get_variable_value(u"\\${TEST_NAME}") + suite_name = BuiltIn().get_variable_value("\\${SUITE_NAME}") + suite_id = suite_name.lower().replace(" ", "_") + suite_path_part = os.path.join(*suite_id.split(".")) + test_name = BuiltIn().get_variable_value("\\${TEST_NAME}") self.file_path = os.path.join( self.output_dir, suite_path_part, - test_name.lower().replace(u" ", u"_") + u".info.json" + test_name.lower().replace(" ", "_") + ".info.json" ) self.data = dict() - self.data[u"version"] = Constants.MODEL_VERSION - self.data[u"start_time"] = start_time - self.data[u"suite_name"] = suite_name - self.data[u"test_name"] = test_name - test_doc = BuiltIn().get_variable_value(u"\\${TEST_DOCUMENTATION}", u"") - self.data[u"test_documentation"] = test_doc + self.data["version"] = Constants.MODEL_VERSION + self.data["start_time"] = start_time + self.data["suite_name"] = suite_name + self.data["test_name"] = test_name + test_doc = BuiltIn().get_variable_value("\\${TEST_DOCUMENTATION}", "") + self.data["test_documentation"] = test_doc # "test_type" is added on flush. # "tags" is detected and added on flush. # "end_time" and "duration" is added on flush. # Robot status and message are added on flush. - self.data[u"result"] = dict(type=u"unknown") - self.data[u"hosts"] = BuiltIn().get_variable_value(u"\\${hosts}") - self.data[u"telemetry"] = list() + self.data["result"] = dict(type="unknown") + self.data["hosts"] = BuiltIn().get_variable_value("\\${hosts}") + self.data["telemetry"] = list() export_dut_type_and_version() export_tg_type_and_version() @@ -205,21 +205,21 @@ class ExportJson(): """ self.warn_on_bad_export() start_time = datetime.datetime.utcnow().strftime( - u"%Y-%m-%dT%H:%M:%S.%fZ" + "%Y-%m-%dT%H:%M:%S.%fZ" ) - suite_name = BuiltIn().get_variable_value(u"\\${SUITE_NAME}") - suite_id = suite_name.lower().replace(u" ", u"_") - suite_path_part = os.path.join(*suite_id.split(u".")) + suite_name = BuiltIn().get_variable_value("\\${SUITE_NAME}") + suite_id = suite_name.lower().replace(" ", "_") + suite_path_part = os.path.join(*suite_id.split(".")) self.file_path = os.path.join( - self.output_dir, suite_path_part, u"teardown.info.json" + self.output_dir, suite_path_part, "teardown.info.json" ) self.data = dict() - self.data[u"version"] = Constants.MODEL_VERSION - self.data[u"start_time"] = start_time - self.data[u"suite_name"] = suite_name + self.data["version"] = Constants.MODEL_VERSION + self.data["start_time"] = start_time + self.data["suite_name"] = suite_name # "end_time" and "duration" is added on flush. - self.data[u"hosts"] = BuiltIn().get_variable_value(u"\\${hosts}") - self.data[u"telemetry"] = list() + self.data["hosts"] = BuiltIn().get_variable_value("\\${hosts}") + self.data["telemetry"] = list() def finalize_suite_setup_export(self): """Add the missing fields to data. Do not write yet. @@ -227,9 +227,9 @@ class ExportJson(): Should be run at the end of suite setup. The write is done at next start (or at the end of global teardown). """ - end_time = datetime.datetime.utcnow().strftime(u"%Y-%m-%dT%H:%M:%S.%fZ") - self.data[u"hosts"] = BuiltIn().get_variable_value(u"\\${hosts}") - self.data[u"end_time"] = end_time + end_time = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ") + self.data["hosts"] = BuiltIn().get_variable_value("\\${hosts}") + self.data["end_time"] = end_time self.export_pending_data() def finalize_test_export(self): @@ -240,15 +240,15 @@ class ExportJson(): The write is done at next start (or at the end of global teardown). """ - end_time = datetime.datetime.utcnow().strftime(u"%Y-%m-%dT%H:%M:%S.%fZ") - message = BuiltIn().get_variable_value(u"\\${TEST_MESSAGE}") - test_tags = BuiltIn().get_variable_value(u"\\${TEST_TAGS}") - self.data[u"end_time"] = end_time - start_float = parse(self.data[u"start_time"]).timestamp() - end_float = parse(self.data[u"end_time"]).timestamp() - self.data[u"duration"] = end_float - start_float - self.data[u"tags"] = list(test_tags) - self.data[u"message"] = message + end_time = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ") + message = BuiltIn().get_variable_value("\\${TEST_MESSAGE}") + test_tags = BuiltIn().get_variable_value("\\${TEST_TAGS}") + self.data["end_time"] = end_time + start_float = parse(self.data["start_time"]).timestamp() + end_float = parse(self.data["end_time"]).timestamp() + self.data["duration"] = end_float - start_float + self.data["tags"] = list(test_tags) + self.data["message"] = message self.process_passed() self.process_test_name() self.process_results() @@ -261,8 +261,8 @@ class ExportJson(): (but before the explicit write in the global suite teardown). The write is done at next start (or explicitly for global teardown). """ - end_time = datetime.datetime.utcnow().strftime(u"%Y-%m-%dT%H:%M:%S.%fZ") - self.data[u"end_time"] = end_time + end_time = datetime.datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S.%fZ") + self.data["end_time"] = end_time self.export_pending_data() def process_test_name(self): @@ -292,28 +292,28 @@ class ExportJson(): :raises RuntimeError: If the data does not contain expected values. """ - suite_part = self.data.pop(u"suite_name").lower().replace(u" ", u"_") - if u"test_name" not in self.data: + suite_part = self.data.pop("suite_name").lower().replace(" ", "_") + if "test_name" not in self.data: # There will be no test_id, provide suite_id instead. - self.data[u"suite_id"] = suite_part + self.data["suite_id"] = suite_part return - test_part = self.data.pop(u"test_name").lower().replace(u" ", u"_") - self.data[u"test_id"] = f"{suite_part}.{test_part}" - tags = self.data[u"tags"] + test_part = self.data.pop("test_name").lower().replace(" ", "_") + self.data["test_id"] = f"{suite_part}.{test_part}" + tags = self.data["tags"] # Test name does not contain thread count. - subparts = test_part.split(u"c-", 1) - if len(subparts) < 2 or subparts[0][-2:-1] != u"-": + subparts = test_part.split("c-", 1) + if len(subparts) < 2 or subparts[0][-2:-1] != "-": # Physical core count not detected, assume it is a TRex test. - if u"--" not in test_part: + if "--" not in test_part: raise RuntimeError(f"Cores not found for {subparts}") - short_name = test_part.split(u"--", 1)[1] + short_name = test_part.split("--", 1)[1] else: short_name = subparts[1] # Add threads to test_part. - core_part = subparts[0][-1] + u"c" + core_part = subparts[0][-1] + "c" for tag in tags: tag = tag.lower() - if len(tag) == 4 and core_part == tag[2:] and tag[1] == u"t": + if len(tag) == 4 and core_part == tag[2:] and tag[1] == "t": test_part = test_part.replace(f"-{core_part}-", f"-{tag}-") break else: @@ -321,24 +321,24 @@ class ExportJson(): f"Threads not found for {test_part} tags {tags}" ) # For long name we need NIC model, which is only in suite name. - last_suite_part = suite_part.split(u".")[-1] + last_suite_part = suite_part.split(".")[-1] # Short name happens to be the suffix we want to ignore. prefix_part = last_suite_part.split(short_name)[0] # Also remove the trailing dash. prefix_part = prefix_part[:-1] # Throw away possible link prefix such as "1n1l-". - nic_code = prefix_part.split(u"-", 1)[-1] + nic_code = prefix_part.split("-", 1)[-1] nic_short = Constants.NIC_CODE_TO_SHORT_NAME[nic_code] long_name = f"{nic_short}-{test_part}" # Set test type. test_type = self._detect_test_type() - self.data[u"test_type"] = test_type + self.data["test_type"] = test_type # Remove trailing test type from names (if present). short_name = short_name.split(f"-{test_type}")[0] long_name = long_name.split(f"-{test_type}")[0] # Store names. - self.data[u"test_name_short"] = short_name - self.data[u"test_name_long"] = long_name + self.data["test_name_short"] = short_name + self.data["test_name_long"] = long_name def process_passed(self): """Process the test status information as boolean. @@ -346,12 +346,12 @@ class ExportJson(): Boolean is used to make post processing more efficient. In case the test status is PASS, we will truncate the test message. """ - status = BuiltIn().get_variable_value(u"\\${TEST_STATUS}") + status = BuiltIn().get_variable_value("\\${TEST_STATUS}") if status is not None: - self.data[u"passed"] = (status == u"PASS") - if self.data[u"passed"]: + self.data["passed"] = (status == "PASS") + if self.data["passed"]: # Also truncate success test messages. - self.data[u"message"] = u"" + self.data["message"] = "" def process_results(self): """Process measured results. @@ -364,32 +364,34 @@ class ExportJson(): telemetry_compress = compress(telemetry_encode, level=9) telemetry_base64 = b2a_base64(telemetry_compress, newline=False) self.data["telemetry"] = [telemetry_base64.decode()] - if u"result" not in self.data: + if "result" not in self.data: return - result_node = self.data[u"result"] - result_type = result_node[u"type"] - if result_type == u"unknown": + result_node = self.data["result"] + result_type = result_node["type"] + if result_type == "unknown": # Device or something else not supported. return - # Compute avg and stdev for mrr. - if result_type == u"mrr": - rate_node = result_node[u"receive_rate"][u"rate"] - stats = AvgStdevStats.for_runs(rate_node[u"values"]) - rate_node[u"avg"] = stats.avg - rate_node[u"stdev"] = stats.stdev + # Compute avg and stdev for mrr (rate and bandwidth). + if result_type == "mrr": + for node_name in ("rate", "bandwidth"): + node = result_node["receive_rate"].get(node_name, None) + if node is not None: + stats = AvgStdevStats.for_runs(node["values"]) + node["avg"] = stats.avg + node["stdev"] = stats.stdev return # Multiple processing steps for ndrpdr. - if result_type != u"ndrpdr": + if result_type != "ndrpdr": return # Filter out invalid latencies. - for which_key in (u"latency_forward", u"latency_reverse"): + for which_key in ("latency_forward", "latency_reverse"): if which_key not in result_node: # Probably just an unidir test. continue - for load in (u"pdr_0", u"pdr_10", u"pdr_50", u"pdr_90"): - if result_node[which_key][load][u"max"] <= 0: + for load in ("pdr_0", "pdr_10", "pdr_50", "pdr_90"): + if result_node[which_key][load]["max"] <= 0: # One invalid number is enough to remove all loads. break else: |