[marketplace contributions] - fix issue with support labels (#27600)

* [marketplace contributions] - fix issue where support labels are not added

* add unit-tests

* update comment

* update path of test

* path cwd

* fallback to master in case checkout failed

* docstrings improvments

* update print string

* add prints

github_workflow_scripts/github_workflow_scripts_tests/handle_external_pr_test.py

@@ -1,3 +1,4 @@
+import os
 import pytest
 
 
@@ -28,13 +29,69 @@
 def test_get_highest_support_label(support_levels, expected_support_level):
     """
     Given:
-        a list of support levels for packs that were changed
+        - a list of support levels for packs that were changed
 
     When:
-        running get_highest_support_label function
+        - running get_highest_support_label function
 
     Then:
-        make sure the highest support level is always returned
+        - make sure the highest support level is always returned
     """
     from Utils.github_workflow_scripts.handle_external_pr import get_highest_support_label
     assert get_highest_support_label(support_levels) == expected_support_level
+
+
+@pytest.mark.parametrize(
+    'fork_owner, expected_fork_owner', [
+        ('test', 'test'),
+        ('xsoar-bot', 'xsoar-contrib')
+    ]
+)
+def test_get_packs_support_level_label(mocker, fork_owner, expected_fork_owner):
+    """
+    Given:
+        - a pack and a fork owner
+
+    When:
+        - running get_packs_support_level_label function
+
+    Then:
+        - make sure correct support label is returned.
+        - fork owner that is being delivered to the Checkout branch is correct.
+    """
+    from Utils.github_workflow_scripts.handle_external_pr import get_packs_support_level_label, Checkout
+    from Utils.github_workflow_scripts.utils import ChangeCWD
+
+    checkout_mocker = mocker.patch.object(Checkout, '__init__', return_value=None)
+    mocker.patch.object(Checkout, '__enter__', return_value=None)
+    mocker.patch.object(Checkout, '__exit__', return_value=None)
+    mocker.patch.object(os, 'getenv', return_value=fork_owner)
+
+    with ChangeCWD('Utils/github_workflow_scripts/github_workflow_scripts_tests/test_files'):
+        assert get_packs_support_level_label(
+            file_paths=['Packs/Pack1/pack_metadata.json'], external_pr_branch='test'
+        ) == 'Xsoar Support Level'
+
+    assert checkout_mocker.call_args.kwargs['fork_owner'] == expected_fork_owner
+
+
+def test_get_packs_support_level_label_checkout_failed(mocker):
+    """
+    Given:
+        - a pack
+
+    When:
+        - running get_packs_support_level_label function when Checkout fails.
+
+    Then:
+        - make sure correct support label is still returned.
+    """
+    from Utils.github_workflow_scripts.handle_external_pr import get_packs_support_level_label, Checkout
+    from Utils.github_workflow_scripts.utils import ChangeCWD
+
+    mocker.patch.object(Checkout, '__init__', return_value=Exception('Error'))
+
+    with ChangeCWD('Utils/github_workflow_scripts/github_workflow_scripts_tests/test_files'):
+        assert get_packs_support_level_label(
+            file_paths=['Packs/Pack1/pack_metadata.json'], external_pr_branch='test'
+        ) == 'Xsoar Support Level'

github_workflow_scripts/github_workflow_scripts_tests/test_files/Packs/Pack1/pack_metadata.json

@@ -0,0 +1,3 @@
+{
+    "support": "xsoar"
+}

github_workflow_scripts/handle_external_pr.py

@@ -124,15 +124,20 @@ def get_packs_support_level_label(file_paths: List[str], external_pr_branch: str
         f'to retrieve support level of {pack_dirs_to_check_support_levels_labels}'
     )
     try:
+        fork_owner = os.getenv('GITHUB_ACTOR')
         with Checkout(
             repo=Repo(Path().cwd(), search_parent_directories=True),
             branch_to_checkout=external_pr_branch,
-            fork_owner=os.getenv('GITHUB_ACTOR')
+            # in marketplace contributions the name of the owner should be xsoar-contrib
+            fork_owner=fork_owner if fork_owner != 'xsoar-bot' else 'xsoar-contrib'
         ):
             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=}')
+        # in case we were not able to checkout correctly, fallback to the files in the master branch to retrieve support labels
+        # in case those files exist.
+        print(f'Received error when trying to checkout to {external_pr_branch} \n{error=}')
+        print('Trying to retrieve support levels from the master branch')
+        packs_support_levels = get_packs_support_levels(pack_dirs_to_check_support_levels_labels)
 
     print(f'{packs_support_levels=}')
     return get_highest_support_label(packs_support_levels) if packs_support_levels else ''

github_workflow_scripts/utils.py

@@ -4,6 +4,7 @@
 import json
 from datetime import datetime
 from typing import Any, Generator, Iterable, Optional, Tuple, Union
+from pathlib import Path
 
 from git import Repo
 
@@ -108,6 +109,7 @@ def __init__(self, repo: Repo, branch_to_checkout: str, fork_owner: Optional[str
             url = f"https://github.com/{fork_owner}/{repo_name}"
             try:
                 self.repo.create_remote(name=forked_remote_name, url=url)
+                print(f'Successfully created remote {forked_remote_name} for repo {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
@@ -145,3 +147,20 @@ 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}")
+
+
+class ChangeCWD:
+    """
+    Temporary changes the cwd to the given dir and then reverts it.
+    Use with 'with' statement.
+    """
+
+    def __init__(self, directory):
+        self.current = Path().cwd()
+        self.directory = directory
+
+    def __enter__(self):
+        os.chdir(self.directory)
+
+    def __exit__(self, *args):
+        os.chdir(self.current)