Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity
zulip/zerver/webhooks/bitbucket3/view.py

373 lines
16 KiB
Python
Raw Blame History

import string
from functools import partial
from inspect import signature
from typing import Any, Callable, Dict, List, Optional
from django.http import HttpRequest, HttpResponse
from zerver.decorator import api_key_only_webhook_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.webhooks.common import UnexpectedWebhookEventType, \
check_send_webhook_message
from zerver.lib.webhooks.git import CONTENT_MESSAGE_TEMPLATE, \
TOPIC_WITH_BRANCH_TEMPLATE, TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE, \
get_commits_comment_action_message, get_create_branch_event_message, \
get_pull_request_event_message, get_push_tag_event_message, \
get_remove_branch_event_message
from zerver.models import UserProfile
from zerver.webhooks.bitbucket2.view import BITBUCKET_FORK_BODY, \
BITBUCKET_REPO_UPDATED_CHANGED, BITBUCKET_TOPIC_TEMPLATE
BRANCH_UPDATED_MESSAGE_TEMPLATE = "{user_name} pushed to branch {branch_name}. Head is now {head}."
PULL_REQUEST_MARKED_AS_NEEDS_WORK_TEMPLATE = "{user_name} marked [PR #{number}]({url}) as \"needs work\"."
PULL_REQUEST_MARKED_AS_NEEDS_WORK_TEMPLATE_WITH_TITLE = """
{user_name} marked [PR #{number} {title}]({url}) as \"needs work\".
""".strip()
PULL_REQUEST_REASSIGNED_TEMPLATE = "{user_name} reassigned [PR #{number}]({url}) to {assignees}."
PULL_REQUEST_REASSIGNED_TEMPLATE_WITH_TITLE = """
{user_name} reassigned [PR #{number} {title}]({url}) to {assignees}.
""".strip()
PULL_REQUEST_REASSIGNED_TO_NONE_TEMPLATE = "{user_name} removed all reviewers from [PR #{number}]({url})."
PULL_REQUEST_REASSIGNED_TO_NONE_TEMPLATE_WITH_TITLE = """
{user_name} removed all reviewers from [PR #{number} {title}]({url})
""".strip()
PULL_REQUEST_OPENED_OR_MODIFIED_TEMPLATE_WITH_REVIEWERS = """
{user_name} {action} [PR #{number}]({url}) from `{source}` to \
`{destination}` (assigned to {assignees} for review)
""".strip()
PULL_REQUEST_OPENED_OR_MODIFIED_TEMPLATE_WITH_REVIEWERS_WITH_TITLE = """
{user_name} {action} [PR #{number} {title}]({url}) from `{source}` to \
`{destination}` (assigned to {assignees} for review)
""".strip()
def fixture_to_headers(fixture_name: str) -> Dict[str, str]:
if fixture_name == "diagnostics_ping":
return {"HTTP_X_EVENT_KEY": "diagnostics:ping"}
return dict()
def get_user_name(payload: Dict[str, Any]) -> str:
user_name = "[{name}]({url})".format(name=payload["actor"]["name"],
url=payload["actor"]["links"]["self"][0]["href"])
return user_name
def ping_handler(payload: Dict[str, Any], include_title: Optional[str]=None
) -> List[Dict[str, str]]:
if include_title:
subject = include_title
else:
subject = "Bitbucket Server Ping"
body = "Congratulations! The Bitbucket Server webhook was configured successfully!"
return [{"subject": subject, "body": body}]
def repo_comment_handler(payload: Dict[str, Any], action: str) -> List[Dict[str, str]]:
repo_name = payload["repository"]["name"]
subject = BITBUCKET_TOPIC_TEMPLATE.format(repository_name=repo_name)
sha = payload["commit"]
commit_url = payload["repository"]["links"]["self"][0]["href"][:-6] # remove the "browse" at the end
commit_url += "commits/%s" % (sha,)
message = payload["comment"]["text"]
if action == "deleted their comment":
message = "~~{message}~~".format(message=message)
body = get_commits_comment_action_message(
user_name=get_user_name(payload),
action=action,
commit_url=commit_url,
sha=sha,
message=message
)
return [{"subject": subject, "body": body}]
def repo_forked_handler(payload: Dict[str, Any]) -> List[Dict[str, str]]:
repo_name = payload["repository"]["origin"]["name"]
subject = BITBUCKET_TOPIC_TEMPLATE.format(repository_name=repo_name)
body = BITBUCKET_FORK_BODY.format(
display_name=payload["actor"]["displayName"],
username=get_user_name(payload),
fork_name=payload["repository"]["name"],
fork_url=payload["repository"]["links"]["self"][0]["href"]
)
return [{"subject": subject, "body": body}]
def repo_modified_handler(payload: Dict[str, Any]) -> List[Dict[str, str]]:
subject_new = BITBUCKET_TOPIC_TEMPLATE.format(repository_name=payload["new"]["name"])
new_name = payload['new']['name']
body = BITBUCKET_REPO_UPDATED_CHANGED.format(
actor=get_user_name(payload),
change="name",
repo_name=payload["old"]["name"],
old=payload["old"]["name"],
new=new_name
) # As of writing this, the only change we'd be notified about is a name change.
punctuation = '.' if new_name[-1] not in string.punctuation else ''
body = "{}{}".format(body, punctuation)
return [{"subject": subject_new, "body": body}]
def repo_push_branch_data(payload: Dict[str, Any], change: Dict[str, Any]) -> Dict[str, str]:
event_type = change["type"]
repo_name = payload["repository"]["name"]
user_name = get_user_name(payload)
branch_name = change["ref"]["displayId"]
branch_head = change["toHash"]
if event_type == "ADD":
body = get_create_branch_event_message(
user_name=user_name,
url=None,
branch_name=branch_name
)
elif event_type == "UPDATE":
body = BRANCH_UPDATED_MESSAGE_TEMPLATE.format(
user_name=user_name,
branch_name=branch_name,
head=branch_head
)
elif event_type == "DELETE":
body = get_remove_branch_event_message(user_name, branch_name)
else:
message = "%s.%s" % (payload["eventKey"], event_type) # nocoverage
raise UnexpectedWebhookEventType("BitBucket Server", message)
subject = TOPIC_WITH_BRANCH_TEMPLATE.format(repo=repo_name, branch=branch_name)
return {"subject": subject, "body": body}
def repo_push_tag_data(payload: Dict[str, Any], change: Dict[str, Any]) -> Dict[str, str]:
event_type = change["type"]
repo_name = payload["repository"]["name"]
tag_name = change["ref"]["displayId"]
if event_type == "ADD":
action = "pushed"
elif event_type == "DELETE":
action = "removed"
else:
message = "%s.%s" % (payload["eventKey"], event_type) # nocoverage
raise UnexpectedWebhookEventType("BitBucket Server", message)
subject = BITBUCKET_TOPIC_TEMPLATE.format(repository_name=repo_name)
body = get_push_tag_event_message(get_user_name(payload), tag_name, action=action)
return {"subject": subject, "body": body}
def repo_push_handler(payload: Dict[str, Any], branches: Optional[str]=None
) -> List[Dict[str, str]]:
data = []
for change in payload["changes"]:
event_target_type = change["ref"]["type"]
if event_target_type == "BRANCH":
branch = change["ref"]["displayId"]
if branches:
if branch not in branches:
continue
data.append(repo_push_branch_data(payload, change))
elif event_target_type == "TAG":
data.append(repo_push_tag_data(payload, change))
else:
message = "%s.%s" % (payload["eventKey"], event_target_type) # nocoverage
raise UnexpectedWebhookEventType("BitBucket Server", message)
return data
def get_assignees_string(pr: Dict[str, Any]) -> Optional[str]:
reviewers = []
for reviewer in pr["reviewers"]:
name = reviewer["user"]["name"]
link = reviewer["user"]["links"]["self"][0]["href"]
reviewers.append("[%s](%s)" % (name, link))
if len(reviewers) == 0:
assignees = None
elif len(reviewers) == 1:
assignees = reviewers[0]
else:
assignees = ", ".join(reviewers[:-1]) + " and " + reviewers[-1]
return assignees
def get_pr_subject(repo: str, type: str, id: str, title: str) -> str:
return TOPIC_WITH_PR_OR_ISSUE_INFO_TEMPLATE.format(repo=repo, type=type, id=id, title=title)
def get_simple_pr_body(payload: Dict[str, Any], action: str, include_title: Optional[bool]) -> str:
pr = payload["pullRequest"]
return get_pull_request_event_message(
user_name=get_user_name(payload),
action=action,
url=pr["links"]["self"][0]["href"],
number=pr["id"],
title=pr["title"] if include_title else None
)
def get_pr_opened_or_modified_body(payload: Dict[str, Any], action: str,
include_title: Optional[bool]) -> str:
pr = payload["pullRequest"]
description = pr.get("description")
assignees_string = get_assignees_string(pr)
if assignees_string:
# Then use the custom message template for this particular integration so that we can
# specify the reviewers at the end of the message (but before the description/message).
parameters = {"user_name": get_user_name(payload),
"action": action,
"url": pr["links"]["self"][0]["href"],
"number": pr["id"],
"source": pr["fromRef"]["displayId"],
"destination": pr["toRef"]["displayId"],
"message": description,
"assignees": assignees_string,
"title": pr["title"] if include_title else None}
if include_title:
body = PULL_REQUEST_OPENED_OR_MODIFIED_TEMPLATE_WITH_REVIEWERS_WITH_TITLE.format(
**parameters
)
else:
body = PULL_REQUEST_OPENED_OR_MODIFIED_TEMPLATE_WITH_REVIEWERS.format(**parameters)
punctuation = ':' if description else '.'
body = "{}{}".format(body, punctuation)
if description:
body += '\n' + CONTENT_MESSAGE_TEMPLATE.format(message=description)
return body
return get_pull_request_event_message(
user_name=get_user_name(payload),
action=action,
url=pr["links"]["self"][0]["href"],
number=pr["id"],
target_branch=pr["fromRef"]["displayId"],
base_branch=pr["toRef"]["displayId"],
message=pr.get("description"),
assignee=assignees_string if assignees_string else None,
title=pr["title"] if include_title else None
)
def get_pr_needs_work_body(payload: Dict[str, Any], include_title: Optional[bool]) -> str:
pr = payload["pullRequest"]
if not include_title:
return PULL_REQUEST_MARKED_AS_NEEDS_WORK_TEMPLATE.format(
user_name=get_user_name(payload),
number=pr["id"],
url=pr["links"]["self"][0]["href"]
)
return PULL_REQUEST_MARKED_AS_NEEDS_WORK_TEMPLATE_WITH_TITLE.format(
user_name=get_user_name(payload),
number=pr["id"],
url=pr["links"]["self"][0]["href"],
title=pr["title"]
)
def get_pr_reassigned_body(payload: Dict[str, Any], include_title: Optional[bool]) -> str:
pr = payload["pullRequest"]
assignees_string = get_assignees_string(pr)
if not assignees_string:
if not include_title:
return PULL_REQUEST_REASSIGNED_TO_NONE_TEMPLATE.format(
user_name=get_user_name(payload),
number=pr["id"],
url=pr["links"]["self"][0]["href"]
)
punctuation = '.' if pr['title'][-1] not in string.punctuation else ''
message = PULL_REQUEST_REASSIGNED_TO_NONE_TEMPLATE_WITH_TITLE.format(
user_name=get_user_name(payload),
number=pr["id"],
url=pr["links"]["self"][0]["href"],
title=pr["title"]
)
message = "{}{}".format(message, punctuation)
return message
if not include_title:
return PULL_REQUEST_REASSIGNED_TEMPLATE.format(
user_name=get_user_name(payload),
number=pr["id"],
url=pr["links"]["self"][0]["href"],
assignees=assignees_string
)
return PULL_REQUEST_REASSIGNED_TEMPLATE_WITH_TITLE.format(
user_name=get_user_name(payload),
number=pr["id"],
url=pr["links"]["self"][0]["href"],
assignees=assignees_string,
title=pr["title"]
)
def pr_handler(payload: Dict[str, Any], action: str,
include_title: Optional[bool]=False) -> List[Dict[str, str]]:
pr = payload["pullRequest"]
subject = get_pr_subject(pr["toRef"]["repository"]["name"], type="PR", id=pr["id"],
title=pr["title"])
if action in ["opened", "modified"]:
body = get_pr_opened_or_modified_body(payload, action, include_title)
elif action == "needs_work":
body = get_pr_needs_work_body(payload, include_title)
elif action == "reviewers_updated":
body = get_pr_reassigned_body(payload, include_title)
else:
body = get_simple_pr_body(payload, action, include_title)
return [{"subject": subject, "body": body}]
def pr_comment_handler(payload: Dict[str, Any], action: str,
include_title: Optional[bool]=False) -> List[Dict[str, str]]:
pr = payload["pullRequest"]
subject = get_pr_subject(pr["toRef"]["repository"]["name"], type="PR", id=pr["id"],
title=pr["title"])
message = payload["comment"]["text"]
if action == "deleted their comment on":
message = "~~{message}~~".format(message=message)
body = get_pull_request_event_message(
user_name=get_user_name(payload),
action=action,
url=pr["links"]["self"][0]["href"],
number=pr["id"],
message=message,
title=pr["title"] if include_title else None
)
return [{"subject": subject, "body": body}]
EVENT_HANDLER_MAP = {
"diagnostics:ping": ping_handler,
"repo:comment:added": partial(repo_comment_handler, action="commented"),
"repo:comment:edited": partial(repo_comment_handler, action="edited their comment"),
"repo:comment:deleted": partial(repo_comment_handler, action="deleted their comment"),
"repo:forked": repo_forked_handler,
"repo:modified": repo_modified_handler,
"repo:refs_changed": repo_push_handler,
"pr:comment:added": partial(pr_comment_handler, action="commented on"),
"pr:comment:edited": partial(pr_comment_handler, action="edited their comment on"),
"pr:comment:deleted": partial(pr_comment_handler, action="deleted their comment on"),
"pr:declined": partial(pr_handler, action="declined"),
"pr:deleted": partial(pr_handler, action="deleted"),
"pr:merged": partial(pr_handler, action="merged"),
"pr:modified": partial(pr_handler, action="modified"),
"pr:opened": partial(pr_handler, action="opened"),
"pr:reviewer:approved": partial(pr_handler, action="approved"),
"pr:reviewer:needs_work": partial(pr_handler, action="needs_work"),
"pr:reviewer:updated": partial(pr_handler, action="reviewers_updated"),
"pr:reviewer:unapproved": partial(pr_handler, action="unapproved"),
} # type Dict[str, Optional[Callable[..., List[Dict[str, str]]]]]
def get_event_handler(eventkey: str) -> Callable[..., List[Dict[str, str]]]:
# The main reason for this function existance is because of mypy
handler = EVENT_HANDLER_MAP.get(eventkey) # type: Any
if handler is None:
raise UnexpectedWebhookEventType("BitBucket Server", eventkey)
return handler
@api_key_only_webhook_view("Bitbucket3")
@has_request_variables
def api_bitbucket3_webhook(request: HttpRequest, user_profile: UserProfile,
payload: Dict[str, Any]=REQ(argument_type="body"),
branches: Optional[str]=REQ(default=None),
user_specified_topic: Optional[str]=REQ("topic", default=None)
) -> HttpResponse:
try:
eventkey = payload["eventKey"]
except KeyError:
eventkey = request.META["HTTP_X_EVENT_KEY"]
handler = get_event_handler(eventkey)
if "branches" in signature(handler).parameters:
data = handler(payload, branches)
elif "include_title" in signature(handler).parameters:
data = handler(payload, include_title=user_specified_topic)
else:
data = handler(payload)
for element in data:
check_send_webhook_message(request, user_profile, element["subject"],
element["body"], unquote_url_parameters=True)
return json_success()
Reference in New Issue View Git Blame Copy Permalink