Add command 'record coverage'

launchable/commands/record/__init__.py

@@ -5,6 +5,7 @@
 from .attachment import attachment
 from .build import build
 from .commit import commit
+from .coverage import coverage
 from .session import session
 from .tests import tests
 
@@ -21,3 +22,4 @@ def record():
 record.add_alias('test', tests)  # type: ignore
 record.add_command(session)
 record.add_command(attachment)
+record.add_command(coverage)

launchable/commands/record/coverage.py

@@ -0,0 +1,135 @@
+import glob
+import os
+from io import BytesIO
+from pathlib import Path
+from typing import List, Optional, Tuple
+
+import click
+from tabulate import tabulate
+
+from ...utils.launchable_client import LaunchableClient
+from ..helper import require_session
+
+
+class CoverageStatus:
+    SUCCESS = "✓ Recorded successfully"
+    FAILED = "⚠ Failed to record"
+
+
+@click.command(help="Record coverage data (LCOV format) related to a test session. "
+               "Passing COVERAGE_FILES is optional for bazel; "
+               "files are auto discovered when the command is run inside project root. "
+               "Other test runners must pass their coverage report file(s) directly.")
+@click.option(
+    '--session',
+    'session',
+    help='In the format builds/<build-name>/test_sessions/<test-session-id>',
+    type=str,
+)
+@click.argument('coverage_files', nargs=-1)
+@click.pass_context
+def coverage(
+        context: click.core.Context,
+        coverage_files: Tuple[str, ...],
+        session: Optional[str] = None,
+):
+    client = LaunchableClient(app=context.obj)
+    sources: List[str] = []
+
+    try:
+        session = require_session(session)
+        assert session is not None
+
+        if coverage_files:
+            sources = list(coverage_files)
+            blocks = [_read(f) for f in sources]
+        else:
+            # No files given: auto-discover from the current directory
+            # Auto discovery is Bazel only for now.
+            sources, blocks = discover_bazel_coverage(Path("."))
+
+        # LCOV tracefiles are just a sequence of SF:...end_of_record blocks, so concatenating
+        # multiple files is itself a valid tracefile. Send them as one upload rather than one
+        # request per file. The server already unions repeated SF: blocks for the same file within
+        # one upload, so this is correct even when the inputs cover overlapping files (e.g. one
+        # report from a unit-test run, another from an integration-test run touching the same
+        # source file)
+        content = b"\n".join(blocks)
+        status, files_recorded, is_reupload = post_coverage(client, session, content)
+    except Exception as e:
+        client.print_exception_and_recover(e)
+        status, files_recorded, is_reupload = CoverageStatus.FAILED, None, False
+
+    display_summary(sources, status, files_recorded, is_reupload)
+
+
+def discover_bazel_coverage(workspace: "str | Path") -> Tuple[List[str], List[bytes]]:
+    """
+    Finds every per-target coverage.dat under <workspace>/bazel-testlogs
+
+    Bazel's own coverage.dat carries no target identity on its own, though TN (Test Name) is a standard LCOV field.
+    So we derive TN by file path (same approach used by `record tests bazel`
+    (bazel-testlogs/<pkg>/<target>/coverage.dat -> TN://<pkg>:<target>).
+    """
+    base = Path(workspace).joinpath('bazel-testlogs')
+    if not base.exists():
+        raise click.UsageError(
+            "No such directory: %s. "
+            "Pass coverage report file(s) directly as arguments instead." % str(base))
+
+    sources: List[str] = []
+    blocks: List[bytes] = []
+    for report_file in sorted(glob.iglob(os.path.join(str(base), '**', 'coverage.dat'), recursive=True)):
+        pkg_and_target = report_file[len(str(base)) + 1:-len("/coverage.dat")]
+        package, target = os.path.dirname(pkg_and_target), os.path.basename(pkg_and_target)
+        label = "//{}:{}".format(package.replace(os.sep, '/'), target)
+        sources.append(report_file)
+        blocks.append(_tag_with_test_name(_read(report_file), label))
+
+    if not blocks:
+        raise click.UsageError("No coverage.dat files found under %s" % str(base))
+
+    return sources, blocks
+
+
+def _tag_with_test_name(content: bytes, label: str) -> bytes:
+    tn_line = "TN:{}\n".format(label).encode('utf-8')
+    tagged = []
+    for line in content.splitlines(keepends=True):
+        if line.startswith(b"SF:"):
+            tagged.append(tn_line)
+        tagged.append(line)
+    return b"".join(tagged)
+
+
+def _read(path: str) -> bytes:
+    with open(path, mode='rb') as f:
+        return f.read()
+
+
+def post_coverage(client: LaunchableClient, session: str, content: bytes) -> Tuple[str, Optional[int], bool]:
+    try:
+        res = client.request(
+            "post", "{}/coverage".format(session), compress=True, payload=BytesIO(content),
+            additional_headers={"Content-Type": "text/plain"})
+        res.raise_for_status()
+        summary = res.json()
+        return CoverageStatus.SUCCESS, summary.get("sourceFilesRecorded"), summary.get("isReupload", False)
+    except Exception as e:
+        click.echo("Failed to upload coverage report: {}".format(str(e)), err=True)
+        return CoverageStatus.FAILED, None, False
+
+
+def display_summary(sources: List[str], status: str, source_files: Optional[int], is_reupload: bool = False):
+    if sources:
+        click.echo(tabulate([[f] for f in sources], ["Coverage files found"], tablefmt="github"))
+        click.echo()
+    rows = [[len(sources), source_files if source_files is not None else "-", status]]
+    click.echo(tabulate(rows, ["Coverage Files", "Source Files", "Status"], tablefmt="github"))
+    if is_reupload:
+        click.echo(
+            click.style(
+                "⚠ This session already had coverage recorded. The previous upload was overwritten. "
+                "Refer to --help if you intended to record multiple files",
+                fg="yellow"),
+            err=True)

