codeship: Strengthen types using WildValue.

Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg 2021-12-16 22:03:22 -08:00 committed by Tim Abbott
parent d5a8e040da
commit be781f19e6
1 changed files with 12 additions and 11 deletions

View File

@ -1,11 +1,10 @@
# Webhooks for external integrations.
from typing import Any, Dict
from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -27,7 +26,7 @@ CODESHIP_STATUS_MAPPER = {
def api_codeship_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: Dict[str, Any] = REQ(argument_type="body"),
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
) -> HttpResponse:
payload = payload["build"]
subject = get_subject_for_http_request(payload)
@ -37,21 +36,23 @@ def api_codeship_webhook(
return json_success(request)
def get_subject_for_http_request(payload: Dict[str, Any]) -> str:
return CODESHIP_TOPIC_TEMPLATE.format(project_name=payload["project_name"])
def get_subject_for_http_request(payload: WildValue) -> str:
return CODESHIP_TOPIC_TEMPLATE.format(
project_name=payload["project_name"].tame(check_string),
)
def get_body_for_http_request(payload: Dict[str, Any]) -> str:
def get_body_for_http_request(payload: WildValue) -> str:
return CODESHIP_MESSAGE_TEMPLATE.format(
build_url=payload["build_url"],
committer=payload["committer"],
branch=payload["branch"],
build_url=payload["build_url"].tame(check_string),
committer=payload["committer"].tame(check_string),
branch=payload["branch"].tame(check_string),
status=get_status_message(payload),
)
def get_status_message(payload: Dict[str, Any]) -> str:
build_status = payload["status"]
def get_status_message(payload: WildValue) -> str:
build_status = payload["status"].tame(check_string)
return CODESHIP_STATUS_MAPPER.get(
build_status, CODESHIP_DEFAULT_STATUS.format(status=build_status)
)