check if the PR is a fork PR

codecov_cli/helpers/encoder.py

@@ -2,6 +2,7 @@
 
 slug_without_subgroups_regex = re.compile(r"[^/\s]+\/[^/\s]+$")
 slug_with_subgroups_regex = re.compile(r"[^/\s]+(\/[^/\s]+)+$")
+encoded_slug_regex = re.compile(r"[^:\s]+(:::[^:\s]+)*(::::[^:\s]+){1}$")
 
 
 def encode_slug(slug: str):
@@ -13,6 +14,16 @@ def encode_slug(slug: str):
     return encoded_slug
 
 
+def decode_slug(slug: str):
+    if slug_encoded_incorrectly(slug):
+        raise ValueError("The slug is not encoded correctly")
+
+    owner, repo = slug.split("::::", 1)
+    decoded_owner = "/".join(owner.split(":::"))
+    decoded_slug = "/".join([decoded_owner, repo])
+    return decoded_slug
+
+
 def slug_without_subgroups_is_invalid(slug: str):
     """
     Checks if slug is in the form of owner/repo
@@ -27,3 +38,12 @@ def slug_with_subgroups_is_invalid(slug: str):
     Returns True if it's invalid, otherwise return False
     """
     return not slug or not slug_with_subgroups_regex.match(slug)
+
+
+def slug_encoded_incorrectly(slug: str):
+    """
+    Checks if slug is encoded incorrectly based on the encoding mechanism we use.
+    Checks if slug is in the form of owner:::subowner::::repo or owner::::repo
+    Returns True if invalid, otherwise returns False
+    """
+    return not slug or not encoded_slug_regex.match(slug)

codecov_cli/helpers/git.py

@@ -3,6 +3,9 @@
 from enum import Enum
 from urllib.parse import urlparse
 
+from codecov_cli.helpers.encoder import decode_slug
+from codecov_cli.helpers.git_services.github import Github
+
 slug_regex = re.compile(r"[^/\s]+\/[^/\s]+$")
 
 logger = logging.getLogger("codecovcli")
@@ -17,6 +20,11 @@ class GitService(Enum):
     BITBUCKET_SERVER = "bitbucket_server"
 
 
