[test_reporting] Refactor parser and upload script to allow JSON files as input

test_reporting/junit_xml_parser.py

@@ -72,12 +72,17 @@
     "name",
     "time",
 ]
+REQUIRED_TESTCASE_JSON_FIELDS = ["result", "error", "summary"]
 
 
 class JUnitXMLValidationError(Exception):
     """Expected errors that are thrown while validating the contents of the JUnit XML file."""
 
 
+class TestResultJSONValidationError(Exception):
+    """Expected errors that are trhown while validating the contents of the Test Result JSON file."""
+
+
 def validate_junit_xml_stream(stream):
     """Validate that a stream containing an XML document is valid JUnit XML.
 
@@ -199,6 +204,15 @@ def validate_junit_xml_archive(directory_name, strict=False):
     return roots
 
 
+def validate_junit_xml_path(path, strict=False):
+    if os.path.isfile(path):
+        roots = [validate_junit_xml_file(path)]
+    else:
+        roots = validate_junit_xml_archive(path, strict)
+
+    return roots
+
+
 def _validate_junit_xml(root):
     _validate_test_summary(root)
     _validate_test_metadata(root)
@@ -429,6 +443,104 @@ def _update_test_cases(current, update):
     return new_cases
 
 
+def validate_junit_json_file(path):
+    """Validate that a JSON file is a valid test report.
+
+    Args:
+        path: The path to the JSON file.
+
+    Returns:
+        The validated JSON file.
+
+    Raises:
+        TestResultJSONValidationError: if any of the following are true:
+            - The provided file doesn't exist
+            - The provided file is unparseable
+            - The provided file is missing required fields
+    """
+    if not os.path.exists(path):
+        print(f"{path} not found")
+        sys.exit(1)
+
+    if not os.path.isfile(path):
+        print(f"{path} is not a JSON file")
+        sys.exit(1)
+
+    try:
+        with open(path) as f:
+            test_result_json = json.load(f)
+    except Exception as e:
+        raise TestResultJSONValidationError(f"Could not load JSON file {path}: {e}") from e
+
+    _validate_json_metadata(test_result_json)
+    _validate_json_summary(test_result_json)
+    _validate_json_cases(test_result_json)
+
+    return test_result_json
+
+
+def _validate_json_metadata(test_result_json):
+    if "test_metadata" not in test_result_json:
+        raise TestResultJSONValidationError("test_metadata section not found in provided JSON file")
+
+    seen_properties = []
+    for prop, value in test_result_json["test_metadata"].items():
+        if prop not in REQUIRED_METADATA_PROPERTIES:
+            continue
+
+        if prop in seen_properties:
+            raise TestResultJSONValidationError(
+                f"duplicate metadata element: {prop} seen more than once"
+            )
+
+        if value is None:  # Some fields may be empty
+            raise TestResultJSONValidationError(
+                f'invalid metadata element: no "value" field provided for {prop}'
+            )
+
+        seen_properties.append(prop)
+
+    if set(seen_properties) < set(REQUIRED_METADATA_PROPERTIES):
+        raise TestResultJSONValidationError("missing metadata element(s)")
+
+
+def _validate_json_summary(test_result_json):
+    if "test_summary" not in test_result_json:
+        raise TestResultJSONValidationError("test_summary section not found in provided JSON file")
+
+    summary = test_result_json["test_summary"]
+
+    for field, expected_type in REQUIRED_TESTSUITE_ATTRIBUTES:
+        if field not in summary:
+            raise TestResultJSONValidationError(f"{field} not found in test_summary section")
+
+        try:
+            expected_type(summary[field])
+        except Exception as e:
+            raise TestResultJSONValidationError(
+                f"invalid type for {field} in test_summary section: "
+                f"expected a number, received "
+                f'"{summary[field]}"'
+            ) from e
+
+
+def _validate_json_cases(test_result_json):
+    if "test_cases" not in test_result_json:
+        raise TestResultJSONValidationError("test_cases section not found in provided JSON file")
+
+    def _validate_test_case(test_case):
+        for attribute in REQUIRED_TESTCASE_ATTRIBUTES + REQUIRED_TESTCASE_JSON_FIELDS:
+            if attribute not in test_case:
+                raise TestResultJSONValidationError(
+                    f'"{attribute}" not found in test case '
+                    f"\"{test_case.get('name', 'Name Not Found')}\""
+                )
+
+    for _, feature in test_result_json["test_cases"].items():
+        for test_case in feature:
+            _validate_test_case(test_case)
+
+
 def _run_script():
     parser = argparse.ArgumentParser(
         description="Validate and convert SONiC JUnit XML files into JSON.",
@@ -457,22 +569,33 @@ def _run_script():
         action="store_true",
         help="Fail validation checks if ANY file in a given directory is not parseable."
     )
+    parser.add_argument(
+        "--json",
+        "-j",
+        action="store_true",
+        help="Load an existing test result JSON file from path_name. Will perform validation only regardless of --validate-only option.",
+    )
 
     args = parser.parse_args()
 
     try:
-        if args.directory:
+        if args.json:
+            validate_junit_json_file(args.file_name)
+        elif args.directory:
             roots = validate_junit_xml_archive(args.file_name, args.strict)
         else:
             roots = [validate_junit_xml_file(args.file_name)]
     except JUnitXMLValidationError as e:
         print(f"XML validation failed: {e}")
         sys.exit(1)
+    except TestResultJSONValidationError as e:
+        print(f"JSON validation failed: {e}")
+        sys.exit(1)
     except Exception as e:
         print(f"Unexpected error occured during validation: {e}")
         sys.exit(2)
 
-    if args.validate_only:
+    if args.validate_only or args.json:
         print(f"{args.file_name} validated succesfully!")
         sys.exit(0)
 

test_reporting/report_uploader.py

@@ -1,10 +1,8 @@
 import argparse
-import os
-import sys
 
 from junit_xml_parser import (
-    validate_junit_xml_file,
-    validate_junit_xml_archive,
+    validate_junit_json_file,
+    validate_junit_xml_path,
     parse_test_result
 )
 from report_data_storage import KustoConnector
@@ -24,23 +22,18 @@ def _run_script():
     parser.add_argument(
         "--external_id", "-e", type=str, help="An external tracking ID to append to the report.",
     )
+    parser.add_argument(
+        "--json", "-j", action="store_true", help="Load an existing test result JSON file from path_name.",
+    )
 
     args = parser.parse_args()
 
-    path = args.path_name
-
-    if not os.path.exists(path):
-        print(f"{path} not found")
-        sys.exit(1)
-
-    # FIXME: This interface is actually really clunky, should just have one method and check file
-    # v. dir internally. Fix in the next PR.
-    if os.path.isfile(path):
-        roots = [validate_junit_xml_file(path)]
+    if args.json:
+        test_result_json = validate_junit_json_file(args.path_name)
     else:
-        roots = validate_junit_xml_archive(path)
+        roots = validate_junit_xml_path(args.path_name)
+        test_result_json = parse_test_result(roots)
 
-    test_result_json = parse_test_result(roots)
     tracking_id = args.external_id if args.external_id else ""
 
     kusto_db = KustoConnector(args.db_name)