import os
import requests
from requests.auth import HTTPBasicAuth
[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):
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
if auth_type == 'basic':
self.auth = HTTPBasicAuth(self.account, self.password)
else:
self.auth = None
self.url_scheme = 'http' if auth_type == 'authless' else 'https'
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 = requests.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 = requests.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 = requests.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 = requests.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 = requests.put(url, auth=self.auth, json=annotation_data, proxies=self._get_http_proxies())
if response.ok:
return response.json()
response.raise_for_status()