tests/commands/record/test_coverage.py

@@ -0,0 +1,146 @@
+import gzip
+import json
+import os
+import tempfile
+from unittest import mock
+
+import responses  # type: ignore
+
+from launchable.utils.http_client import get_base_url
+from tests.cli_test_case import CliTestCase
+
+
+class CoverageTest(CliTestCase):
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_coverage(self):
+        TEST_CONTENT = b"SF:src/Foo.java\nDA:1,1\nend_of_record\n"
+
+        coverage_file = tempfile.NamedTemporaryFile(delete=False)
+        coverage_file.write(TEST_CONTENT)
+        coverage_file.close()
+
+        body = None
+
+        def verify_body(request):
+            nonlocal body
+            body = gzip.decompress(b''.join(list(request.body)))
+            return (200, [], json.dumps({"sourceFilesRecorded": 1, "filesWithData": 1}))
+
+        responses.add_callback(
+            responses.POST,
+            f"{get_base_url()}/intake/organizations/{self.organization}/workspaces/"
+            f"{self.workspace}/builds/{self.build_name}/test_sessions/{self.session_id}/coverage",
+            callback=verify_body)
+
+        result = self.cli("record", "coverage", "--session", self.session, coverage_file.name)
+        self.assert_success(result)
+        self.assertEqual(TEST_CONTENT, body)
+        self.assertIn(coverage_file.name, result.output)
+        self.assertIn("✓ Recorded successfully", result.output)
+        self.assertIn("Coverage Files", result.output)
+        self.assertIn("Source Files", result.output)
+
+        os.unlink(coverage_file.name)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_coverage_concatenates_multiple_files_into_one_request(self):
+        content_a = b"SF:src/A.java\nDA:1,1\nend_of_record\n"
+        content_b = b"SF:src/B.java\nDA:1,0\nend_of_record\n"
+
+        file_a = tempfile.NamedTemporaryFile(delete=False)
+        file_a.write(content_a)
+        file_a.close()
+        file_b = tempfile.NamedTemporaryFile(delete=False)
+        file_b.write(content_b)
+        file_b.close()
+
+        requests_received = []
+
+        def verify_body(request):
+            requests_received.append(gzip.decompress(b''.join(list(request.body))))
+            return (200, [], json.dumps({"sourceFilesRecorded": 2, "filesWithData": 1}))
+
+        responses.add_callback(
+            responses.POST,
+            f"{get_base_url()}/intake/organizations/{self.organization}/workspaces/"
+            f"{self.workspace}/builds/{self.build_name}/test_sessions/{self.session_id}/coverage",
+            callback=verify_body)
+
+        result = self.cli("record", "coverage", "--session", self.session, file_a.name, file_b.name)
+
+        self.assert_success(result)
+        # Exactly one request, not one per file, carrying both files' content concatenated.
+        self.assertEqual(1, len(requests_received))
+        self.assertEqual(content_a + b"\n" + content_b, requests_received[0])
+        self.assertIn(file_a.name, result.output)
+        self.assertIn(file_b.name, result.output)
+        self.assertIn("Coverage Files", result.output)
+        self.assertIn("Source Files", result.output)
+
+        os.unlink(file_a.name)
+        os.unlink(file_b.name)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_coverage_bazel_auto_discovery_tags_each_target_and_sends_one_request(self):
+        with tempfile.TemporaryDirectory() as workspace:
+            testlogs = os.path.join(workspace, "bazel-testlogs")
+            net_dir = os.path.join(testlogs, "net", "AllTests")
+            strings_dir = os.path.join(testlogs, "util", "strings", "AllTests")
+            os.makedirs(net_dir)
+            os.makedirs(strings_dir)
+
+            with open(os.path.join(net_dir, "coverage.dat"), "wb") as f:
+                f.write(b"SF:src/net/URIs.java\nDA:1,1\nend_of_record\n")
+            with open(os.path.join(strings_dir, "coverage.dat"), "wb") as f:
+                f.write(b"SF:src/util/strings/Strings.java\nDA:1,0\nend_of_record\n")
+
+            requests_received = []
+
+            def verify_body(request):
+                requests_received.append(gzip.decompress(b''.join(list(request.body))))
+                return (200, [], json.dumps({"sourceFilesRecorded": 2, "filesWithData": 1}))
+
+            responses.add_callback(
+                responses.POST,
+                f"{get_base_url()}/intake/organizations/{self.organization}/workspaces/"
+                f"{self.workspace}/builds/{self.build_name}/test_sessions/{self.session_id}/coverage",
+                callback=verify_body)
+
+            cwd = os.getcwd()
+            os.chdir(workspace)
+            try:
+                result = self.cli("record", "coverage", "--session", self.session)
+            finally:
+                os.chdir(cwd)
+
+            self.assert_success(result)
+            self.assertEqual(1, len(requests_received))
+            body = requests_received[0]
+
+            # Each target's SF: block is preceded by a TN: record derived from its path, since
+            # Bazel's own coverage.dat carries no target identity on its own.
+            self.assertIn(b"TN://net:AllTests\nSF:src/net/URIs.java\n", body)
+            self.assertIn(b"TN://util/strings:AllTests\nSF:src/util/strings/Strings.java\n", body)
+
+    @responses.activate
+    @mock.patch.dict(os.environ, {"LAUNCHABLE_TOKEN": CliTestCase.launchable_token})
+    def test_coverage_upload_failure(self):
+        coverage_file = tempfile.NamedTemporaryFile(delete=False)
+        coverage_file.write(b"not valid lcov")
+        coverage_file.close()
+
+        responses.add(
+            responses.POST,
+            "{}/intake/organizations/{}/workspaces/{}/builds/{}/test_sessions/{}/coverage".format(
+                get_base_url(), self.organization, self.workspace, self.build_name, self.session_id),
+            json={"error": "No SF: blocks found in coverage report"},
+            status=400)
+
+        result = self.cli("record", "coverage", "--session", self.session, coverage_file.name)
+
+        self.assertIn("⚠ Failed to record", result.output)
+
+        os.unlink(coverage_file.name)