Source code for galaxy.tools.error_reports.plugins.github

"""The module describes the ``github`` error plugin plugin."""

import logging
from urllib.parse import urlparse

from galaxy.util import (
    string_as_bool,
    unicodify,
)
from .base_git import BaseGitPlugin

log = logging.getLogger(__name__)


[docs]class GithubPlugin(BaseGitPlugin):
    """Send error report to GitHub."""

    plugin_type = "github"

[docs]    def __init__(self, **kwargs):
        self.app = kwargs["app"]
        self.redact_user_details_in_bugreport = self.app.config.redact_user_details_in_bugreport
        self.verbose = string_as_bool(kwargs.get("verbose", False))
        self.user_submission = string_as_bool(kwargs.get("user_submission", False))

        # GitHub settings
        self.github_base_url = kwargs.get("github_base_url", "https://github.com")
        self.github_api_url = kwargs.get("github_api_url", "https://api.github.com")
        self.git_default_repo_owner = kwargs.get("github_default_repo_owner", False)
        self.git_default_repo_name = kwargs.get("github_default_repo_name", False)
        self.git_default_repo_only = string_as_bool(kwargs.get("github_default_repo_only", True))

        try:
            import github

            self.github = github.Github(
                kwargs["github_oauth_token"],
                # Allow running against GH enterprise deployments.
                base_url=self.github_api_url,
            )
            # Connect to the default repository and fill up the issue cache for it
            repo = self.github.get_repo(f"{self.git_default_repo_owner}/{self.git_default_repo_name}")
            self._fill_issue_cache(repo, "default")

            # We'll also cache labels which we'll use for tagging issues.
            self.label_cache["default"] = {}
            self._fill_label_cache(repo, "default")
        except ImportError:
            log.error("Please install pygithub to submit bug reports to GitHub")
            self.github = None

[docs]    def submit_report(self, dataset, job, tool, **kwargs):
        """Submit the error report to sentry"""
        log.info(self.github)

        if self.github:
            # Determine the ToolShed url, initially we connect with HTTP and if redirect to HTTPS is set up,
            # this will be detected by requests and used further down the line. Also cache this so everything is
            # as fast as possible
            log.info(tool.tool_shed)
            ts_url = self._determine_ts_url(tool)
            log.info("GitLab error reporting - Determined ToolShed is %s", ts_url)

            # Find the repo inside the ToolShed
            ts_repourl = self._get_gitrepo_from_ts(job, ts_url)

            # Determine the GitLab project URL and the issue cache key
            github_projecturl = (
                urlparse(ts_repourl).path[1:]
                if (ts_repourl and not self.git_default_repo_only)
                else "/".join((self.git_default_repo_owner, self.git_default_repo_name))
            )
            issue_cache_key = self._get_issue_cache_key(job, ts_repourl)

            # Connect to the repo
            if github_projecturl not in self.git_project_cache:
                self.git_project_cache[github_projecturl] = self.github.get_repo(f"{github_projecturl}")
            gh_project = self.git_project_cache[github_projecturl]

            # Make sure we keep a cache of the issues, per tool in this case
            if issue_cache_key not in self.issue_cache:
                self._fill_issue_cache(gh_project, issue_cache_key)

            # Retrieve label
            label = self.get_label(
                f"{unicodify(job.tool_id)}/{unicodify(job.tool_version)}", gh_project, issue_cache_key
            )

            # Generate information for the tool
            error_title = self._generate_error_title(job)

            # Generate the error message
            error_message = self._generate_error_message(dataset, job, kwargs)

            log.info(error_title in self.issue_cache[issue_cache_key])
            if error_title not in self.issue_cache[issue_cache_key]:
                # Create a new issue.
                self._create_issue(issue_cache_key, error_title, error_message, gh_project, label=label)
            else:
                self._append_issue(issue_cache_key, error_title, error_message)
            return (
                "Submitted error report to GitHub. Your issue number is [#{}]({}/{}/issues/{})".format(
                    self.issue_cache[issue_cache_key][error_title].number,
                    self.github_base_url,
                    github_projecturl,
                    self.issue_cache[issue_cache_key][error_title].number,
                ),
                "success",
            )

    def _create_issue(self, issue_cache_key, error_title, error_mesage, project, **kwargs):
        # Create a new issue.
        self.issue_cache[issue_cache_key][error_title] = project.create_issue(
            title=error_title,
            body=error_mesage,
            # Label it with a tag: tool_id/tool_version
            labels=[kwargs.get("label")],
        )

    def _append_issue(self, issue_cache_key, error_title, error_message, **kwargs):
        # Create comment on an issue
        self.issue_cache[issue_cache_key][error_title].create_comment(error_message)

    def _fill_issue_cache(self, git_project, issue_cache_key):
        # We want to ensure that we don't generate a thousand issues when
        # multiple users report a bug. So, we need to de-dupe issues. In
        # order to de-dupe, we need to know which are open. So, we'll keep
        # a cache of open issues and just add to it whenever we create a
        # new one.
        self.issue_cache[issue_cache_key] = {}
        for issue in git_project.get_issues(state="open"):
            log.info(issue)
            self.issue_cache[issue_cache_key][issue.title] = issue
        log.info(self.issue_cache)

    def _fill_label_cache(self, git_project, issue_cache_key):
        for label in git_project.get_labels():
            log.info(label)
            self.label_cache[issue_cache_key][label.name] = label
        log.info(self.label_cache)

[docs]    def get_label(self, label, git_project, issue_cache_key):
        # If we don't have this label, then create it + cache it.
        if issue_cache_key not in self.label_cache:
            self.label_cache[issue_cache_key] = {}
            self._fill_label_cache(git_project, issue_cache_key)

        if label not in self.label_cache[issue_cache_key]:
            self.label_cache[issue_cache_key][label] = git_project.create_label(name=label, color="ffffff")

        return self.label_cache[issue_cache_key][label]


__all__ = ("GithubPlugin",)