diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
new file mode 100644
index 0000000..35fd633
--- /dev/null
+++ b/.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
diff --git a/.gitignore b/.gitignore
new file mode 100644
index 0000000..c18dd8d
--- /dev/null
+++ b/.gitignore
@@ -0,0 +1 @@
+__pycache__/
diff --git a/action.yml b/action.yml
index de49b1a..8164026 100644
--- a/action.yml
+++ b/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"
diff --git a/test_upload_coverage.py b/test_upload_coverage.py
new file mode 100644
index 0000000..2a233cd
--- /dev/null
+++ b/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()
diff --git a/upload_coverage.py b/upload_coverage.py
new file mode 100755
index 0000000..2858b61
--- /dev/null
+++ b/upload_coverage.py
@@ -0,0 +1,122 @@
+#!/usr/bin/env python3
+import base64
+import gzip
+import json
+import os
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Mapping, Optional
+
+
+PERMISSIONS_ERROR = (
+    "Coverage upload returned 403 Forbidden. Ensure the calling job has "
+    "'code-quality: write' permission. See https://github.com/actions/upload-code-coverage#permissions"
+)
+
+
+def emit_annotation(level: str, message: str) -> None:
+    print(f"::{level}::{message}")
+
+
+def encode_coverage_report(file_path: str) -> str:
+    data = Path(file_path).read_bytes()
+    return base64.b64encode(gzip.compress(data)).decode("ascii")
+
+
+def build_payload(
+    *,
+    file_path: str,
+    language: str,
+    label: str,
+    commit_oid: str,
+    ref: str = "",
+    pr_number: str = "",
+) -> dict:
+    payload = {
+        "commit_oid": commit_oid,
+        "coverage_report": encode_coverage_report(file_path),
+        "language_name": language,
+        "label": label,
+    }
+
+    if pr_number:
+        payload["pull_request_number"] = int(pr_number)
+    elif ref:
+        payload["ref"] = ref
+    else:
+        raise ValueError("Either PR_NUMBER or REF must be provided")
+
+    return payload
+
+
+def upload_report(
+    *,
+    payload: dict,
+    repository: str,
+    api_url: str,
+    token: str,
+    opener=urllib.request.urlopen,
+) -> int:
+    request = urllib.request.Request(
+        url=f"{api_url.rstrip('/')}/repos/{repository}/code-coverage/report",
+        data=json.dumps(payload).encode("utf-8"),
+        headers={
+            "Authorization": f"Bearer {token}",
+            "Accept": "application/vnd.github+json",
+            "Content-Type": "application/json",
+        },
+        method="PUT",
+    )
+
+    try:
+        with opener(request) as response:
+            response.read()
+            return response.getcode()
+    except urllib.error.HTTPError as error:
+        body = error.read().decode("utf-8", errors="replace")
+        if error.code == 403 and "not authorized" in body.lower():
+            emit_annotation("error", PERMISSIONS_ERROR)
+        else:
+            emit_annotation("error", f"Coverage upload failed with status {error.code}: {body}")
+        return error.code
+    except urllib.error.URLError as error:
+        emit_annotation("error", f"Coverage upload failed: {error.reason}")
+        return 1
+
+
+def main(environ: Optional[Mapping[str, str]] = None, opener=urllib.request.urlopen) -> int:
+    env = dict(os.environ if environ is None else environ)
+
+    file_path = env.get("INPUT_FILE", "")
+    if not file_path or not Path(file_path).is_file():
+        emit_annotation("error", f"Coverage file not found: {file_path}")
+        return 1
+
+    try:
+        payload = build_payload(
+            file_path=file_path,
+            language=env.get("INPUT_LANGUAGE", ""),
+            label=env.get("INPUT_LABEL", ""),
+            commit_oid=env.get("COMMIT_OID", ""),
+            ref=env.get("REF", ""),
+            pr_number=env.get("PR_NUMBER", ""),
+        )
+    except ValueError as error:
+        emit_annotation("error", str(error))
+        return 1
+
+    status = upload_report(
+        payload=payload,
+        repository=env.get("GITHUB_REPOSITORY", ""),
+        api_url=env.get("GITHUB_API_URL", "https://api.github.com"),
+        token=env.get("GH_TOKEN", ""),
+        opener=opener,
+    )
+
+    return 0 if 200 <= status < 300 else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())