Add ability to merge coverage reports

cardano_node_tests/chang_us_coverage.py

@@ -36,11 +36,11 @@ def get_args() -> argparse.Namespace:
 
 
 def _get_color(status: str) -> str:
-    if status == requirements.Statuses.SUCCESS:
+    if status == requirements.Statuses.success.name:
         return "green"
-    if status == requirements.Statuses.FAILURE:
+    if status == requirements.Statuses.failure.name:
         return "red"
-    if status == requirements.Statuses.PARTIAL_SUCCESS:
+    if status == requirements.Statuses.partial_success.name:
         return "yellow"
     return "grey"
 
@@ -62,7 +62,7 @@ def main() -> None:
 
     for req_id, req_data in chang_group.items():
         # Partial or uncovered requirements should be ignored
-        if req_id.startswith("int") or req_data["status"] == requirements.Statuses.UNCOVERED:
+        if req_id.startswith("int") or req_data["status"] == requirements.Statuses.uncovered.name:
             continue
 
         color = _get_color(req_data["status"])

cardano_node_tests/dump_requirements_coverage.py

@@ -2,6 +2,7 @@
 """Generate coverage results for external requirements."""
 
 import argparse
+import json
 import logging
 
 from cardano_node_tests.utils import helpers
@@ -13,13 +14,6 @@
 def get_args() -> argparse.Namespace:
     """Get command line arguments."""
     parser = argparse.ArgumentParser(description=__doc__.split("\n", maxsplit=1)[0])
-    parser.add_argument(
-        "-a",
-        "--artifacts-base-dir",
-        required=True,
-        type=helpers.check_dir_arg,
-        help="Path to a directory with testing artifacts",
-    )
     parser.add_argument(
         "-m",
         "--requirements-mapping",
@@ -32,6 +26,19 @@ def get_args() -> argparse.Namespace:
         required=True,
         help="File where to save coverage results",
     )
+    parser.add_argument(
+        "-a",
+        "--artifacts-base-dir",
+        type=helpers.check_dir_arg,
+        help="Path to a directory with testing artifacts",
+    )
+    parser.add_argument(
+        "-i",
+        "--input-files",
+        nargs="+",
+        type=helpers.check_file_arg,
+        help="Path to coverage results to merge into a final result",
+    )
     return parser.parse_args()
 
 
@@ -42,9 +49,24 @@ def main() -> None:
     )
     args = get_args()
 
-    executed_req = requirements.collect_executed_req(base_dir=args.artifacts_base_dir)
+    if not (args.artifacts_base_dir or args.input_files):
+        LOGGER.error("Either `--artifacts-base-dir` or `--input-files` must be provided")
+        return
+
+    executed_req = {}
+    if args.artifacts_base_dir:
+        executed_req = requirements.collect_executed_req(base_dir=args.artifacts_base_dir)
+
+    input_reqs = []
+    if args.input_files:
+        for input_file in args.input_files:
+            with open(input_file, encoding="utf-8") as in_fp:
+                input_reqs.append(json.load(in_fp))
+
+    merged_reqs = requirements.merge_reqs(executed_req, *input_reqs)
+
     report = requirements.get_mapped_req(
-        mapping=args.requirements_mapping, executed_req=executed_req
+        mapping=args.requirements_mapping, executed_req=merged_reqs
     )
 
     helpers.write_json(out_file=args.output_file, content=report)

cardano_node_tests/utils/requirements.py

@@ -1,5 +1,6 @@
 """Functionality for tracking execution of external requirements."""
 
+import enum
 import json
 import logging
 import pathlib as pl
@@ -14,11 +15,11 @@ class GroupsKnown:
     CHANG_US: tp.Final[str] = "chang_us"
 
 
-class Statuses:
-    SUCCESS: tp.Final[str] = "success"
-    FAILURE: tp.Final[str] = "failure"
-    UNCOVERED: tp.Final[str] = "uncovered"
-    PARTIAL_SUCCESS: tp.Final[str] = "partial_success"
+class Statuses(enum.Enum):
+    success = 1
+    failure = 2
+    partial_success = 3
+    uncovered = 4
 
 
 class Req:
@@ -48,14 +49,24 @@ def _get_dest_dir(self) -> pl.Path:
         return dest_dir
 
     def success(self) -> bool:
-        content = {"id": self.id, "group": self.group, "url": self.url, "status": Statuses.SUCCESS}
+        content = {
+            "id": self.id,
+            "group": self.group,
+            "url": self.url,
+            "status": Statuses.success.name,
+        }
         helpers.write_json(
             out_file=self._get_dest_dir() / f"{self.basename}_success.json", content=content
         )
         return True
 
     def failure(self) -> bool:
-        content = {"id": self.id, "group": self.group, "url": self.url, "status": Statuses.FAILURE}
+        content = {
+            "id": self.id,
+            "group": self.group,
+            "url": self.url,
+            "status": Statuses.failure.name,
+        }
         helpers.write_json(
             out_file=self._get_dest_dir() / f"{self.basename}_init.json", content=content
         )
@@ -86,7 +97,7 @@ def collect_executed_req(base_dir: pl.Path) -> dict:
 
         req_id = req_rec["id"]
         id_collected = group_collected.get(req_id)
-        if id_collected and id_collected["status"] == Statuses.SUCCESS:
+        if id_collected and id_collected["status"] == Statuses.success.name:
             continue
         if not id_collected:
             id_collected = {}
@@ -97,6 +108,22 @@ def collect_executed_req(base_dir: pl.Path) -> dict:
     return collected
 
 
+def merge_reqs(*reqs: tp.Dict[str, dict]) -> dict:
+    """Merge requirements."""
+    merged: tp.Dict[str, dict] = {}
+    for report in reqs:
+        for gname, greqs in report.items():
+            merged_group = merged.get(gname) or {}
+            for req_id, req_data in greqs.items():
+                merged_rec = merged_group.get(req_id) or {}
+                merged_status_val = Statuses[merged_rec.get("status") or "uncovered"].value
+                req_status_val = Statuses[req_data["status"]].value
+                if not merged_rec or req_status_val < merged_status_val:
+                    merged_group[req_id] = req_data
+            merged[gname] = merged_group
+    return merged
+
+
 def get_mapped_req(mapping: pl.Path, executed_req: dict) -> dict:
     """Get mapped requirements."""
     with open(mapping, encoding="utf-8") as in_fp:
@@ -116,22 +143,22 @@ def get_mapped_req(mapping: pl.Path, executed_req: dict) -> dict:
                 if not url:
                     url = executed_group.get(p_req, {}).get("url")
 
-                if p_status == Statuses.SUCCESS:
+                if p_status == Statuses.success.name:
                     dependencies_success.append(p_req)
-                elif p_status == Statuses.FAILURE:
+                elif p_status == Statuses.failure.name:
                     dependencies_failures.append(p_req)
 
             # If any partial requirement failed, the overall outcome would be failed
             if dependencies_failures:
-                status = Statuses.FAILURE
+                status = Statuses.failure.name
             # If none partial requirement is covered, the overall outcome would be uncovered
             elif not (dependencies_success or dependencies_failures):
-                status = Statuses.UNCOVERED
+                status = Statuses.uncovered.name
             # If all partial requirements are successful, the overall outcome would be success
             elif len(dependencies_success) == len(dependencies):
-                status = Statuses.SUCCESS
+                status = Statuses.success.name
             else:
-                status = Statuses.PARTIAL_SUCCESS
+                status = Statuses.partial_success.name
 
             executed_req[group][req_id] = {"status": status, "url": url}