import json
import os
from requests.auth import HTTPBasicAuth
from bitbucket_pipes_toolkit.core import TokenAuth
from bitbucket_pipes_toolkit.helpers import HttpRequestsHandler
[docs]
class CodeInsights:
"""This is a wrapper class that uses Bitbucket CodeInsights API to create Code reports.
Args:
auth_type (str): The type of authentication mechanism (basic, authless)
auth_proxy_host (str): Hostname of the pipelines authentication proxy.
account (str): Account, that will be used to authenticate the API requests.
app_password (str): Bitbucket app password created for a user who owns the account.
username (str): The name of the account that owns the repository.
repo (str): Repository slug.
With the authless authentication type the pipe will proxy API requests via internal authentication proxy. In
that case username and password are not required. To create an annotation without authentication in
pipelines, pass the 'localhost' as the auth_proxy_host parameter.
"""
def __init__(self, repo, username, auth_type='authless', auth_proxy_host="host.docker.internal", app_password=None, account=None, token=None):
self.account = account
self.password = app_password
self.username = username
self.repo_slug = repo
self.auth_type = auth_type
self.auth_proxy_host = auth_proxy_host
self.token = token
if auth_type == 'basic':
self.auth = HTTPBasicAuth(self.account, self.password)
elif auth_type == 'access_token':
self.auth = TokenAuth(self.token)
else:
self.auth = None
self.url_scheme = 'http' if auth_type == 'authless' else 'https'
self.http_client = HttpRequestsHandler()
def _get_http_proxies(self):
if self.auth_type == 'authless' and os.getenv('CI') is not None:
return {"http": f"http://{self.auth_proxy_host}:29418", "https": f"https://{self.auth_proxy_host}:29418"}
return {}
[docs]
def create_report(self, commit, report_data):
"""Send a request to create a report for a given commit.
Args:
commit (str): Hash of the commit to create report for.
report_data (dict): Dictionary with the data to create a report.
Example of report data:
report_data = {
"type": "report",
"uuid": "{asdasd-565656-asdad-565655}",
"report_type": "BUG",
"external_id": "10", # required
"title": "Bug report", # required
"details": "This bug report is auto generated by bug tool.", # required
"result": "FAILED",
"reporter": "Created by atlassians bug tool.",
"link": "https://bug-tool.atlassian.com/report/10",
"logo_url": "https://bug-tool.atlassian.com/logo.png",
"data": [
{
"title": "FAILED",
"type": "BOOLEAN",
"value": true
},
],
"created_on": "2020-01-08T00:56:20.593Z",
"updated_on": "2020-01-09T12:00:10.123Z"
}
Returns:
json: Response json data.
"""
report_id = report_data.get('uuid', report_data.get('external_id'))
url = f'{self.url_scheme}://api.bitbucket.org/2.0/repositories/{self.username}/{self.repo_slug}/commit/{commit}/reports/{report_id}'
response = self.http_client.make_session_request("PUT", url, auth=self.auth, json=report_data, proxies=self._get_http_proxies())
if response.ok:
return response.json()
response.raise_for_status()
[docs]
def get_reports(self, commit):
"""Send a request to get all the reports for a given commit.
Args:
commit (str): Hash of the commit to get reports for.
Returns:
json: Response json data.
"""
url = f'{self.url_scheme}://api.bitbucket.org/2.0/repositories/{self.username}/{self.repo_slug}/commit/{commit}/reports'
response = self.http_client.make_session_request("GET", url, auth=self.auth, proxies=self._get_http_proxies())
if response.ok:
return response.json()
response.raise_for_status()
[docs]
def get_report(self, commit, report_id):
"""Send a request to get report for a given commit.
Args:
commit (str): Hash of the commit to get report for.
report_id (str): Report id.
Returns:
json: Response json data.
"""
url = f'{self.url_scheme}://api.bitbucket.org/2.0/repositories/{self.username}/{self.repo_slug}/commit/{commit}/reports/{report_id}'
response = self.http_client.make_session_request("GET", url, auth=self.auth, proxies=self._get_http_proxies())
if response.ok:
return response.json()
response.raise_for_status()
[docs]
def delete_report(self, commit, report_id):
"""Send a request to delete report for a given commit.
Args:
commit (str): Hash of the commit to delete report for.
report_id (str): Report id.
Returns:
bool: True if response is ok.
"""
url = f'{self.url_scheme}://api.bitbucket.org/2.0/repositories/{self.username}/{self.repo_slug}/commit/{commit}/reports/{report_id}'
response = self.http_client.make_session_request("DELETE", url, auth=self.auth, proxies=self._get_http_proxies())
if response.ok:
return True
response.raise_for_status()
[docs]
def create_annotation(self, commit, report_id, annotation_data):
"""Send a request to create an annotation for a given commit and report id.
Args:
commit (str): Hash of the commit to create annotation for.
report_id (str): Report id.
annotation_data (dict): Dictionary with the data to create an annotation.
Example of annotation data:
annotation_data = {
"type": "report_annotation",
"annotation_type": "BUG",
"uuid": "{asdasd-1231321-asdad-2131321}",
"external_id": "100", # required
"summary": "This line has a bug!", # required
"details": "This is a really bad bug, caused by IEEE09999, you need to ensure arrays dont go out of bounds.",
"path": "src/main/java/com/atlassian/pipelines/Service.java",
"line": 10,
"severity": "HIGH",
"result": "FAILED",
"link": "https://bug-tool.atlassian.com/report/10/bug/100",
"created_on": "2020-01-08T00:56:20.593Z",
"updated_on": "2020-01-09T12:00:10.123Z"
}
Returns:
json: Response json data.
"""
annotation_id = annotation_data.get('uuid', annotation_data.get('external_id'))
url = f'{self.url_scheme}://api.bitbucket.org/2.0/repositories/{self.username}/{self.repo_slug}/commit/{commit}/reports/{report_id}/annotations/{annotation_id}'
response = self.http_client.make_session_request("PUT", url, auth=self.auth, json=annotation_data, proxies=self._get_http_proxies())
if response.ok:
return response.json()
response.raise_for_status()
[docs]
def create_bulk_annotations(self, commit, report_id, annotations_data):
"""Send a request to create bulk annotations for a given report uuid.
Args:
commit (str): Hash of the commit to create annotation for.
report_id (str): Report id.
annotations_data (list[dict]): List of Dictionaries with the data to create an annotation.
Example of annotations data:
annotations_data = [{
"type": "report_annotation",
"annotation_type": "BUG",
"uuid": "{asdasd-1231321-asdad-2131321}",
"external_id": "100", # required
"summary": "This line has a bug!", # required
"details": "This is a really bad bug, caused by IEEE09999, you need to ensure arrays dont go out of bounds.",
"path": "src/main/java/com/atlassian/pipelines/Service.java",
"line": 10,
"severity": "HIGH",
"result": "FAILED",
"link": "https://bug-tool.atlassian.com/report/10/bug/100",
"created_on": "2020-01-08T00:56:20.593Z",
"updated_on": "2020-01-09T12:00:10.123Z"
}]
Returns:
json: Response json data (array of created annotations)
"""
url = f'{self.url_scheme}://api.bitbucket.org/2.0/repositories/{self.username}/{self.repo_slug}/commit/{commit}/reports/{report_id}/annotations'
headers = {
"Accept": "application/json",
"Content-Type": "application/json",
}
payload = json.dumps(annotations_data)
response = self.http_client.make_session_request(
"POST", url, auth=self.auth, data=payload, headers=headers, proxies=self._get_http_proxies())
if response.ok:
return response.json()
response.raise_for_status()