Add upload_coverage_files for uploading coverage reports from disk

README.md

@@ -63,6 +63,17 @@ with ProoflineClient() as client:
     )
 ```
 
+To skip reading the files yourself, pass their paths to `upload_coverage_files`. Each report is named by its path relative to `root` (the current directory by default), with forward slashes, and `tested_sha` defaults to `GITHUB_SHA` when it is not passed. A missing or malformed commit raises `ValueError` before any request is sent.
+
+```
+with ProoflineClient() as client:
+    report = client.upload_coverage_files(
+        UUID(os.environ["PROOFLINE_ORGANIZATION_ID"]),
+        UUID(os.environ["PROOFLINE_REPOSITORY_ID"]),
+        ["python/lcov.info", "web/lcov.info"],
+    )
+```
+
 The response names the recorded report and its covered and instrumented line counts. An identical upload returns that report with `created=False`. Refused reports raise `ProoflineError`. Uploading a report does not establish a check verdict.
 
 ## Knowledge and plans

proofline/client.py

@@ -4,7 +4,8 @@
 
 import math
 from asyncio import sleep as async_sleep
-from collections.abc import AsyncIterator, Awaitable, Callable, Iterator
+from asyncio import to_thread
+from collections.abc import AsyncIterator, Awaitable, Callable, Iterable, Iterator
 from time import monotonic, sleep
 from types import TracebackType
 from typing import Any, Self, TypedDict, TypeVar, Unpack, cast
@@ -70,6 +71,7 @@
 from ._generated.models.stop_response import StopResponse
 from ._generated.types import UNSET, Response, Unset
 from ._transport import ProoflineError
+from .coverage import PathInput, coverage_request
 from .exports import ExportDataset, ExportFormat, ExportPage, decode_export
 from .fixes import AsyncFixesClient, FixesClient
 from .knowledge import AsyncKnowledgeClient, KnowledgeClient
@@ -412,6 +414,19 @@ def upload_coverage_reports(
             body=body,
         )
 
+    def upload_coverage_files(
+        self,
+        organization_id: UUID,
+        repository_id: UUID,
+        paths: Iterable[PathInput],
+        *,
+        tested_sha: str | None = None,
+        root: PathInput | None = None,
+    ) -> CoverageUploadResponse:
+        """Upload coverage report files; ``tested_sha`` defaults to ``GITHUB_SHA``."""
+        body = coverage_request(paths, tested_sha=tested_sha, root=root)
+        return self.upload_coverage_reports(organization_id, repository_id, body=body)
+
     def request_file_review(
         self, organization_id: UUID, repository_id: UUID, *, body: FileReviewRequest
     ) -> FileReviewResponse:
@@ -761,6 +776,19 @@ async def upload_coverage_reports(
             body=body,
         )
 
+    async def upload_coverage_files(
+        self,
+        organization_id: UUID,
+        repository_id: UUID,
+        paths: Iterable[PathInput],
+        *,
+        tested_sha: str | None = None,
+        root: PathInput | None = None,
+    ) -> CoverageUploadResponse:
+        """Upload coverage report files; ``tested_sha`` defaults to ``GITHUB_SHA``."""
+        body = await to_thread(coverage_request, paths, tested_sha=tested_sha, root=root)
+        return await self.upload_coverage_reports(organization_id, repository_id, body=body)
+
     async def request_file_review(
         self, organization_id: UUID, repository_id: UUID, *, body: FileReviewRequest
     ) -> FileReviewResponse:

proofline/coverage.py

@@ -0,0 +1,63 @@
+"""Build coverage upload requests from report files on disk."""
+
+from __future__ import annotations
+
+import os
+import re
+from collections.abc import Iterable
+from pathlib import Path
+
+from ._generated.models.coverage_upload_request import CoverageUploadRequest
+from ._generated.models.uploaded_report import UploadedReport
+
+PathInput = str | os.PathLike[str]
+
+MAX_REPORTS = 50
+_FULL_SHA = re.compile(r"[0-9a-f]{40}")
+
+
+def tested_sha_value(tested_sha: str | None) -> str:
+    """Return the full commit SHA, falling back to ``GITHUB_SHA`` from CI."""
+    value = tested_sha if tested_sha is not None else os.environ.get("GITHUB_SHA")
+    if value is None:
+        raise ValueError(
+            "Pass tested_sha or set GITHUB_SHA to the full 40-character commit the tests ran on."
+        )
+    value = value.strip()
+    if not _FULL_SHA.fullmatch(value):
+        raise ValueError("The tested commit must be a full 40-character lowercase hex SHA.")
+    return value
+
+
+def coverage_request(
+    paths: Iterable[PathInput],
+    *,
+    tested_sha: str | None = None,
+    root: PathInput | None = None,
+) -> CoverageUploadRequest:
+    """Read report files into a coverage upload request.
+
+    Each report is named by its path relative to ``root`` (the current directory by
+    default), written with forward slashes, so ``web/lcov.info`` is the same name on
+    every platform. Reports must sit inside ``root`` and must have distinct names.
+    """
+    if isinstance(paths, (str, os.PathLike)):
+        raise TypeError("paths must be a sequence of report paths, not a single path.")
+    sha = tested_sha_value(tested_sha)
+    base = Path(root if root is not None else Path.cwd()).resolve()
+    files = [Path(path) for path in paths]
+    if not 1 <= len(files) <= MAX_REPORTS:
+        raise ValueError(f"Upload 1 to {MAX_REPORTS} coverage reports per request.")
+    reports: list[UploadedReport] = []
+    seen: set[str] = set()
+    for file in files:
+        resolved = (base / file).resolve()
+        try:
+            name = resolved.relative_to(base).as_posix()
+        except ValueError:
+            raise ValueError(f"{file} is outside the report root {base}.") from None
+        if name in seen:
+            raise ValueError(f"The coverage report name {name} appears more than once.")
+        seen.add(name)
+        reports.append(UploadedReport(name=name, content=resolved.read_text(encoding="utf-8")))
+    return CoverageUploadRequest(tested_sha=sha, reports=reports)

tests/test_coverage_files.py

@@ -0,0 +1,154 @@
+from __future__ import annotations
+
+import json
+from collections.abc import AsyncIterator
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import Any
+
+import pytest
+from pydantic import SecretStr
+
+from proofline import AsyncProoflineClient, CoverageUploadResponse, ProoflineClient
+from proofline.coverage import MAX_REPORTS, coverage_request
+
+from .fixtures import ORG, REPOSITORY, REVIEW, TOKEN
+from .http_server import Reply, Request, Server
+
+SHA = "a" * 40
+LCOV = "SF:src/main.py\nDA:1,1\nDA:2,0\nend_of_record\n"
+MODES = pytest.mark.parametrize("mode", ["sync", "async"])
+
+
+@pytest.fixture(autouse=True)
+def no_ci_sha(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.delenv("GITHUB_SHA", raising=False)
+
+
+@asynccontextmanager
+async def session(mode: str, url: str) -> AsyncIterator[Any]:
+    if mode == "sync":
+        with ProoflineClient(SecretStr(TOKEN), base_url=url) as client:
+            yield client
+    else:
+        async with AsyncProoflineClient(SecretStr(TOKEN), base_url=url) as client:
+            yield client
+
+
+def write(root: Path, name: str, content: str = LCOV) -> Path:
+    path = root / name
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
+    return path
+
+
+def test_names_are_relative_to_root_with_forward_slashes(tmp_path: Path) -> None:
+    web = write(tmp_path, "web/lcov.info", "web report")
+    write(tmp_path, "python/lcov.info", "python report")
+
+    request = coverage_request([web, Path("python/lcov.info")], tested_sha=SHA, root=tmp_path)
+
+    assert request.tested_sha == SHA
+    assert [(r.name, r.content) for r in request.reports] == [
+        ("web/lcov.info", "web report"),
+        ("python/lcov.info", "python report"),
+    ]
+
+
+def test_root_defaults_to_the_current_directory(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    write(tmp_path, "lcov.info")
+    monkeypatch.chdir(tmp_path)
+
+    request = coverage_request(["lcov.info"], tested_sha=SHA)
+
+    assert [r.name for r in request.reports] == ["lcov.info"]
+
+
+def test_tested_sha_defaults_to_github_sha(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    write(tmp_path, "lcov.info")
+    monkeypatch.setenv("GITHUB_SHA", "b" * 40)
+
+    assert coverage_request(["lcov.info"], root=tmp_path).tested_sha == "b" * 40
+    assert coverage_request(["lcov.info"], tested_sha=SHA, root=tmp_path).tested_sha == SHA
+
+
+@pytest.mark.parametrize("sha", [None, "", "abc123", "A" * 40, "a" * 41])
+def test_missing_or_malformed_commit_is_refused(tmp_path: Path, sha: str | None) -> None:
+    write(tmp_path, "lcov.info")
+
+    with pytest.raises(ValueError, match="40-character"):
+        coverage_request(["lcov.info"], tested_sha=sha, root=tmp_path)
+
+
+def test_unusable_report_sets_are_refused(tmp_path: Path) -> None:
+    write(tmp_path, "lcov.info")
+    outside = write(tmp_path.parent, f"{tmp_path.name}-outside.info")
+
+    with pytest.raises(ValueError, match="outside the report root"):
+        coverage_request([outside], tested_sha=SHA, root=tmp_path)
+    with pytest.raises(ValueError, match="more than once"):
+        coverage_request(["lcov.info", "./lcov.info"], tested_sha=SHA, root=tmp_path)
+    with pytest.raises(ValueError, match="1 to 50"):
+        coverage_request([], tested_sha=SHA, root=tmp_path)
+    with pytest.raises(ValueError, match="1 to 50"):
+        names = [f"r{i}.info" for i in range(MAX_REPORTS + 1)]
+        for name in names:
+            write(tmp_path, name)
+        coverage_request(names, tested_sha=SHA, root=tmp_path)
+    with pytest.raises(TypeError, match="sequence"):
+        coverage_request("lcov.info", tested_sha=SHA, root=tmp_path)
+    with pytest.raises(FileNotFoundError):
+        coverage_request(["missing.info"], tested_sha=SHA, root=tmp_path)
+
+
+@MODES
+async def test_upload_coverage_files_posts_the_files_for_the_commit(
+    mode: str, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    write(tmp_path, "python/lcov.info")
+    monkeypatch.setenv("GITHUB_SHA", SHA)
+
+    def respond(request: Request, index: int) -> Reply:
+        assert request.method == "POST"
+        assert request.path == (
+            f"/api/v1/organizations/{ORG}/repositories/{REPOSITORY}/coverage-reports"
+        )
+        assert json.loads(request.body) == {
+            "tested_sha": SHA,
+            "reports": [{"name": "python/lcov.info", "content": LCOV}],
+        }
+        return Reply(
+            status=201,
+            body={
+                "report_id": str(REVIEW),
+                "tested_sha": SHA,
+                "created": True,
+                "formats": ["LCOV"],
+                "instrumented_lines": 2,
+                "covered_lines": 1,
+            },
+        )
+
+    with Server(respond) as server:
+        async with session(mode, server.url) as client:
+            result = client.upload_coverage_files(
+                ORG, REPOSITORY, [Path("python/lcov.info")], root=tmp_path
+            )
+            if mode == "async":
+                result = await result
+        assert isinstance(result, CoverageUploadResponse)
+        assert (result.created, result.covered_lines, result.instrumented_lines) == (True, 1, 2)
+        assert len(server.requests) == 1
+
+
+@MODES
+async def test_invalid_input_sends_no_request(mode: str, tmp_path: Path) -> None:
+    with Server(lambda request, index: Reply(status=500, body={})) as server:
+        async with session(mode, server.url) as client:
+            with pytest.raises(ValueError, match="40-character"):
+                result = client.upload_coverage_files(ORG, REPOSITORY, ["lcov.info"])
+                if mode == "async":
+                    await result
+        assert server.requests == []