sfdo-tooling__cumulusci-3624

{
  "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\": {\n./cumulusci/tasks/github/merge.py:52:        self.options[\"skip_future_releases\"] = process_bool_arg(\n./cumulusci/tasks/github/merge.py:53:            self.options.get(\"skip_future_releases\") or True\n./cumulusci/tasks/github/merge.py:108:        skip_future_releases = self.options[\"skip_future_releases\"]\n./cumulusci/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": "./cumulusci/core/utils.py:53:def process_bool_arg(arg: T.Union[int, str, None]):\n",
  "stderr": "",
  "exit_code": 0
}

{
  "status": "ok",
  "filepath": "cumulusci/core/utils.py",
  "content": "    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\",",
  "start_line": 50,
  "end_line": 70,
  "total_lines": 364,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "cumulusci/core/utils.py",
  "content": "            \"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": 70,
  "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",
  "start_line": 1,
  "end_line": 90,
  "total_lines": 364,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "cumulusci/tasks/github/base.py",
  "content": "from cumulusci.core.github import get_github_api_for_repo\nfrom cumulusci.core.tasks import BaseTask\n\n\nclass BaseGithubTask(BaseTask):\n    def _init_task(self):\n        super()._init_task()\n        self.github_config = self.project_config.keychain.get_service(\"github\")\n        self.github = get_github_api_for_repo(\n            self.project_config.keychain, self.project_config.repo_url\n        )\n\n    def get_repo(self):\n        return self.github.repository(\n            self.project_config.repo_owner, self.project_config.repo_name\n        )",
  "start_line": 1,
  "end_line": 16,
  "total_lines": 16,
  "is_truncated": false
}

{
  "status": "ok",
  "filepath": "cumulusci/core/tasks.py",
  "content": "\"\"\" Tasks are the basic unit of execution in CumulusCI.\n\nSubclass BaseTask or a descendant to define custom task logic\n\"\"\"\nimport contextlib\nimport logging\nimport os\nimport re\nimport threading\nimport time\nfrom contextlib import nullcontext\nfrom typing import Any, Callable, Dict, List, Optional, Type, Union\n\nfrom pydantic.error_wrappers import ValidationError\n\nfrom cumulusci import __version__\nfrom cumulusci.core.config import TaskConfig\nfrom cumulusci.core.config.org_config import OrgConfig\nfrom cumulusci.core.config.project_config import BaseProjectConfig\nfrom cumulusci.core.debug import DebugMode, get_debug_mode\nfrom cumulusci.core.exceptions import (\n    ServiceNotConfigured,\n    ServiceNotValid,\n    TaskOptionsError,\n    TaskRequiresSalesforceOrg,\n)\nfrom cumulusci.core.flowrunner import FlowCoordinator, StepSpec, StepVersion\nfrom cumulusci.utils import cd\nfrom cumulusci.utils.logging import redirect_output_to_logger\nfrom cumulusci.utils.metaprogramming import classproperty\nfrom cumulusci.utils.options import CCIOptions\n\nCURRENT_TASK = threading.local()\n\nPROJECT_CONFIG_RE = re.compile(r\"\\$project_config.(\\w+)\")\nCAPTURE_TASK_OUTPUT = os.environ.get(\"CAPTURE_TASK_OUTPUT\")\n\n\n@contextlib.contextmanager\ndef stacked_task(task):\n    if not hasattr(CURRENT_TASK, \"stack\"):\n        CURRENT_TASK.stack = []\n    CURRENT_TASK.stack.append(task)\n    try:\n        yield\n    finally:\n        CURRENT_TASK.stack.pop()\n\n\nclass BaseTask:\n    \"\"\"BaseTask provides the core execution logic for a Task\n\n    Subclass BaseTask and provide a `_run_task()` method with your\n    code.\n    \"\"\"\n\n    task_docs: str = \"\"\n    Options: Optional[Type[CCIOptions]] = None\n    salesforce_task: bool = False  # Does this task require a salesforce org?\n    name: Optional[str]\n    stepnum: Optional[StepVersion]\n    result: Any\n    return_values: dict\n    debug_mode: DebugMode\n    logger: logging.Logger\n    options: dict\n\n    poll_complete: bool\n    poll_count: int\n    poll_interval_level: int\n    poll_interval_s: int\n\n    def __init__(\n        self,\n        project_config: BaseProjectConfig,\n        task_config: TaskConfig,\n        org_config: Optional[OrgConfig] = None,\n        flow: Optional[FlowCoordinator] = None,\n        name: Optional[str] = None,\n        stepnum: Optional[StepVersion] = None,\n        logger: Optional[logging.Logger] = None,\n        **kwargs,\n    ):\n        self.project_config = project_config\n        self.task_config = task_config\n        self.org_config = org_config\n\n        self._reset_poll()\n\n        # dict of return_values that can be used by task callers\n        self.return_values = {}\n        # simple result object for introspection, often a return_code\n        self.result = None\n        # the flow for this task execution\n        self.flow = flow\n        # the task's name in the flow\n        self.name = name\n        # the task's stepnumber in the flow\n        self.stepnum = stepnum\n\n        self.debug_mode = get_debug_mode()\n        if logger:\n            self.logger = logger\n        else:\n            self._init_logger()\n\n        self._init_options(kwargs)\n        self._validate_options()\n\n    def _init_logger(self):\n        \"\"\"Initializes self.logger\"\"\"\n        if self.flow:\n            self.logger = self.flow.logger.getChild(self.__class__.__name__)\n        else:\n            self.logger = logging.getLogger(__name__)\n\n    @classproperty\n    def task_options(cls):  # type: ignore  -- doesn't like the signature override below\n        \"Convert Options dict into old fashioned task_options syntax\"\n        if cls.Options:\n            return cls.Options.as_task_options()\n        else:\n            return {}\n\n    task_options: Union[classproperty, Dict]\n\n    def _init_options(self, kwargs):\n        \"\"\"Initializes self.options\"\"\"\n        if self.task_config.options is None:\n            self.options = {}\n        else:\n            self.options = self.task_config.options.copy()\n\n        if kwargs:\n            self.options.update(kwargs)\n\n        # Handle dynamic lookup of project_config values via $project_config.attr\n        for option, value in self.options.items():\n            if isinstance(value, str):\n                value = PROJECT_CONFIG_RE.sub(\n                    lambda match: str(self.project_config.lookup(match.group(1), None)),\n                    value,\n                )\n                self.options[option] = value\n\n        if self.Options:\n            try:\n                specials = [\"debug_before\", \"debug_after\", \"no_prompt\"]\n                options_without_specials = {\n                    opt: val for opt, val in self.options.items() if opt not in specials",
  "start_line": 1,
  "end_line": 150,
  "total_lines": 390,
  "is_truncated": true
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}

{
  "error": "Invoking `write_file()` failed as the following mandatory input parameters are not present:\nfilepath\nYou could retry calling this tool, but it is IMPORTANT for you to provide all the mandatory parameters."
}