Add a new transplant-report command

This new command can be used to copy over the coverage report from one commit to another, without having to trigger new uploads.

Author: Swatinem

codecov_cli/commands/commit.py

@@ -6,7 +6,6 @@
 
 from codecov_cli.fallbacks import CodecovOption, FallbackFieldEnum
 from codecov_cli.helpers.args import get_cli_args
-from codecov_cli.helpers.git import GitService
 from codecov_cli.helpers.options import global_options
 from codecov_cli.services.commit import create_commit_logic
 from codecov_cli.types import CommandContext

codecov_cli/commands/create_report_result.py

@@ -37,5 +37,12 @@ def create_report_results(
                 ),
             )
             create_report_results_logic(
-                commit_sha, code, slug, git_service, token, enterprise_url, fail_on_error, args
+                commit_sha,
+                code,
+                slug,
+                git_service,
+                token,
+                enterprise_url,
+                fail_on_error,
+                args,
             )

codecov_cli/commands/empty_upload.py

@@ -76,5 +76,12 @@ def empty_upload(
                 ),
             )
             return empty_upload_logic(
-                commit_sha, slug, token, git_service, enterprise_url, fail_on_error, force, args
+                commit_sha,
+                slug,
+                token,
+                git_service,
+                enterprise_url,
+                fail_on_error,
+                force,
+                args,
             )

codecov_cli/commands/transplant_report.py

@@ -0,0 +1,49 @@
+import logging
+import typing
+
+import click
+import sentry_sdk
+
+from codecov_cli.helpers.args import get_cli_args
+from codecov_cli.helpers.options import global_options
+from codecov_cli.services.report import transplant_report_logic
+from codecov_cli.types import CommandContext
+
+logger = logging.getLogger("codecovcli")
+
+
+@click.command(hidden=True)
+@click.option(
+    "--from-sha",
+    help="SHA (with 40 chars) of the commit from which to forward coverage reports",
+    required=True,
+)
+@global_options
+@click.pass_context
+def transplant_report(
+    ctx: CommandContext,
+    from_sha: str,
+    commit_sha: str,
+    slug: typing.Optional[str],
+    token: typing.Optional[str],
+    git_service: typing.Optional[str],
+    fail_on_error: bool,
+):
+    with sentry_sdk.start_transaction(op="task", name="Transplant Report"):
+        with sentry_sdk.start_span(name="transplant_report"):
+            enterprise_url = ctx.obj.get("enterprise_url")
+            args = get_cli_args(ctx)
+            logger.debug(
+                "Starting transplant report process",
+                extra=dict(extra_log_attributes=args),
+            )
+            transplant_report_logic(
+                from_sha,
+                commit_sha,
+                slug,
+                token,
+                git_service,
+                enterprise_url,
+                fail_on_error,
+                args,
+            )

codecov_cli/helpers/glob.py

@@ -31,44 +31,46 @@ def translate(pat, *, recursive=False, include_hidden=False, seps=None):
             seps = (os.path.sep, os.path.altsep)
         else:
             seps = os.path.sep
-    escaped_seps = ''.join(map(re.escape, seps))
-    any_sep = f'[{escaped_seps}]' if len(seps) > 1 else escaped_seps
-    not_sep = f'[^{escaped_seps}]'
+    escaped_seps = "".join(map(re.escape, seps))
+    any_sep = f"[{escaped_seps}]" if len(seps) > 1 else escaped_seps
+    not_sep = f"[^{escaped_seps}]"
     if include_hidden:
-        one_last_segment = f'{not_sep}+'
-        one_segment = f'{one_last_segment}{any_sep}'
-        any_segments = f'(?:.+{any_sep})?'
-        any_last_segments = '.*'
+        one_last_segment = f"{not_sep}+"
+        one_segment = f"{one_last_segment}{any_sep}"
+        any_segments = f"(?:.+{any_sep})?"
+        any_last_segments = ".*"
     else:
-        one_last_segment = f'[^{escaped_seps}.]{not_sep}*'
-        one_segment = f'{one_last_segment}{any_sep}'
-        any_segments = f'(?:{one_segment})*'
-        any_last_segments = f'{any_segments}(?:{one_last_segment})?'
+        one_last_segment = f"[^{escaped_seps}.]{not_sep}*"
+        one_segment = f"{one_last_segment}{any_sep}"
+        any_segments = f"(?:{one_segment})*"
+        any_last_segments = f"{any_segments}(?:{one_last_segment})?"
 
     results = []
     parts = re.split(any_sep, pat)
     last_part_idx = len(parts) - 1
     for idx, part in enumerate(parts):
