Refactor upload logic into testable Python script

.github/workflows/ci.yml

@@ -0,0 +1,15 @@
+name: CI
+
+on:
+  push:
+    branches: [main]
+  pull_request:
+
+jobs:
+  test:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+
+      - name: Run unit tests
+        run: python3 -m unittest test_upload_coverage.py -v

.gitignore

@@ -0,0 +1 @@
+__pycache__/

action.yml

@@ -26,88 +26,41 @@ runs:
         INPUT_FILE: ${{ inputs.file }}
         INPUT_LANGUAGE: ${{ inputs.language }}
         INPUT_LABEL: ${{ inputs.label }}
+        GITHUB_EVENT_NAME: ${{ github.event_name }}
+        GITHUB_REPOSITORY: ${{ github.repository }}
+        GITHUB_API_URL: ${{ github.api_url }}
       run: |
         set -euo pipefail
-
-        # Support GitHub Enterprise Cloud when using gh CLI
-        export GH_HOST="${GITHUB_SERVER_URL#*://}"
 
-        if [ ! -f "$INPUT_FILE" ]; then
-          echo "::error::Coverage file not found: $INPUT_FILE"
-          exit 1
-        fi
+        export GH_HOST="${GITHUB_SERVER_URL#*://}"
 
-        # Skip coverage upload for merge queue runs — coverage should be
-        # uploaded for PRs and the default branch, making merge queue
-        # uploads unnecessary.
-        if [ "${{ github.event_name }}" = "merge_group" ]; then
+        if [ "$GITHUB_EVENT_NAME" = "merge_group" ]; then
           echo "::warning::Skipping coverage upload for merge queue. Configure your workflow to upload coverage for PRs and the default branch instead. To avoid spinning up a runner, add \"if: github.event_name != 'merge_group'\" to the upload job."
           exit 0
         fi
 
-        # Skip coverage upload for fork PRs — the token won't have write
-        # permissions to the base repository.
         if [ "${{ github.event.pull_request.head.repo.full_name }}" != "" ] && \
-           [ "${{ github.event.pull_request.head.repo.full_name }}" != "${{ github.repository }}" ]; then
+           [ "${{ github.event.pull_request.head.repo.full_name }}" != "$GITHUB_REPOSITORY" ]; then
           echo "::notice::Skipping coverage upload for fork PR (from ${{ github.event.pull_request.head.repo.full_name }})"
           exit 0
         fi
 
-        # Resolve the commit SHA and ref. On pull_request events, github.sha
-        # and github.ref point to the merge commit — use the PR head instead.
-        if [ "${{ github.event_name }}" = "pull_request" ] || [ "${{ github.event_name }}" = "pull_request_target" ]; then
+        if [ "$GITHUB_EVENT_NAME" = "pull_request" ] || [ "$GITHUB_EVENT_NAME" = "pull_request_target" ]; then
           COMMIT_OID="${{ github.event.pull_request.head.sha }}"
-          REF="refs/heads/${{ github.event.pull_request.head.ref }}"
+          REF=""
           PR_NUMBER="${{ github.event.pull_request.number }}"
         else
           COMMIT_OID="${{ github.sha }}"
           REF="${{ github.ref }}"
-          # For push events, check if this branch has an open PR.
           PR_NUMBER=$(gh pr list \
-            --repo "${{ github.repository }}" \
+            --repo "$GITHUB_REPOSITORY" \
             --head "${{ github.ref_name }}" \
             --state open \
             --json number \
             --jq '.[0].number // empty' 2>/dev/null || true)
         fi
 
