Support level tags for external contributions bug fix (#27167)

* fix bug on new files

* add existing file path change

* add new partner pack for testing

* update test

* update

* add python to pipenv

* test change cwd works

* remove panorama

* check if works without change-cwd

* add the option to checkout branch

* try to see behaivor with checkout

* comment out

* fix import issue

* search in paranets path

* fixes

* align the print

* update

* remove test files

* remove pipfile

* handle bug

* add support to checkout forked branches

* test

* add support for checking out forked branches

* add print

* pragma no cover

* small fix

* enhancments

* bug fix

* remove unused imports

* add forked repo even if not content

* uncomment main code

* always checkout to the branch

* remove  copy

* cr fixes

* cr fixes

* empty packs support level in case of exception

github_workflow_scripts/handle_external_pr.py

@@ -1,14 +1,17 @@
 #!/usr/bin/env python3
 import json
+import os
 
 from typing import List, Set
+from pathlib import Path
 
 import urllib3
 from blessings import Terminal
 from github import Github
+from git import Repo
 from github.Repository import Repository
 
-from utils import get_env_var, timestamped_print
+from utils import get_env_var, timestamped_print, Checkout
 from demisto_sdk.commands.common.tools import get_pack_metadata, get_pack_name
 
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
@@ -64,7 +67,26 @@ def determine_reviewer(potential_reviewers: List[str], repo: Repository) -> str:
     return selected_reviewer
 
 
-def get_packs_support_level_label(file_paths: List[str]) -> str:
+def get_packs_support_levels(pack_dirs: Set[str]) -> Set[str]:
+    """
+    Get the pack support levels from the pack metadata.
+
+    Args:
+        pack_dirs (set): paths to the packs that were changed
+    """
+    packs_support_levels = set()
+
+    for pack_dir in pack_dirs:
+        if pack_support_level := get_pack_metadata(pack_dir).get('support'):
+            print(f'Pack support level for pack {pack_dir} is {pack_support_level}')
+            packs_support_levels.add(pack_support_level)
+        else:
+            print(f'Could not find pack support level for pack {pack_dir}')
+
+    return packs_support_levels
+
+
+def get_packs_support_level_label(file_paths: List[str], external_pr_branch: str) -> str:
     """
     Get The contributions' support level label.
 
@@ -79,32 +101,41 @@ def get_packs_support_level_label(file_paths: List[str]) -> str:
 
     Args:
         file_paths(str): file paths
+        external_pr_branch (str): the branch of the external PR.
 
     Returns:
         highest support level of the packs that were changed, empty string in case no packs were changed.
     """
-    changed_pack_dirs = set()
+    pack_dirs_to_check_support_levels_labels = set()
+
     for file_path in file_paths:
         try:
             if 'Packs' in file_path and (pack_name := get_pack_name(file_path)):
-                changed_pack_dirs.add(f'Packs/{pack_name}')
+                pack_dirs_to_check_support_levels_labels.add(f'Packs/{pack_name}')
         except Exception as err:
             print(f'Could not retrieve pack name from file {file_path}, {err=}')
 
-    print(f'{changed_pack_dirs=}')
-
-    packs_support_levels = set()
-
-    for pack_dir in changed_pack_dirs:
-        if pack_support_level := get_pack_metadata(pack_dir).get('support'):
-            print(f'Pack support level for pack {pack_dir} is {pack_support_level}')
-            packs_support_levels.add(pack_support_level)
+    print(f'{pack_dirs_to_check_support_levels_labels=}')
+
+    # # we need to check out to the contributor branch in his forked repo in order to retrieve the files cause workflow
+    # runs on demisto master while the contributions changes are on the contributors branch
+    print(
+        f'Trying to checkout to forked branch {external_pr_branch} '
+        f'to retrieve support level of {pack_dirs_to_check_support_levels_labels}'
+    )
+    try:
+        with Checkout(
+            repo=Repo(Path().cwd(), search_parent_directories=True),
+            branch_to_checkout=external_pr_branch,
+            fork_owner=os.getenv('GITHUB_ACTOR')
+        ):
+            packs_support_levels = get_packs_support_levels(pack_dirs_to_check_support_levels_labels)
+    except Exception as error:
+        packs_support_levels = set()
+        print(f'Received error when trying to checkout to {external_pr_branch} forked content repo\n{error=}')
 
     print(f'{packs_support_levels=}')
-
-    if packs_support_levels:
-        return get_highest_support_label(packs_support_levels)
-    return ''
+    return get_highest_support_label(packs_support_levels) if packs_support_levels else ''
 
 
 def get_highest_support_label(packs_support_levels: Set[str]):
@@ -148,14 +179,15 @@ def main():
     repo_name = 'content'
     gh = Github(get_env_var('CONTENTBOT_GH_ADMIN_TOKEN'), verify=False)
     content_repo = gh.get_repo(f'{org_name}/{repo_name}')
+
     pr_number = payload.get('pull_request', {}).get('number')
     pr = content_repo.get_pull(pr_number)
 
-    changed_files_paths = [file.filename for file in pr.get_files()]
-    print(f'{changed_files_paths=} for {pr_number=}')
+    pr_files = [file.filename for file in pr.get_files()]
+    print(f'{pr_files=} for {pr_number=}')
 
     labels_to_add = [CONTRIBUTION_LABEL]
-    if support_label := get_packs_support_level_label(changed_files_paths):
+    if support_label := get_packs_support_level_label(pr_files, pr.head.ref):
         labels_to_add.append(support_label)
 
     # Add 'Contribution' + support Label to the external PR

github_workflow_scripts/utils.py

@@ -92,25 +92,56 @@ class Checkout:  # pragma: no cover
     previously current branch.
     """
 
-    def __init__(self, repo: Repo, branch_to_checkout: str):
+    def __init__(self, repo: Repo, branch_to_checkout: str, fork_owner: Optional[str] = None, repo_name: str = 'content'):
         """Initializes instance attributes.
         Arguments:
             repo: git repo object
             branch_to_checkout: The branch or commit hash to check out.
+            fork_owner (str): The owner of the forked repository.
+                Leave it as None if the branch is in the same repository.
+            repo_name (str): the name of the forked repo (without the owner)
         """
         self.repo = repo
-        self.repo.remote().fetch(branch_to_checkout)
-        self._branch_to_checkout = branch_to_checkout
+
+        if fork_owner:
+            forked_remote_name = f'{fork_owner}_{repo_name}_{branch_to_checkout}_remote'
+            url = f"https://github.com/{fork_owner}/{repo_name}"
+            try:
+                self.repo.create_remote(name=forked_remote_name, url=url)
+            except Exception as error:
+                print(f'could not create remote from {url}, {error=}')
+                # handle the case where the name of the forked repo is not content
+                if github_event_path := os.getenv("GITHUB_EVENT_PATH"):
+                    try:
+                        payload = json.loads(github_event_path)
+                    except ValueError:
+                        print('failed to load GITHUB_EVENT_PATH')
+                        raise ValueError(f'cannot checkout to the forked branch {branch_to_checkout} of the owner {fork_owner}')
+                    # forked repo name includes fork_owner + repo name, for example foo/content.
+                    forked_repo_name = payload.get("pull_request", {}).get("head", {}).get("repo", {}).get("full_name")
+                    self.repo.create_remote(name=forked_remote_name, url=f"https://github.com/{forked_repo_name}")
+                else:
+                    raise
+
+            forked_remote = self.repo.remote(forked_remote_name)
+            forked_remote.fetch(branch_to_checkout)
+            self.branch_to_checkout = f'refs/remotes/{forked_remote_name}/{branch_to_checkout}'
+        else:
+            self.branch_to_checkout = branch_to_checkout
+            self.repo.remote().fetch(branch_to_checkout)
+
         try:
             self._original_branch = self.repo.active_branch.name
         except TypeError:
             self._original_branch = self.repo.git.rev_parse('HEAD')
 
     def __enter__(self):
         """Checks out the given branch"""
-        self.repo.git.checkout(self._branch_to_checkout)
+        self.repo.git.checkout(self.branch_to_checkout)
+        print(f'Checked out to branch {self.branch_to_checkout}')
         return self
 
     def __exit__(self, *args):
         """Checks out the previous branch"""
         self.repo.git.checkout(self._original_branch)
+        print(f"Checked out to original branch {self._original_branch}")