-        if part == '*':
+        if part == "*":
             results.append(one_segment if idx < last_part_idx else one_last_segment)
-        elif recursive and part == '**':
+        elif recursive and part == "**":
             if idx < last_part_idx:
-                if parts[idx + 1] != '**':
+                if parts[idx + 1] != "**":
                     results.append(any_segments)
             else:
                 results.append(any_last_segments)
         else:
             if part:
-                if not include_hidden and part[0] in '*?':
-                    results.append(r'(?!\.)')
-                results.extend(_translate(part, f'{not_sep}*', not_sep)[0])
+                if not include_hidden and part[0] in "*?":
+                    results.append(r"(?!\.)")
+                results.extend(_translate(part, f"{not_sep}*", not_sep)[0])
             if idx < last_part_idx:
                 results.append(any_sep)
-    res = ''.join(results)
-    return fr'(?s:{res})\Z'
+    res = "".join(results)
+    return rf"(?s:{res})\Z"
+
+
+_re_setops_sub = re.compile(r"([&~|])").sub
 
 
-_re_setops_sub = re.compile(r'([&~|])').sub
 def _translate(pat, star, question_mark):
     res = []
     add = res.append
@@ -77,69 +79,70 @@ def _translate(pat, star, question_mark):
     i, n = 0, len(pat)
     while i < n:
         c = pat[i]
-        i = i+1
-        if c == '*':
+        i = i + 1
+        if c == "*":
             # store the position of the wildcard
             star_indices.append(len(res))
             add(star)
             # compress consecutive `*` into one
-            while i < n and pat[i] == '*':
+            while i < n and pat[i] == "*":
                 i += 1
-        elif c == '?':
+        elif c == "?":
             add(question_mark)
-        elif c == '[':
+        elif c == "[":
             j = i
-            if j < n and pat[j] == '!':
-                j = j+1
-            if j < n and pat[j] == ']':
-                j = j+1
-            while j < n and pat[j] != ']':
-                j = j+1
+            if j < n and pat[j] == "!":
+                j = j + 1
+            if j < n and pat[j] == "]":
+                j = j + 1
+            while j < n and pat[j] != "]":
+                j = j + 1
             if j >= n:
-                add('\\[')
+                add("\\[")
             else:
                 stuff = pat[i:j]
-                if '-' not in stuff:
-                    stuff = stuff.replace('\\', r'\\')
+                if "-" not in stuff:
+                    stuff = stuff.replace("\\", r"\\")
                 else:
                     chunks = []
-                    k = i+2 if pat[i] == '!' else i+1
+                    k = i + 2 if pat[i] == "!" else i + 1
                     while True:
-                        k = pat.find('-', k, j)
+                        k = pat.find("-", k, j)
                         if k < 0:
                             break
                         chunks.append(pat[i:k])
-                        i = k+1
-                        k = k+3
+                        i = k + 1
+                        k = k + 3
                     chunk = pat[i:j]
                     if chunk:
                         chunks.append(chunk)
                     else:
-                        chunks[-1] += '-'
+                        chunks[-1] += "-"
                     # Remove empty ranges -- invalid in RE.
-                    for k in range(len(chunks)-1, 0, -1):
-                        if chunks[k-1][-1] > chunks[k][0]:
-                            chunks[k-1] = chunks[k-1][:-1] + chunks[k][1:]
+                    for k in range(len(chunks) - 1, 0, -1):
+                        if chunks[k - 1][-1] > chunks[k][0]:
+                            chunks[k - 1] = chunks[k - 1][:-1] + chunks[k][1:]
                             del chunks[k]
                     # Escape backslashes and hyphens for set difference (--).
                     # Hyphens that create ranges shouldn't be escaped.
-                    stuff = '-'.join(s.replace('\\', r'\\').replace('-', r'\-')
-                                     for s in chunks)
-                i = j+1
+                    stuff = "-".join(
+                        s.replace("\\", r"\\").replace("-", r"\-") for s in chunks
+                    )
+                i = j + 1
                 if not stuff:
                     # Empty range: never match.
-                    add('(?!)')
-                elif stuff == '!':
+                    add("(?!)")
+                elif stuff == "!":
                     # Negated empty range: match any character.
-                    add('.')
+                    add(".")
                 else:
                     # Escape set operations (&&, ~~ and ||).
