zulip/zerver/webhooks/bitbucket/view.py

from typing import Any, Mapping, Optional

from django.http import HttpRequest, HttpResponse

from zerver.decorator import authenticated_rest_api_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import check_dict
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import TOPIC_WITH_BRANCH_TEMPLATE, get_push_commits_event_message
from zerver.models import UserProfile


@authenticated_rest_api_view(webhook_client_name="Bitbucket")
@has_request_variables
def api_bitbucket_webhook(
    request: HttpRequest,
    user_profile: UserProfile,
    payload: Mapping[str, Any] = REQ(validator=check_dict([])),
    branches: Optional[str] = REQ(default=None),
) -> HttpResponse:
    repository = payload["repository"]

    commits = [
        {
            "name": commit.get("author") or payload.get("user"),
            "sha": commit.get("raw_node"),
            "message": commit.get("message"),
            "url": "{}{}commits/{}".format(
                payload.get("canon_url"), repository.get("absolute_url"), commit.get("raw_node")
            ),
        }
        for commit in payload["commits"]
    ]

    if len(commits) == 0:
        # Bitbucket doesn't give us enough information to really give
        # a useful message :/
        subject = repository["name"]
        content = "{} [force pushed]({}).".format(
            payload.get("user", "Someone"),
            payload["canon_url"] + repository["absolute_url"],
        )
    else:
        branch = payload["commits"][-1]["branch"]
        if branches is not None and branches.find(branch) == -1:
            return json_success()

        committer = payload.get("user")
        content = get_push_commits_event_message(
            committer if committer is not None else "Someone", None, branch, commits
        )
        subject = TOPIC_WITH_BRANCH_TEMPLATE.format(repo=repository["name"], branch=branch)

    check_send_webhook_message(request, user_profile, subject, content, unquote_url_parameters=True)
    return json_success()