+def get_git_service(git):
+    if git == "github":
+        return Github()
+
+
 def parse_slug(remote_repo_url: str):
     """
     Extracts a slug from git remote urls. returns None if the url is invalid
@@ -82,3 +90,13 @@ def parse_git_service(remote_repo_url: str):
             extra=dict(remote_repo_url=remote_repo_url),
         )
         return None
+
+
+def is_fork_pr(pr_num, slug, service):
+    decoded_slug = decode_slug(slug)
+    git_service = get_git_service(service)
+    if git_service:
+        pull_dict = git_service.get_pull_request(decoded_slug, pr_num)
+        if pull_dict and pull_dict["head"]["slug"] != decoded_slug:
+            return True
+    return False

codecov_cli/helpers/git_services/github.py

@@ -0,0 +1,32 @@
+import json
+
+import requests
+
+
+class Github:
+    api_url = "https://api.github.com"
+    api_version = "2022-11-28"
+
+    def get_pull_request(self, slug, pr_number):
+        pull_url = f"/repos/{slug}/pulls/{pr_number}"
+        url = self.api_url + pull_url
+        headers = {"X-GitHub-Api-Version": self.api_version}
+        response = requests.get(url, headers=headers)
+        if response.status_code == 200:
+            res = json.loads(response.text)
+            return {
+                "url": res["url"],
+                "head": {
+                    "sha": res["head"]["sha"],
+                    "label": res["head"]["label"],
+                    "ref": res["head"]["ref"],
+                    "slug": res["head"]["repo"]["full_name"],
+                },
+                "base": {
+                    "sha": res["base"]["sha"],
+                    "label": res["base"]["label"],
+                    "ref": res["base"]["ref"],
+                    "slug": res["base"]["repo"]["full_name"],
+                },
+            }
+        return None

codecov_cli/services/commit/__init__.py

@@ -4,6 +4,7 @@
 
 from codecov_cli.helpers.config import CODECOV_API_URL
 from codecov_cli.helpers.encoder import encode_slug
+from codecov_cli.helpers.git import is_fork_pr
 from codecov_cli.helpers.request import (
     get_token_header_or_fail,
     log_warnings_and_errors_if_any,
@@ -49,7 +50,11 @@ def send_commit_data(
         "pullid": pr,
         "branch": branch,
     }
-    headers = get_token_header_or_fail(token)
+    headers = (
+        {}
+        if not token and is_fork_pr(pr, slug, service)
+        else get_token_header_or_fail(token)
+    )
     upload_url = enterprise_url or CODECOV_API_URL
     url = f"{upload_url}/upload/{service}/{slug}/commits"
     return send_post_request(url=url, data=data, headers=headers)

tests/helpers/test_encoder.py

@@ -1,6 +1,11 @@
 import pytest
 
-from codecov_cli.helpers.encoder import encode_slug, slug_without_subgroups_is_invalid
+from codecov_cli.helpers.encoder import (
+    decode_slug,
+    encode_slug,
+    slug_encoded_incorrectly,
+    slug_without_subgroups_is_invalid,
+)
 
 
 @pytest.mark.parametrize(
@@ -53,3 +58,47 @@ def test_invalid_slug(slug):
 def test_valid_slug():
     slug = "owner/repo"
     assert not slug_without_subgroups_is_invalid(slug)
+
+
+@pytest.mark.parametrize(
+    "slug",
+    [
+        ("invalid_slug"),
+        (""),
+        (":"),
+        (":::"),
+        ("::::"),
+        ("random string"),
+        ("owner:::subgroup:::repo"),
+        ("owner:::repo"),
+        ("owner::::subgroup::::repo"),
+        (None),
+    ],
+)
+def test_invalid_encoded_slug(slug):
+    assert slug_encoded_incorrectly(slug)
+    with pytest.raises(ValueError) as ex:
+        decode_slug(slug)
+
+
+@pytest.mark.parametrize(
+    "encoded_slug",
+    [
+        ("owner::::repo"),
+        ("owner:::subgroup::::repo"),
+    ],
+)
+def test_valid_encoded_slug(encoded_slug):
+    assert not slug_encoded_incorrectly(encoded_slug)
+
+
+@pytest.mark.parametrize(
+    "encoded_slug, decoded_slug",
+    [
+        ("owner::::repo", "owner/repo"),
+        ("owner:::subgroup::::repo", "owner/subgroup/repo"),
+    ],
+)
+def test_decode_slug(encoded_slug, decoded_slug):
+    expected_encoded_slug = decode_slug(encoded_slug)
+    assert expected_encoded_slug == decoded_slug

tests/helpers/test_git.py

@@ -1,6 +1,11 @@
+import json
+
 import pytest
+import requests
+from requests import Response
 
 from codecov_cli.helpers import git
+from codecov_cli.helpers.git_services.github import Github
 
 
 @pytest.mark.parametrize(
@@ -119,3 +124,90 @@ def test_parse_git_service_valid_address(address, git_service):
 )
 def test_parse_git_service_invalid_service(url):
     assert git.parse_git_service(url) is None
+
+
+def test_get_git_service_class():
+    assert isinstance(git.get_git_service("github"), Github)
+    assert git.get_git_service("gitlab") == None
+    assert git.get_git_service("bitbucket") == None
+
+
+def test_pr_is_not_fork_pr(mocker):
+    def mock_request(*args, headers={}, **kwargs):
+        assert headers["X-GitHub-Api-Version"] == "2022-11-28"
+        res = {
+            "url": "https://api.github.com/repos/codecov/codecov-cli/pulls/1",
+            "head": {
+                "sha": "123",
+                "label": "codecov-cli:branch",
+                "ref": "branch",
+                "repo": {"full_name": "codecov/codecov-cli"},
+            },
+            "base": {
+                "sha": "123",
+                "label": "codecov-cli:main",
+                "ref": "main",
+                "repo": {"full_name": "codecov/codecov-cli"},
+            },
+        }
+        response = Response()
+        response.status_code = 200
+        response._content = json.dumps(res).encode("utf-8")
+        return response
+
+    mocker.patch.object(
+        requests,
+        "get",
+        side_effect=mock_request,
+    )
+    encoded_slug = "codecov::::codecov-cli"
+    assert not git.is_fork_pr(1, encoded_slug, "github")
+
+
+def test_pr_is_fork_pr(mocker):
+    def mock_request(*args, headers={}, **kwargs):
+        assert headers["X-GitHub-Api-Version"] == "2022-11-28"
+        res = {
+            "url": "https://api.github.com/repos/codecov/codecov-cli/pulls/325",
+            "head": {
+                "sha": "123",
+                "label": "codecov-cli:branch",
+                "ref": "branch",
+                "repo": {"full_name": "user_forked_repo/codecov-cli"},
+            },
+            "base": {
+                "sha": "123",
+                "label": "codecov-cli:main",
+                "ref": "main",
+                "repo": {"full_name": "codecov/codecov-cli"},
+            },
+        }
+        response = Response()
+        response.status_code = 200
+        response._content = json.dumps(res).encode("utf-8")
+        return response
+
+    mocker.patch.object(
+        requests,
+        "get",
+        side_effect=mock_request,
+    )
+    encoded_slug = "codecov::::codecov-cli"
+    assert git.is_fork_pr(1, encoded_slug, "github")
+
+
+def test_pr_not_found(mocker):
+    def mock_request(*args, headers={}, **kwargs):
+        assert headers["X-GitHub-Api-Version"] == "2022-11-28"
+        response = Response()
+        response.status_code = 404
+        response._content = b"not-found"
+        return response
+
+    mocker.patch.object(
+        requests,
+        "get",
+        side_effect=mock_request,
+    )
+    encoded_slug = "codecov::::codecov-cli"
+    assert not git.is_fork_pr(1, encoded_slug, "github")