-                    stuff = _re_setops_sub(r'\\\1', stuff)
-                    if stuff[0] == '!':
-                        stuff = '^' + stuff[1:]
-                    elif stuff[0] in ('^', '['):
-                        stuff = '\\' + stuff
-                    add(f'[{stuff}]')
+                    stuff = _re_setops_sub(r"\\\1", stuff)
+                    if stuff[0] == "!":
+                        stuff = "^" + stuff[1:]
+                    elif stuff[0] in ("^", "["):
+                        stuff = "\\" + stuff
+                    add(f"[{stuff}]")
         else:
             add(re.escape(c))
     assert i == n

codecov_cli/helpers/request.py

@@ -79,7 +79,9 @@ def wrapper(*args, **kwargs):
                 )
                 sleep(backoff_time(retry))
                 retry += 1
-        raise Exception(f"Request failed after too many retries. URL: {kwargs.get('url', args[0] if args else 'Unknown')}")
+        raise Exception(
+            f"Request failed after too many retries. URL: {kwargs.get('url', args[0] if args else 'Unknown')}"
+        )
 
     return wrapper
 

codecov_cli/main.py

@@ -19,6 +19,7 @@
 from codecov_cli.commands.upload import do_upload
 from codecov_cli.commands.upload_coverage import upload_coverage
 from codecov_cli.commands.upload_process import upload_process
+from codecov_cli.commands.transplant_report import transplant_report
 from codecov_cli.helpers.ci_adapters import get_ci_adapter, get_ci_providers_list
 from codecov_cli.helpers.config import load_cli_config
 from codecov_cli.helpers.logging_utils import configure_logger
@@ -76,6 +77,7 @@ def cli(
 
 cli.add_command(do_upload)
 cli.add_command(create_commit)
+cli.add_command(transplant_report)
 cli.add_command(create_report)
 cli.add_command(create_report_results)
 cli.add_command(get_report_results)

codecov_cli/plugins/compress_pycoverage_contexts.py

@@ -64,7 +64,9 @@ def run_preparation(self, collector) -> PreparationPluginReturn:
                 )
                 return PreparationPluginReturn(
                     success=False,
-                    messages=[f"File to compress {self.file_to_compress} is not a file."],
+                    messages=[
+                        f"File to compress {self.file_to_compress} is not a file."
+                    ],
                 )
             # Create in and out streams
             fd_in = open(self.file_to_compress, "rb")

codecov_cli/plugins/gcov.py

@@ -40,7 +40,9 @@ def run_preparation(self, collector) -> PreparationPluginReturn:
                 logger.warning(f"{self.executable} is not installed or can't be found.")
                 return
 
-            filename_include_regex = globs_to_regex(["*.gcno", *self.patterns_to_include])
+            filename_include_regex = globs_to_regex(
+                ["*.gcno", *self.patterns_to_include]
+            )
             filename_exclude_regex = globs_to_regex(self.patterns_to_ignore)
 
             matched_paths = [

codecov_cli/services/report/__init__.py

@@ -101,9 +101,7 @@ def send_reports_result_request(
     enterprise_url,
     args,
 ):
-    data = {
-        "cli_args": args,
-    }
+    data = {"cli_args": args}
     headers = get_token_header(token)
     upload_url = enterprise_url or CODECOV_API_URL
     url = f"{upload_url}/upload/{service}/{encoded_slug}/commits/{commit_sha}/reports/{report_code}/results"
@@ -167,3 +165,28 @@ def send_reports_result_get_request(
         time.sleep(5)
         number_tries += 1
     return response_obj
+
+
+def transplant_report_logic(
+    from_sha: str,
+    to_sha: str,
+    slug: typing.Optional[str],
+    token: typing.Optional[str],
+    service: typing.Optional[str],
+    enterprise_url: typing.Optional[str] = None,
+    fail_on_error: bool = False,
+    args: typing.Union[dict, None] = None,
+):
+    slug = encode_slug(slug)
+    headers = get_token_header(token)
+
+    data = {"cli_args": args, "from_sha": from_sha, "to_sha": to_sha}
+
+    base_url = enterprise_url or CODECOV_INGEST_URL
+    url = f"{base_url}/upload/{service}/{slug}/commits/transplant"
+    sending_result = send_post_request(url=url, data=data, headers=headers)
+
+    log_warnings_and_errors_if_any(
+        sending_result, "Transplanting report", fail_on_error
+    )
+    return sending_result