lidarr: Strengthen types using WildValue.

zerver/webhooks/lidarr/view.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, List
+from typing import Dict, List
 
 from django.http import HttpRequest, HttpResponse
 
@@ -6,6 +6,7 @@ from zerver.decorator import webhook_view
 from zerver.lib.exceptions import UnsupportedWebhookEventType
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
+from zerver.lib.validator import WildValue, check_bool, check_string, to_wild_value
 from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message
 from zerver.models import UserProfile
 
@@ -39,7 +40,7 @@ LIDARR_TRACKS_LIMIT = 20
 ALL_EVENT_TYPES = ["Test", "Grab", "Rename", "Retag", "Download"]
 
 
-def get_tracks_content(tracks_data: List[Dict[str, Any]]) -> str:
+def get_tracks_content(tracks_data: List[Dict[str, str]]) -> str:
     tracks_content = ""
     for track in tracks_data[:LIDARR_TRACKS_LIMIT]:
         tracks_content += LIDARR_TRACKS_ROW_TEMPLATE.format(track_title=track.get("title"))
@@ -57,77 +58,87 @@ def get_tracks_content(tracks_data: List[Dict[str, Any]]) -> str:
 def api_lidarr_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:
     body = get_body_for_http_request(payload)
     subject = get_subject_for_http_request(payload)
 
-    check_send_webhook_message(request, user_profile, subject, body, payload["eventType"])
+    check_send_webhook_message(
+        request, user_profile, subject, body, payload["eventType"].tame(check_string)
+    )
     return json_success(request)
 
 
-def get_subject_for_http_request(payload: Dict[str, Any]) -> str:
-    if payload["eventType"] == "Test":
+def get_subject_for_http_request(payload: WildValue) -> str:
+    if payload["eventType"].tame(check_string) == "Test":
         topic = LIDARR_TOPIC_TEMPLATE_TEST
     else:
-        topic = LIDARR_TOPIC_TEMPLATE.format(artist_name=payload["artist"]["name"])
+        topic = LIDARR_TOPIC_TEMPLATE.format(
+            artist_name=payload["artist"]["name"].tame(check_string)
+        )
 
     return topic
 
 
-def get_body_for_album_grabbed_event(payload: Dict[str, Any]) -> str:
+def get_body_for_album_grabbed_event(payload: WildValue) -> str:
     return LIDARR_MESSAGE_TEMPLATE_ALBUM_GRABBED.format(
-        artist_name=payload["artist"]["name"], album_name=payload["albums"][0]["title"]
+        artist_name=payload["artist"]["name"].tame(check_string),
+        album_name=payload["albums"][0]["title"].tame(check_string),
     )
 
 
-def get_body_for_tracks_renamed_event(payload: Dict[str, Any]) -> str:
-    return LIDARR_MESSAGE_TEMPLATE_TRACKS_RENAMED.format(artist_name=payload["artist"]["name"])
+def get_body_for_tracks_renamed_event(payload: WildValue) -> str:
+    return LIDARR_MESSAGE_TEMPLATE_TRACKS_RENAMED.format(
+        artist_name=payload["artist"]["name"].tame(check_string)
+    )
 
 
-def get_body_for_tracks_retagged_event(payload: Dict[str, Any]) -> str:
-    return LIDARR_MESSAGE_TEMPLATE_TRACKS_RETAGGED.format(artist_name=payload["artist"]["name"])
+def get_body_for_tracks_retagged_event(payload: WildValue) -> str:
+    return LIDARR_MESSAGE_TEMPLATE_TRACKS_RETAGGED.format(
+        artist_name=payload["artist"]["name"].tame(check_string)
+    )
 
 
-def get_body_for_tracks_imported_upgrade_event(payload: Dict[str, Any]) -> str:
+def get_body_for_tracks_imported_upgrade_event(payload: WildValue) -> str:
     tracks_data = []
     for track in payload["tracks"]:
-        tracks_data.append({"title": track["title"]})
+        tracks_data.append({"title": track["title"].tame(check_string)})
 
     data = {
-        "artist_name": payload["artist"]["name"],
+        "artist_name": payload["artist"]["name"].tame(check_string),
         "tracks_final_data": get_tracks_content(tracks_data),
     }
 
     return LIDARR_MESSAGE_TEMPLATE_TRACKS_IMPORTED_UPGRADE.format(**data)
 
 
-def get_body_for_tracks_imported_event(payload: Dict[str, Any]) -> str:
+def get_body_for_tracks_imported_event(payload: WildValue) -> str:
     tracks_data = []
     for track in payload["tracks"]:
-        tracks_data.append({"title": track["title"]})
+        tracks_data.append({"title": track["title"].tame(check_string)})
 
     data = {
-        "artist_name": payload["artist"]["name"],
+        "artist_name": payload["artist"]["name"].tame(check_string),
         "tracks_final_data": get_tracks_content(tracks_data),
     }
 
     return LIDARR_MESSAGE_TEMPLATE_TRACKS_IMPORTED.format(**data)
 
 
-def get_body_for_http_request(payload: Dict[str, Any]) -> str:
-    if payload["eventType"] == "Test":
+def get_body_for_http_request(payload: WildValue) -> str:
+    event_type = payload["eventType"].tame(check_string)
+    if event_type == "Test":
         return get_setup_webhook_message("Lidarr")
-    elif payload["eventType"] == "Grab":
+    elif event_type == "Grab":
         return get_body_for_album_grabbed_event(payload)
-    elif payload["eventType"] == "Rename":
+    elif event_type == "Rename":
         return get_body_for_tracks_renamed_event(payload)
-    elif payload["eventType"] == "Retag":
+    elif event_type == "Retag":
         return get_body_for_tracks_retagged_event(payload)
-    elif payload["eventType"] == "Download" and "isUpgrade" in payload:
-        if payload["isUpgrade"]:
+    elif event_type == "Download" and "isUpgrade" in payload:
+        if payload["isUpgrade"].tame(check_bool):
             return get_body_for_tracks_imported_upgrade_event(payload)
         else:
             return get_body_for_tracks_imported_event(payload)
     else:
-        raise UnsupportedWebhookEventType(payload["eventType"])
+        raise UnsupportedWebhookEventType(event_type)