sfdo-tooling__cumulusci-3624

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "cumulusci/tasks/github:\n__init__.py\nbase.py\ncommit_status.py\nmerge.py\npublish.py\npull_request.py\nrelease.py\nrelease_report.py\ntag.py\ntests\nutil.py\n\ncumulusci/tasks/github/tests:\n__init__.py\ntest_commit_status.py\ntest_merge.py\ntest_publish.py\ntest_pull_request.py\ntest_release.py\ntest_release_report.py\ntest_tag.py\ntest_util.py\nutil_github_api.py\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "error",
  "error_type": "CommandError",
  "error_message": "",
  "details": {
    "stdout": "",
    "stderr": "",
    "exit_code": 1
  }
}

{
  "status": "ok",
  "stdout": "cumulusci/tasks/github/merge.py:31:        \"skip_future_releases\": {\ncumulusci/tasks/github/merge.py:52:        self.options[\"skip_future_releases\"] = process_bool_arg(\ncumulusci/tasks/github/merge.py:53:            self.options.get(\"skip_future_releases\") or True\ncumulusci/tasks/github/merge.py:108:        skip_future_releases = self.options[\"skip_future_releases\"]\ncumulusci/tasks/github/merge.py:143:                if skip_future_releases and self._is_future_release_branch(\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "cumulusci/tasks/github/merge.py",
  "content": "import http.client\n\nimport github3.exceptions\nfrom github3 import GitHubError\n\nfrom cumulusci.core.exceptions import GithubApiNotFoundError\nfrom cumulusci.core.utils import process_bool_arg\nfrom cumulusci.tasks.github.base import BaseGithubTask\nfrom cumulusci.utils.git import is_release_branch\n\n\nclass MergeBranch(BaseGithubTask):\n    task_docs = \"\"\"\n    Merges the most recent commit on the current branch into other branches depending on the value of source_branch.\n\n    If source_branch is a branch that does not start with the specified branch_prefix, then the commit will be\n    merged to all branches that begin with branch_prefix and are not themselves child branches (i.e. branches don't contain '__' in their name).\n\n    If source_branch begins with branch_prefix, then the commit is merged to all child branches of source_branch.\n    \"\"\"\n    task_options = {  # TODO: should use `class Options instead`\n        \"commit\": {\n            \"description\": \"The commit to merge into feature branches.  Defaults to the current head commit.\"\n        },\n        \"source_branch\": {\n            \"description\": \"The source branch to merge from.  Defaults to project__git__default_branch.\"\n        },\n        \"branch_prefix\": {\n            \"description\": \"A list of prefixes of branches that should receive the merge.  Defaults to project__git__prefix_feature\"\n        },\n        \"skip_future_releases\": {\n            \"description\": \"If true, then exclude branches that start with the branch prefix if they are not for the lowest release number. Defaults to True.\"\n        },\n        \"update_future_releases\": {\n            \"description\": \"If true, then include release branches that are not the lowest release number even if they are not child branches. Defaults to False.\"\n        },\n    }\n\n    def _init_options(self, kwargs):\n        super()._init_options(kwargs)\n\n        if \"commit\" not in self.options:\n            self.options[\"commit\"] = self.project_config.repo_commit\n        if \"branch_prefix\" not in self.options:\n            self.options[\n                \"branch_prefix\"\n            ] = self.project_config.project__git__prefix_feature\n        if \"source_branch\" not in self.options:\n            self.options[\n                \"source_branch\"\n            ] = self.project_config.project__git__default_branch\n        self.options[\"skip_future_releases\"] = process_bool_arg(\n            self.options.get(\"skip_future_releases\") or True\n        )\n        self.options[\"update_future_releases\"] = process_bool_arg(\n            self.options.get(\"update_future_releases\") or False\n        )\n\n    def _init_task(self):\n        super()._init_task()\n        self.repo = self.get_repo()\n\n    def _run_task(self):\n        self._validate_source_branch(self.options[\"source_branch\"])\n        branches_to_merge = self._get_branches_to_merge()\n\n        for branch in branches_to_merge:\n            self._merge(\n                branch.name,\n                self.options[\"source_branch\"],\n                self.options[\"commit\"],\n            )\n\n    def _validate_source_branch(self, source_branch):\n        \"\"\"Validates that the source branch exists in the repository\"\"\"\n        try:\n            self.repo.branch(source_branch)\n        except github3.exceptions.NotFoundError:\n            message = f\"Branch {source_branch} not found\"\n            raise GithubApiNotFoundError(message)\n\n    def _get_existing_prs(self, source_branch, branch_prefix):\n        \"\"\"Returns the existing pull requests from the source branch\n        to other branches that are candidates for merging.\"\"\"\n        existing_prs = []\n        for pr in self.repo.pull_requests(state=\"open\"):\n            if pr.base.ref.startswith(branch_prefix) and pr.head.ref == source_branch:\n                existing_prs.append(pr.base.ref)\n        return existing_prs\n\n    def _get_branches_to_merge(self):\n        \"\"\"\n        If source_branch is the default branch (or a branch that doesn't start with a prefix), we\n        gather all branches with branch_prefix that are not child branches.\n        NOTE: We only include the _next_ closest release branch when automerging from main.\n        A change on main may conflict with the current contents of the lowest release branch.\n        In this case, we would like for that conflict to only need to be resolved once\n        (not once for each release branch).\n\n        If source_branch starts with branch prefix, we gather\n        all branches with branch_prefix that are direct descendents of source_branch.\n\n        If update_future_releases is True, and source_branch is a release branch\n        then we also collect all future release branches.\n        \"\"\"\n        repo_branches = list(self.repo.branches())\n        next_release = self._get_next_release(repo_branches)\n        skip_future_releases = self.options[\"skip_future_releases\"]\n        update_future_releases = self._update_future_releases(next_release)\n\n        child_branches = []\n        main_descendents = []\n        release_branches = []\n        for branch in repo_branches:\n            # check for adding future release branches\n            if update_future_releases and self._is_future_release_branch(\n                branch.name, next_release\n            ):\n                release_branches.append(branch)\n                continue\n\n            # check if we looking at the source_branch\n            if branch.name == self.options[\"source_branch\"]:\n                self.logger.debug(f\"Skipping branch {branch.name}: is source branch\")\n                continue\n\n            # check for branch prefix match\n            elif not branch.name.startswith(self.options[\"branch_prefix\"]):\n                self.logger.debug(\n                    f\"Skipping branch {branch.name}: does not match prefix '{self.options['branch_prefix']}'\"\n                )\n                continue\n\n            # check if source_branch doesn't have prefix and is not a child (e.g. main)\n            elif (\n                not self.options[\"source_branch\"].startswith(\n                    self.options[\"branch_prefix\"]\n                )\n                and \"__\" not in branch.name\n            ):\n                # only merge to the lowest numbered release branch\n                # when merging from a branch without a prefix (e.g. main)\n                if skip_future_releases and self._is_future_release_branch(\n                    branch.name, next_release\n                ):\n                    continue\n                main_descendents.append(branch)\n\n            # else, we have a branch that starts with branch_prefix\n            # check is this branch is a direct descendent",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 287,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "53:def process_bool_arg(arg: T.Union[int, str, None]):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "cumulusci/core/utils.py",
  "content": "def process_bool_arg(arg: T.Union[int, str, None]):\n    \"\"\"Determine True/False from argument.\n\n    Similar to parts of the Salesforce API, there are a few true-ish and false-ish strings,\n        but \"True\" and \"False\" are the canonical ones.\n\n    None is accepted as \"False\" for backwards compatiblity reasons, but this usage is deprecated.\n    \"\"\"\n    if isinstance(arg, (int, bool)):\n        return bool(arg)\n    elif arg is None:\n        # backwards compatible behaviour that some tasks\n        # rely upon.\n        import traceback\n\n        warnings.warn(\"\".join(traceback.format_stack(limit=4)), DeprecationWarning)\n        warnings.warn(\n            \"Future versions of CCI will not accept 'None' as an argument to process_bool_arg\",",
  "start_line": 53,
  "end_line": 70,
  "total_lines": 364,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "cumulusci/core/utils.py",
  "content": "def process_bool_arg(arg: T.Union[int, str, None]):\n    \"\"\"Determine True/False from argument.\n\n    Similar to parts of the Salesforce API, there are a few true-ish and false-ish strings,\n        but \"True\" and \"False\" are the canonical ones.\n\n    None is accepted as \"False\" for backwards compatiblity reasons, but this usage is deprecated.\n    \"\"\"\n    if isinstance(arg, (int, bool)):\n        return bool(arg)\n    elif arg is None:\n        # backwards compatible behaviour that some tasks\n        # rely upon.\n        import traceback\n\n        warnings.warn(\"\".join(traceback.format_stack(limit=4)), DeprecationWarning)\n        warnings.warn(\n            \"Future versions of CCI will not accept 'None' as an argument to process_bool_arg\",\n            DeprecationWarning,\n        )\n\n        return False\n    elif isinstance(arg, str):\n        # these are values that Salesforce's bulk loader accepts\n        # there doesn't seem to be any harm in acccepting the\n        # full list to be coordinated with a \"Salesforce standard\"\n        if arg.lower() in [\"yes\", \"y\", \"true\", \"on\", \"1\"]:\n            return True",
  "start_line": 53,
  "end_line": 80,
  "total_lines": 364,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "cumulusci/core/utils.py",
  "content": "\"\"\" Utilities for CumulusCI Core\"\"\"\n\nimport copy\nimport glob\nimport json\nimport time\nimport typing as T\nimport warnings\nfrom datetime import datetime, timedelta\nfrom logging import getLogger\nfrom pathlib import Path\n\nimport pytz\n\nfrom cumulusci.core.debug import get_debug_mode\nfrom cumulusci.core.exceptions import (\n    ConfigMergeError,\n    CumulusCIException,\n    TaskOptionsError,\n)\nfrom cumulusci.utils.options import parse_list_of_pairs_dict_arg\n\n\ndef import_global(path: str):\n    \"\"\"Import a class from a string module class path\"\"\"\n    components = path.split(\".\")\n    module = components[:-1]\n    module = \".\".join(module)\n\n    if get_debug_mode():\n        import sys\n\n        logger = getLogger(__file__)\n        logger.info(f\"Importing {path}\")\n        logger.info(f\"Looking in sys.path {sys.path}\")\n        if components[0] == \"tasks\":\n            logger.info(f\"With tasks.__path__ {sys.modules['tasks'].__path__}\")\n\n    mod = __import__(module, fromlist=[str(components[-1])])\n    return getattr(mod, str(components[-1]))\n\n\n# For backwards-compatibility\nimport_class = import_global\n\n\ndef parse_datetime(dt_str, format):\n    \"\"\"Create a timezone-aware datetime object from a datetime string.\"\"\"\n    t = time.strptime(dt_str, format)\n    return datetime(t[0], t[1], t[2], t[3], t[4], t[5], t[6], pytz.UTC)\n\n\ndef process_bool_arg(arg: T.Union[int, str, None]):\n    \"\"\"Determine True/False from argument.\n\n    Similar to parts of the Salesforce API, there are a few true-ish and false-ish strings,\n        but \"True\" and \"False\" are the canonical ones.\n\n    None is accepted as \"False\" for backwards compatiblity reasons, but this usage is deprecated.\n    \"\"\"\n    if isinstance(arg, (int, bool)):\n        return bool(arg)\n    elif arg is None:\n        # backwards compatible behaviour that some tasks\n        # rely upon.\n        import traceback\n\n        warnings.warn(\"\".join(traceback.format_stack(limit=4)), DeprecationWarning)\n        warnings.warn(\n            \"Future versions of CCI will not accept 'None' as an argument to process_bool_arg\",\n            DeprecationWarning,\n        )\n\n        return False\n    elif isinstance(arg, str):\n        # these are values that Salesforce's bulk loader accepts\n        # there doesn't seem to be any harm in acccepting the\n        # full list to be coordinated with a \"Salesforce standard\"\n        if arg.lower() in [\"yes\", \"y\", \"true\", \"on\", \"1\"]:\n            return True\n        elif arg.lower() in [\"no\", \"n\", \"false\", \"off\", \"0\"]:\n            return False\n    raise TypeError(f\"Cannot interpret as boolean: `{arg}`\")\n\n\ndef process_glob_list_arg(arg):\n    \"\"\"Convert a list of glob patterns or filenames into a list of files\n    The initial list can take the form of a comma-separated string or\n    a proper list. Order is preserved, but duplicates will be removed.\n\n    Note: this function processes glob patterns, but doesn't validate\n    that the files actually exist. For example, if the pattern is\n    'foo.bar' and there is no file named 'foo.bar', the literal string\n    'foo.bar' will be included in the returned files.\n\n    Similarly, if the pattern is '*.baz' and it doesn't match any files,\n    the literal string '*.baz' will be returned.\n    \"\"\"\n    initial_list = process_list_arg(arg)\n\n    if not arg:\n        return []\n\n    files = []\n    for path in initial_list:\n        more_files = glob.glob(path, recursive=True)\n        if len(more_files):\n            files += sorted(more_files)\n        else:\n            files.append(path)\n    # In python 3.6+ dict is ordered, so we'll use it to weed\n    # out duplicates. We can't use a set because sets aren't ordered.\n    return list(dict.fromkeys(files))\n\n\ndef process_list_arg(arg):\n    \"\"\"Parse a string into a list separated by commas with whitespace stripped\"\"\"\n    if isinstance(arg, Path):\n        arg = str(arg)\n\n    if isinstance(arg, list):\n        return arg\n    elif isinstance(arg, str):\n        args = []\n        for part in arg.split(\",\"):\n            args.append(part.strip())\n        return args\n    elif arg is None:\n        # backwards compatible behaviour.\n        return None\n    else:\n        getLogger(__file__).warn(\n            f\"Unknown option type `{type(arg)}` for value `{arg}`.\"\n            \"This will be an error in a future version of CCI.\"\n        )\n\n\ndef process_list_of_pairs_dict_arg(arg):\n    \"\"\"Process an arg in the format \"aa:bb,cc:dd\" \"\"\"\n    try:\n        return parse_list_of_pairs_dict_arg(arg)\n    except TypeError as e:\n        raise TaskOptionsError(e) from e\n\n\ndef decode_to_unicode(content):\n    \"\"\"decode ISO-8859-1 to unicode, when using sf api\"\"\"\n    if content and not isinstance(content, str):\n        try:\n            # Try to decode ISO-8859-1 to unicode",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 364,
  "is_truncated": true
}

