Source code for tcms.issuetracker.bitbucket

# -*- coding: utf-8 -*-
import requests
from requests.auth import HTTPBasicAuth

from tcms.core.contrib.linkreference.models import LinkReference
from tcms.issuetracker.base import IssueTrackerType


class BitBucketAPI:
    """
    BitBucket API interaction class.

    :meta private:
    """

    def __init__(self, base_url=None, api_username=None, api_password=None):
        api_version = "2.0"
        self.endpoint_url = self._construct_endpoint_url(api_version, base_url)
        self.headers = {
            "Accept": "application/json",
            "Content-type": "application/json",
        }
        self.auth = HTTPBasicAuth(api_username, api_password)

    def create_issue(self, data):
        url = f"{self.endpoint_url}/issues"
        return self._request(
            "POST", url, headers=self.headers, auth=self.auth, json=data
        )

    def get_issue(self, issue_id):
        url = f"{self.endpoint_url}/issues/{issue_id}"
        return self._request("GET", url, headers=self.headers, auth=self.auth)

    def update_issue(self, issue_id, data):
        url = f"{self.endpoint_url}/issues/{issue_id}/changes"
        return self._request(
            "POST", url, headers=self.headers, auth=self.auth, json=data
        )

    def add_comment(self, issue_id, comment):
        url = f"{self.endpoint_url}/issues/{issue_id}/comments/"
        return self._request(
            "POST", url, headers=self.headers, auth=self.auth, json=comment
        )

    def get_comments(self, issue_id):
        url = f"{self.endpoint_url}/issues/{issue_id}/comments?sort=-updated_on"
        return self._request("GET", url, headers=self.headers, auth=self.auth)

    def delete_comment(self, issue_id, comment_id):
        url = f"{self.endpoint_url}/issues/{issue_id}/comments/{comment_id}"
        return self._request("DELETE", url, headers=self.headers, auth=self.auth)

    @staticmethod
    def _request(method, url, **kwargs):
        if method == "DELETE":
            return requests.request(method, url, timeout=30, **kwargs)
        return requests.request(method, url, timeout=30, **kwargs).json()

    @staticmethod
    def _construct_endpoint_url(api_version, url):
        splitted_url = url.replace("https://", "").split("/")
        base_url = "https://api.bitbucket.org"
        workspace = splitted_url[1]
        repository = splitted_url[2]
        endpoint_url = f"{base_url}/{api_version}/repositories/{workspace}/{repository}"
        return endpoint_url


[docs] class BitBucket(IssueTrackerType): """ Support for BitBucket. Requires: :base_url: Repository URL - e.g. https://bitbucket.org/{workspace}/{repository} :api_username: BitBucket Username :api_password: BitBucket App Password - needs Issues: Read & write permission. .. note:: You can leave the ``api_url`` field blank because the integration code doesn't use it! .. warning:: ``api_username`` is your BitBucket username, which you use to log in. .. note:: ``api_password`` is "App Password" created in BitBucket. Here is a guide about creating and using an "App Password"; https://support.atlassian.com/bitbucket-cloud/docs/app-passwords/ """ def _rpc_connection(self): (api_username, api_password) = self.rpc_credentials return BitBucketAPI( self.bug_system.base_url, api_username=api_username, api_password=api_password, )
[docs] def is_adding_testcase_to_issue_disabled(self): (api_username, api_password) = self.rpc_credentials return not (self.bug_system.base_url and api_username and api_password)
def _report_issue(self, execution, user): """ BitBucket creates the Issue with Title and Description """ data = { "title": f"Failed test: {execution.case.summary}", "kind": "bug", "priority": "major", "content": { "raw": self._report_comment(execution, user).replace("\n", "\r\n") }, } try: issue = self.rpc.create_issue(data) issue_url = f"{self.bug_system.base_url}/issues/{issue['id']}" # add a link reference that will be shown in the UI LinkReference.objects.get_or_create( execution=execution, url=issue_url, is_defect=True, ) return (issue, issue_url) except Exception: # pylint: disable=broad-except # something above didn't work so return a link for manually # entering issue details with info pre-filled url = self.bug_system.base_url if not url.endswith("/"): url += "/" return (None, url + "issues/new")
[docs] def post_comment(self, execution, bug_id): comment_body = {"content": {"raw": self.text(execution).replace("\n", "\n\n")}} self.rpc.add_comment(bug_id, comment_body)
[docs] def details(self, url): """ Return issue details from BitBucket """ issue = self.rpc.get_issue(self.bug_id_from_url(url)) return { "title": issue["title"], "description": issue["content"]["raw"], }