-        # Gzip and base64-encode the report. We write to files and use jq
-        # --rawfile to avoid hitting the OS argument length limit on large
-        # coverage reports.
-        gzip -c "$INPUT_FILE" | base64 -w 0 > __coverage_b64.txt
-
-        # The API requires either ref or pull_request_number (not both).
-        if [ -n "${PR_NUMBER:-}" ]; then
-          # pull-request-based upload: provide pull_request_number, omit ref
-          jq -n \
-            --arg commit_oid "$COMMIT_OID" \
-            --rawfile coverage_report __coverage_b64.txt \
-            --arg language_name "$INPUT_LANGUAGE" \
-            --arg report_label "$INPUT_LABEL" \
-            --argjson pr_number "$PR_NUMBER" \
-            '{"commit_oid": $commit_oid, "coverage_report": $coverage_report, "language_name": $language_name, "label": $report_label, "pull_request_number": $pr_number}' \
-            > __body.json
-        else
-          # ref-based upload: provide ref, omit pull_request_number
-          jq -n \
-            --arg commit_oid "$COMMIT_OID" \
-            --arg ref "$REF" \
-            --rawfile coverage_report __coverage_b64.txt \
-            --arg language_name "$INPUT_LANGUAGE" \
-            --arg report_label "$INPUT_LABEL" \
-            '{"commit_oid": $commit_oid, "ref": $ref, "coverage_report": $coverage_report, "language_name": $language_name, "label": $report_label}' \
-            > __body.json
-        fi
-
-        UPLOAD_OUTPUT=$(gh api --method PUT "/repos/${{ github.repository }}/code-coverage/report" \
-          --input __body.json 2>&1) || {
-          if echo "$UPLOAD_OUTPUT" | grep -qi "not authorized"; then
-            echo "::error::Coverage upload returned 403 Forbidden. Ensure the calling job has 'code-quality: write' permission. See https://github.com/actions/upload-code-coverage#permissions"
-          else
-            echo "::error::Coverage upload failed: $UPLOAD_OUTPUT"
-          fi
-          rm -f __coverage_b64.txt __body.json
-          exit 1
-        }
-
-        rm -f __coverage_b64.txt __body.json
+        COMMIT_OID="$COMMIT_OID" \
+        REF="$REF" \
+        PR_NUMBER="$PR_NUMBER" \
+        python3 "$GITHUB_ACTION_PATH/upload_coverage.py"

test_upload_coverage.py