{
  "status": "ok",
  "stdout": "cumulusci/core/utils.py:53:def process_bool_arg(arg: T.Union[int, str, None]):\ncumulusci/core/tests/test_utils.py:19:    def test_process_bool_arg(self):\ncumulusci/core/tests/test_utils.py:21:            assert utils.process_bool_arg(arg)\ncumulusci/core/tests/test_utils.py:24:            assert not utils.process_bool_arg(arg)\ncumulusci/core/tests/test_utils.py:29:            assert utils.process_bool_arg(None) is False\ncumulusci/core/tests/test_utils.py:32:            utils.process_bool_arg(datetime.datetime.now())\ncumulusci/core/tests/test_utils.py:35:            utils.process_bool_arg(\"xyzzy\")\ncumulusci/salesforce_api/package_install.py:74:                options.activate_remote_site_settings = process_bool_arg(\ncumulusci/tasks/marketing_cloud/api.py:101:            if process_bool_arg(self.options.get(\"activate_if_existing\"))\ncumulusci/tasks/github/release_report.py:35:        self.options[\"include_beta\"] = process_bool_arg(\ncumulusci/tasks/github/release_report.py:38:        self.options[\"print\"] = process_bool_arg(self.options.get(\"print\", False))\ncumulusci/tasks/github/publish.py:83:        self.options[\"create_release\"] = process_bool_arg(\ncumulusci/tasks/github/publish.py:86:        self.options[\"release_body\"] = process_bool_arg(\ncumulusci/tasks/github/publish.py:89:        self.options[\"dry_run\"] = process_bool_arg(self.options.get(\"dry_run\", False))\ncumulusci/tasks/github/merge.py:52:        self.options[\"skip_future_releases\"] = process_bool_arg(\ncumulusci/tasks/github/merge.py:55:        self.options[\"update_future_releases\"] = process_bool_arg(\ncumulusci/tasks/metadata_etl/base.py:70:            self.options[\"managed\"] = process_bool_arg(self.options[\"managed\"] or False)\ncumulusci/tasks/metadata_etl/base.py:76:            self.options[\"namespaced_org\"] = process_bool_arg(\ncumulusci/tasks/metadata_etl/duplicate_rules.py:21:        status = \"true\" if process_bool_arg(self.options[\"active\"]) else \"false\"\ncumulusci/tasks/metadata_etl/help_text.py:27:        self.options[\"overwrite\"] = process_bool_arg(\ncumulusci/tasks/metadata_etl/layouts.py:275:            required = process_bool_arg(field.get(\"required\") or False)\ncumulusci/tasks/metadata_etl/layouts.py:276:            read_only = process_bool_arg(field.get(\"read_only\") or False)\ncumulusci/tasks/metadata_etl/layouts.py:313:                str(process_bool_arg(page.get(\"show_label\") or False)).lower(),\ncumulusci/tasks/metadata_etl/layouts.py:317:                str(process_bool_arg(page.get(\"show_scrollbars\") or False)).lower(),\ncumulusci/tasks/metadata_etl/layouts.py:493:        self._place_first = process_bool_arg(self.options.get(\"place_first\") or False)\ncumulusci/tasks/metadata_etl/picklists.py:77:                if process_bool_arg(x.get(\"default\", False))\ncumulusci/tasks/metadata_etl/picklists.py:177:        default = str(process_bool_arg(entry.get(\"default\", False))).lower\ncumulusci/tasks/metadeploy.py:74:        self.dry_run = process_bool_arg(self.options.get(\"dry_run\") or False)\ncumulusci/tasks/metadeploy.py:75:        self.publish = not self.dry_run and process_bool_arg(\ncumulusci/tasks/connectedapp.py:85:        self.options[\"connect\"] = process_bool_arg(self.options.get(\"connect\") or False)\ncumulusci/tasks/connectedapp.py:86:        self.options[\"overwrite\"] = process_bool_arg(\ncumulusci/tasks/salesforce/sourcetracking.py:57:        self.options[\"snapshot\"] = process_bool_arg(self.options.get(\"snapshot\", False))\ncumulusci/tasks/salesforce/sourcetracking.py:306:        self.options[\"snapshot\"] = process_bool_arg(kwargs.get(\"snapshot\", True))\ncumulusci/tasks/salesforce/UninstallPackage.py:22:        self.options[\"purge_on_delete\"] = process_bool_arg(\ncumulusci/tasks/salesforce/UninstallLocalNamespacedBundles.py:32:        self.options[\"managed\"] = process_bool_arg(self.options.get(\"managed\", False))\ncumulusci/tasks/salesforce/UninstallLocalNamespacedBundles.py:35:        self.options[\"purge_on_delete\"] = process_bool_arg(\ncumulusci/tasks/salesforce/composite.py:87:            managed = process_bool_arg(self.options[\"managed\"])\ncumulusci/tasks/salesforce/BaseUninstallMetadata.py:19:        self.options[\"purge_on_delete\"] = process_bool_arg(\ncumulusci/tasks/salesforce/BaseUninstallMetadata.py:22:        self.options[\"dry_run\"] = process_bool_arg(self.options.get(\"dry_run\", False))\ncumulusci/tasks/salesforce/Deploy.py:67:        self.check_only = process_bool_arg(self.options.get(\"check_only\", False))\ncumulusci/tasks/salesforce/Deploy.py:124:            return not process_bool_arg(self.options.get(\"unmanaged\", True))\ncumulusci/tasks/salesforce/Deploy.py:129:            return process_bool_arg(self.options.get(\"namespaced_org\", False))\ncumulusci/tasks/salesforce/Deploy.py:140:            \"clean_meta_xml\": process_bool_arg(\ncumulusci/tasks/salesforce/insert_record.py:26:        self.use_tooling = process_bool_arg(self.options.get(\"tooling\", False))\ncumulusci/tasks/salesforce/enable_prediction.py:41:            self.options[\"managed\"] = proc",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "cumulusci/tasks/github/merge.py:53:            self.options.get(\"skip_future_releases\") or True\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "stdout": "Case 1 (not provided): skip_future_releases = True (Expected: True)\nCase 2 (provided True): skip_future_releases = True (Expected: True)\nCase 3 (provided False): skip_future_releases = True (Expected: False)\nCase 4 (provided 'False'): skip_future_releases = False (Expected: False)\n",
  "stderr": "",
  "exit_code": 0
}