webhooks: Migrate travis to use @typed_endpoint.

To perform the same check, we define a Pydantic model. This includes
some keys "build_url" and "type" that we did not check for previously.
This commit is contained in:
Zixuan James Li 2023-08-14 17:54:59 -04:00 committed by Tim Abbott
parent 4037196fb2
commit 7d683018bd
1 changed files with 21 additions and 21 deletions

View File

@ -1,12 +1,12 @@
# Webhooks for external integrations. # Webhooks for external integrations.
from typing import Dict
from django.http import HttpRequest, HttpResponse from django.http import HttpRequest, HttpResponse
from pydantic import BaseModel, Json
from typing_extensions import Annotated
from zerver.decorator import webhook_view 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.response import json_success
from zerver.lib.validator import check_bool, check_dict, check_string from zerver.lib.typed_endpoint import ApiParamConfig, typed_endpoint
from zerver.lib.webhooks.common import check_send_webhook_message from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile from zerver.models import UserProfile
@ -24,26 +24,26 @@ Build status: {} {}
Details: [changes]({}), [build log]({})""" Details: [changes]({}), [build log]({})"""
class TravisPayload(BaseModel):
author_name: str
status_message: str
compare_url: str
build_url: str
type: str
@webhook_view("Travis", all_event_types=ALL_EVENT_TYPES) @webhook_view("Travis", all_event_types=ALL_EVENT_TYPES)
@has_request_variables @typed_endpoint
def api_travis_webhook( def api_travis_webhook(
request: HttpRequest, request: HttpRequest,
user_profile: UserProfile, user_profile: UserProfile,
ignore_pull_requests: bool = REQ(json_validator=check_bool, default=True), *,
message: Dict[str, object] = REQ( message: Annotated[Json[TravisPayload], ApiParamConfig("payload")],
"payload", ignore_pull_requests: Json[bool] = True,
json_validator=check_dict(
[
("author_name", check_string),
("status_message", check_string),
("compare_url", check_string),
]
),
),
) -> HttpResponse: ) -> HttpResponse:
event = str(message["type"]) event = message.type
message_status = message["status_message"] message_status = message.status_message
if ignore_pull_requests and message["type"] == "pull_request": if ignore_pull_requests and message.type == "pull_request":
return json_success(request) return json_success(request)
if message_status in GOOD_STATUSES: if message_status in GOOD_STATUSES:
@ -56,11 +56,11 @@ def api_travis_webhook(
emoji = f"(No emoji specified for status '{message_status}'.)" emoji = f"(No emoji specified for status '{message_status}'.)"
body = MESSAGE_TEMPLATE.format( body = MESSAGE_TEMPLATE.format(
message["author_name"], message.author_name,
message_status, message_status,
emoji, emoji,
message["compare_url"], message.compare_url,
message["build_url"], message.build_url,
) )
topic = "builds" topic = "builds"