@@ -0,0 +1,190 @@
+import base64
+import gzip
+import io
+import json
+import shutil
+import unittest
+from contextlib import redirect_stdout
+from pathlib import Path
+from tempfile import mkdtemp
+from urllib.error import HTTPError, URLError
+from unittest import mock
+
+import upload_coverage
+
+
+class FakeResponse:
+    def __init__(self, status=201, body=b""):
+        self.status = status
+        self.body = body
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc, tb):
+        return False
+
+    def read(self):
+        return self.body
+
+    def getcode(self):
+        return self.status
+
+
+class UploadCoverageTests(unittest.TestCase):
+    def setUp(self):
+        self.test_dir = Path(mkdtemp(dir=Path(__file__).parent))
+        self.coverage_file = self.test_dir / "coverage.xml"
+        self.coverage_contents = b"<coverage branch-rate=\"0.5\" />\n"
+        self.coverage_file.write_bytes(self.coverage_contents)
+        self.base_env = {
+            "INPUT_FILE": str(self.coverage_file),
+            "INPUT_LANGUAGE": "Python",
+            "INPUT_LABEL": "code-coverage/test",
+            "COMMIT_OID": "deadbeef",
+            "REF": "refs/heads/main",
+            "PR_NUMBER": "",
+            "GITHUB_REPOSITORY": "octo-org/octo-repo",
+            "GITHUB_API_URL": "https://api.github.com",
+            "GH_TOKEN": "test-token",
+        }
+
+    def tearDown(self):
+        shutil.rmtree(self.test_dir)
+
+    def run_main(self, env=None, opener=None):
+        stdout = io.StringIO()
+        opener = opener or mock.Mock(return_value=FakeResponse())
+        with redirect_stdout(stdout):
+            exit_code = upload_coverage.main(environ=env or self.base_env, opener=opener)
+        return exit_code, stdout.getvalue(), opener
+
+    def request_payload(self, opener):
+        request = opener.call_args.args[0]
+        return json.loads(request.data.decode("utf-8"))
+
+    def test_file_not_found_exits_with_error_annotation(self):
+        env = dict(self.base_env, INPUT_FILE=str(self.test_dir / "missing.xml"))
+
+        exit_code, output, opener = self.run_main(env=env)
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("::error::Coverage file not found", output)
+        opener.assert_not_called()
+
+    def test_successful_upload_exits_zero(self):
+        exit_code, output, opener = self.run_main()
+
+        self.assertEqual(0, exit_code)
+        self.assertEqual("", output)
+        opener.assert_called_once()
+
+    def test_successful_with_pr_number_uses_pull_request_number(self):
+        env = dict(self.base_env, REF="", PR_NUMBER="42")
+        exit_code, _, opener = self.run_main(env=env)
+
+        payload = self.request_payload(opener)
+        self.assertEqual(0, exit_code)
+        self.assertEqual(42, payload["pull_request_number"])
+        self.assertNotIn("ref", payload)
+
+    def test_successful_with_ref_uses_ref_and_omits_pull_request_number(self):
+        exit_code, _, opener = self.run_main()
+
+        payload = self.request_payload(opener)
+        self.assertEqual("refs/heads/main", payload["ref"])
+        self.assertNotIn("pull_request_number", payload)
+
+    def test_403_not_authorized_exits_with_permissions_error(self):
+        error = HTTPError(
+            url="https://api.github.com/repos/octo-org/octo-repo/code-coverage/report",
+            code=403,
+            msg="Forbidden",
+            hdrs=None,
+            fp=io.BytesIO(b'{"message":"not authorized"}'),
+        )
+
+        exit_code, output, _ = self.run_main(opener=mock.Mock(side_effect=error))
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("code-quality: write", output)
+
+    def test_403_other_message_exits_with_generic_error(self):
+        error = HTTPError(
+            url="https://api.github.com/repos/octo-org/octo-repo/code-coverage/report",
+            code=403,
+            msg="Forbidden",
+            hdrs=None,
+            fp=io.BytesIO(b'{"message":"forbidden"}'),
+        )
+
+        exit_code, output, _ = self.run_main(opener=mock.Mock(side_effect=error))
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("Coverage upload failed with status 403", output)
+        self.assertIn('forbidden', output)
+
+    def test_400_bad_request_exits_with_status_and_body(self):
+        error = HTTPError(
+            url="https://api.github.com/repos/octo-org/octo-repo/code-coverage/report",
+            code=400,
+            msg="Bad Request",
+            hdrs=None,
+            fp=io.BytesIO(b'{"message":"bad request"}'),
+        )
+
+        exit_code, output, _ = self.run_main(opener=mock.Mock(side_effect=error))
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("status 400", output)
+        self.assertIn("bad request", output)
+
+    def test_500_server_error_exits_with_status_and_body(self):
+        error = HTTPError(
+            url="https://api.github.com/repos/octo-org/octo-repo/code-coverage/report",
+            code=500,
+            msg="Server Error",
+            hdrs=None,
+            fp=io.BytesIO(b'{"message":"boom"}'),
+        )
+
+        exit_code, output, _ = self.run_main(opener=mock.Mock(side_effect=error))
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("status 500", output)
+        self.assertIn("boom", output)
+
+    def test_network_error_exits_with_error(self):
+        exit_code, output, _ = self.run_main(opener=mock.Mock(side_effect=URLError("connection refused")))
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("connection refused", output)
+
+    def test_payload_structure_and_encoding(self):
+        exit_code, _, opener = self.run_main()
+
+        request = opener.call_args.args[0]
+        payload = self.request_payload(opener)
+        decoded = gzip.decompress(base64.b64decode(payload["coverage_report"]))
+
+        self.assertEqual(0, exit_code)
+        self.assertEqual("PUT", request.get_method())
+        self.assertEqual("Bearer test-token", request.headers["Authorization"])
+        self.assertEqual("deadbeef", payload["commit_oid"])
+        self.assertEqual("Python", payload["language_name"])
+        self.assertEqual("code-coverage/test", payload["label"])
+        self.assertEqual(self.coverage_contents, decoded)
+
+    def test_neither_ref_nor_pr_number_exits_with_error(self):
+        env = dict(self.base_env, REF="", PR_NUMBER="")
+
+        exit_code, output, opener = self.run_main(env=env)
+
+        self.assertEqual(1, exit_code)
+        self.assertIn("::error::", output)
+        self.assertIn("Either PR_NUMBER or REF must be provided", output)
+        opener.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()