enable code coverage for opa

Author: ffortier

MODULE.bazel.lock

@@ -14,7 +14,7 @@ def _opa_check_test_impl(ctx):
     if ctx.files.schema_files:
         files.extend(ctx.files.schema_files)
 
-    args = ["set -xe\n"]
+    args = ["set -e\n"]
 
     args.append(toolchain.opa.short_path)
     args.append("check")

opa/private/opa_check.bzl

@@ -16,9 +16,15 @@ def _opa_test_impl(ctx):
     if ctx.attr.verbose:
         args.append("--verbose")
 
+    tool_path = toolchain.opa.short_path
+
+    if ctx.configuration.coverage_enabled:
+        tool_path = "%s %s %s" % (ctx.executable._opa_coverage.short_path, str(ctx.label), tool_path)
+        runfiles = runfiles.merge(ctx.attr._opa_coverage[DefaultInfo].default_runfiles)
+
     ctx.actions.write(
         output = test_file,
-        content = "%s test %s %s" % (toolchain.opa.short_path, bundle.short_path, " ".join(args + [f.short_path for f in ctx.files.srcs])),
+        content = "%s test %s %s" % (tool_path, bundle.short_path, " ".join(args + [f.short_path for f in ctx.files.srcs])),
         is_executable = True,
     )
 
@@ -27,6 +33,7 @@ def _opa_test_impl(ctx):
             executable = test_file,
             runfiles = runfiles,
         ),
+        coverage_common.instrumented_files_info(ctx, source_attributes = ["srcs"], dependency_attributes = ["deps"]),
     ]
 
 opa_test = rule(
@@ -51,6 +58,7 @@ opa_test = rule(
             doc = "set verbose reporting mode",
             default = False,
         ),
+        "_opa_coverage": attr.label(default = "//tools:opa_coverage", executable = True, cfg = "exec"),
     },
     toolchains = ["//tools:toolchain_type"],
 )

opa/private/opa_test.bzl

@@ -18,6 +18,12 @@ py_binary(
     srcs = ["opa_upgrade.py"],
 )
 
+py_binary(
+    name = "opa_coverage",
+    srcs = ["opa_coverage.py"],
+    visibility = ["//visibility:public"],
+)
+
 toolchain_type(name = "toolchain_type")
 
 opa_toolchain(

tools/BUILD.bazel

@@ -0,0 +1,109 @@
+from typing import List, Dict, Any, cast
+from subprocess import run, PIPE
+from dataclasses import dataclass
+from os import environ
+
+import json
+import sys
+
+
+@dataclass
+class CoverageRangeContent:
+    row: int
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(row=data.get("data") or 0)
+
+
+@dataclass
+class CoverageRange:
+    start: CoverageRangeContent
+    end: CoverageRangeContent
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            start=CoverageRangeContent.from_dict(data.get("start")),
+            end=CoverageRangeContent.from_dict(data.get("end")),
+        )
+
+
+@dataclass
+class CoverageFile:
+    covered: List[CoverageRange]
+    not_covered: List[CoverageRange]
+    covered_lines: int
+    not_covered_lines: int
+    coverage: float
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            covered=[CoverageRange.from_dict(r) for r in data.get("covered") or []],
+            not_covered=[
+                CoverageRange.from_dict(r) for r in data.get("not_covered") or []
+            ],
+            covered_lines=int(data.get("covered_lines") or 0),
+            not_covered_lines=int(data.get("not_covered_lines") or 0),
+            coverage=float(data.get("coverage")),
+        )
+
+
+@dataclass
+class CoverageReport:
+    files: Dict[str, CoverageFile]
+    covered_lines: int
+    not_covered_lines: int
+    coverage: float
+
+    @classmethod
+    def from_dict(cls, data):
+        return cls(
+            files={
+                file: CoverageFile.from_dict(report)
+                for file, report in data.get("files").items()
+            },
+            covered_lines=int(data.get("covered_lines")),
+            not_covered_lines=int(data.get("not_covered_lines")),
+            coverage=float(data.get("coverage")),
+        )
+
+
+def main(argv: List[str]):
+    _program, test_name, opa, cmd, *args = argv
+
+    coverage_enabled = bool(environ.get("COVERAGE", 0))
+    coverage_output_file = environ.get("COVERAGE_OUTPUT_FILE")
+
+    assert coverage_enabled
+    assert coverage_output_file is not None
+
+    proc = run([opa, cmd, "--format=json", "--coverage"] + args, stdout=PIPE)
+
+    if proc.returncode != 0:
+        exit(proc.returncode)
+
+    result = CoverageReport.from_dict(json.loads(proc.stdout))
+
+    with open(coverage_output_file, "w") as f:
+        for file, report in result.files.items():
+            if file in args:
+                continue  # Skip test sources
+            print(f"TN:{test_name.removeprefix('@@')}", file=f)
+            print(f"SF:{file}", file=f)
+            print(f"FNF:0", file=f)
+            print(f"FNH:0", file=f)
+
+            for r in report.covered:
+                for i in range(r.start.row, r.end.row + 1):
+                    print(f"DA:{i},1", file=f)
+
+            print(f"LF:{report.covered_lines+report.not_covered_lines}", file=f)
+            print(f"LH:{report.covered_lines}", file=f)
+            print("end_of_record", file=f)
+            print(file=f)
+
+
+if __name__ == "__main__":
+    main(sys.argv)