sonarqube: Strengthen types using WildValue.

This commit is contained in:
Hari Prashant Bhimaraju
2022-06-23 20:51:26 +05:30
committed by Tim Abbott
parent 146dd68cf2
commit 31c7f8266d

View File

@@ -1,12 +1,10 @@
# Webhooks for external integrations. # Webhooks for external integrations.
from typing import Any, Dict, List, Mapping
from django.http import HttpRequest, HttpResponse from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view from zerver.decorator import webhook_view
from zerver.lib.request import REQ, has_request_variables 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 WildValue, check_none_or, check_string, to_wild_value
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
@@ -35,15 +33,15 @@ def parse_metric_name(metric_name: str) -> str:
return " ".join(metric_name.split("_")) return " ".join(metric_name.split("_"))
def parse_condition(condition: Mapping[str, Any]) -> str: def parse_condition(condition: WildValue) -> str:
metric = condition["metric"] metric = condition["metric"].tame(check_string)
metric_name = parse_metric_name(metric) metric_name = parse_metric_name(metric)
operator = condition["operator"] operator = condition["operator"].tame(check_string)
operator = INVERSE_OPERATORS.get(operator, operator) operator = INVERSE_OPERATORS.get(operator, operator)
value = condition.get("value", "no value") value = condition.get("value", "no value").tame(check_string)
status = condition["status"].lower() status = condition["status"].tame(check_string).lower()
threshold = condition["errorThreshold"] threshold = condition["errorThreshold"].tame(check_string)
if value == "no value": if value == "no value":
return TEMPLATES["no_value"].format(metric_name, status) return TEMPLATES["no_value"].format(metric_name, status)
@@ -53,28 +51,28 @@ def parse_condition(condition: Mapping[str, Any]) -> str:
return template.format(metric_name, status, value, operator, threshold) return template.format(metric_name, status, value, operator, threshold)
def parse_conditions(conditions: List[Mapping[str, Any]]) -> str: def parse_conditions(conditions: WildValue) -> str:
return "\n".join( return "\n".join(
[ [
parse_condition(condition) parse_condition(condition)
for condition in conditions for condition in conditions
if condition["status"].lower() != "ok" and condition["status"].lower() != "no_value" if condition["status"].tame(check_string).lower() != "ok"
and condition["status"].tame(check_string).lower() != "no_value"
] ]
) )
def render_body_with_branch(payload: Mapping[str, Any]) -> str: def render_body_with_branch(payload: WildValue) -> str:
project_name = payload["project"]["name"] project_name = payload["project"]["name"].tame(check_string)
project_url = payload["project"]["url"] project_url = payload["project"]["url"].tame(check_string)
quality_gate_status = payload["qualityGate"]["status"].lower() quality_gate_status = payload["qualityGate"]["status"].tame(check_string).lower()
if quality_gate_status == "ok": if quality_gate_status == "ok":
quality_gate_status = "success" quality_gate_status = "success"
else: else:
quality_gate_status = "error" quality_gate_status = "error"
branch = payload["branch"]["name"] branch = payload["branch"]["name"].tame(check_string)
conditions = payload["qualityGate"]["conditions"] conditions = parse_conditions(payload["qualityGate"]["conditions"])
conditions = parse_conditions(conditions)
if not conditions: if not conditions:
return MESSAGE_WITH_BRANCH_AND_WITHOUT_CONDITIONS.format( return MESSAGE_WITH_BRANCH_AND_WITHOUT_CONDITIONS.format(
@@ -88,16 +86,15 @@ def render_body_with_branch(payload: Mapping[str, Any]) -> str:
return msg return msg
def render_body_without_branch(payload: Mapping[str, Any]) -> str: def render_body_without_branch(payload: WildValue) -> str:
project_name = payload["project"]["name"] project_name = payload["project"]["name"].tame(check_string)
project_url = payload["project"]["url"] project_url = payload["project"]["url"].tame(check_string)
quality_gate_status = payload["qualityGate"]["status"].lower() quality_gate_status = payload["qualityGate"]["status"].tame(check_string).lower()
if quality_gate_status == "ok": if quality_gate_status == "ok":
quality_gate_status = "success" quality_gate_status = "success"
else: else:
quality_gate_status = "error" quality_gate_status = "error"
conditions = payload["qualityGate"]["conditions"] conditions = parse_conditions(payload["qualityGate"]["conditions"])
conditions = parse_conditions(conditions)
if not conditions: if not conditions:
return MESSAGE_WITHOUT_BRANCH_AND_CONDITIONS.format( return MESSAGE_WITHOUT_BRANCH_AND_CONDITIONS.format(
@@ -116,12 +113,12 @@ def render_body_without_branch(payload: Mapping[str, Any]) -> str:
def api_sonarqube_webhook( def api_sonarqube_webhook(
request: HttpRequest, request: HttpRequest,
user_profile: UserProfile, user_profile: UserProfile,
payload: Dict[str, Any] = REQ(argument_type="body"), payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
) -> HttpResponse: ) -> HttpResponse:
project = payload["project"]["name"] project = payload["project"]["name"].tame(check_string)
branch = None branch = None
if "branch" in payload.keys(): if "branch" in payload.keys():
branch = payload["branch"].get("name", None) branch = payload["branch"].get("name").tame(check_none_or(check_string))
if branch: if branch:
topic = TOPIC_WITH_BRANCH.format(project, branch) topic = TOPIC_WITH_BRANCH.format(project, branch)
message = render_body_with_branch(payload) message = render_body_with_branch(payload)