Coverage for app/venv/lib/python3.14/site-packages/weblate/addons/webhooks.py: 31%

1# Copyright © Michal Čihař <michal@weblate.org> 

2# 

3# SPDX-License-Identifier: GPL-3.0-or-later 

4 

5from __future__ import annotations 

6 

7import json 

8from typing import TYPE_CHECKING 

9 

10import jsonschema.exceptions 

11import requests 

12from django.template.loader import render_to_string 

13from django.utils import timezone as dj_timezone 

14from django.utils.translation import gettext_lazy 

15from drf_spectacular.utils import OpenApiResponse, OpenApiWebhook, extend_schema 

16from standardwebhooks.webhooks import Webhook 

17from weblate_schemas import load_schema, validate_schema 

18 

19from weblate.addons.base import ChangeBaseAddon 

20from weblate.addons.forms import BaseWebhooksAddonForm, WebhooksAddonForm 

21from weblate.trans.util import split_plural 

22from weblate.utils.requests import request 

23from weblate.utils.site import get_site_url 

24from weblate.utils.views import key_name 

25 

26if TYPE_CHECKING: 26 ↛ 27line 26 didn't jump to line 27 because the condition on line 26 was never true

27 from collections.abc import Mapping 

28 

29 from weblate.addons.models import AddonActivityLog 

30 from weblate.trans.models import Change 

31 

32 PayloadType = Mapping[str, int | str | list] 

33 

34 

35class MessageNotDeliveredError(Exception): 

36 """Exception raised when a message could not be delivered.""" 

37 

38 

39class JSONWebhookBaseAddon(ChangeBaseAddon): 

40 icon = "webhook.svg" 

41 

42 def build_webhook_payload(self, change: Change) -> PayloadType: 

43 raise NotImplementedError 

44 

45 def build_headers(self, change: Change, payload: PayloadType) -> dict[str, str]: 

46 return {} 

47 

48 def render_activity_log(self, activity: AddonActivityLog) -> str: 

49 return render_to_string( 

50 "addons/webhook_log.html", 

51 {"activity": activity, "details": activity.details.get("result")}, 

52 ) 

53 

54 def send_message( 

55 self, change: Change, headers: dict, payload: PayloadType 

56 ) -> requests.Response: 

57 try: 

58 return request( 

59 method="post", 

60 url=self.instance.configuration["webhook_url"], 

61 json=payload, 

62 headers=headers, 

63 timeout=15, 

64 raise_for_status=False, 

65 ) 

66 except requests.exceptions.ConnectionError as error: 

67 raise MessageNotDeliveredError from error 

68 

69 def change_event( 

70 self, change: Change, activity_log_id: int | None = None 

71 ) -> dict | None: 

72 """Deliver notification message.""" 

73 config = self.instance.configuration 

74 events = {int(event) for event in config["events"]} 

75 if change.action in events: 

76 payload = self.build_webhook_payload(change) 

77 headers = self.build_headers(change, payload) 

78 response = self.send_message(change, headers, payload) 

79 

80 return { 

81 "request": {"headers": headers, "payload": payload}, 

82 "response": { 

83 "status_code": response.status_code, 

84 "content": response.text, 

85 "headers": dict(response.headers), 

86 }, 

87 } 

88 return None 

89 

90 

91class WebhookAddon(JSONWebhookBaseAddon): 

92 """Class for Webhooks Addon.""" 

93 

94 name = "weblate.webhook.webhook" 

95 verbose = gettext_lazy("Webhook") 

96 description = gettext_lazy( 

97 "Sends notifications to external services based on selected events, following the Standard Webhooks specification." 

98 ) 

99 

100 settings_form = WebhooksAddonForm 

101 

102 def build_webhook_payload(self, change: Change) -> PayloadType: 

103 """Build a Schema-valid payload from change event.""" 

104 data: dict[str, int | str | list] = { 

105 "change_id": change.id, 

106 "action": change.get_action_display(), 

107 "timestamp": change.timestamp.isoformat(), 

108 } 

109 if change.target: 

110 data["target"] = split_plural(change.target) 

111 if change.old: 

112 data["old"] = split_plural(change.old) 

113 if change.unit: 

114 data["source"] = split_plural(change.unit.source) 

115 if url := change.get_absolute_url(): 

116 data["url"] = url 

117 if change.author: 

118 data["author"] = change.author.username 

119 if change.user: 

120 data["user"] = change.user.username 

121 if change.project: 

122 data["project"] = change.project.slug 

123 if change.component: 

124 data["component"] = change.component.slug 

125 if change.translation: 

126 data["translation"] = change.translation.language.code 

127 if change.component and (category := change.component.category): 

128 categories = [] 

129 while category is not None: 

130 categories.append(category.slug) 

131 category = category.category 

132 if categories: 

133 data["category"] = categories 

134 self.validate_payload(data) 

135 return data 

136 

137 def build_headers(self, change: Change, payload: PayloadType) -> dict[str, str]: 

138 """Build headers following Standard Webhooks specifications.""" 

139 webhook_id = change.get_uuid().hex 

140 attempt_time = dj_timezone.now() 

141 headers: dict[str, str] = { 

142 "webhook-timestamp": str(attempt_time.timestamp()), 

143 "webhook-id": webhook_id, 

144 "webhook-signature": "", 

145 } 

146 if secret := self.instance.configuration.get("secret", ""): 

147 headers["webhook-signature"] = Webhook(secret).sign( 

148 webhook_id, attempt_time, json.dumps(payload) 

149 ) 

150 

151 return headers 

152 

153 def validate_payload(self, payload) -> None: 

154 try: 

155 validate_schema(payload, "weblate-messaging.schema.json") 

156 except ( 

157 jsonschema.exceptions.ValidationError, 

158 jsonschema.exceptions.SchemaError, 

159 ) as error: 

160 raise MessageNotDeliveredError from error 

161 

162 

163change_event_webhook = OpenApiWebhook( 

164 name="AddonWebhook", 

165 decorator=extend_schema( 

166 summary="A Webhook event for an addon", 

167 description="Pushes events to a notification URL.", 

168 tags=["webhooks", "addons"], 

169 request={"application/json": load_schema("weblate-messaging.schema.json")}, 

170 responses={ 

171 "2XX": OpenApiResponse(), 

172 }, 

173 ), 

174) 

175 

176 

177class SlackWebhookAddon(JSONWebhookBaseAddon): 

178 name = "weblate.webhook.slack" 

179 verbose = gettext_lazy("Slack Webhooks") 

180 description = gettext_lazy( 

181 "Sends notification to a Slack channel based on selected events." 

182 ) 

183 icon = "slack.svg" 

184 settings_form = BaseWebhooksAddonForm 

185 

186 def build_webhook_payload(self, change: Change) -> PayloadType: 

187 message_header = "" 

188 if change.path_object: 

189 message_header += key_name(change.path_object) + " - " 

190 message_header += change.get_action_display() 

191 payload: dict[str, list] = { 

192 "blocks": [ 

193 { 

194 "type": "header", 

195 "text": {"type": "plain_text", "text": message_header}, 

196 }, 

197 ] 

198 } 

199 

200 change_details = change.get_details_display() or "No details" 

201 

202 payload["blocks"].append( 

203 { 

204 "type": "section", 

205 "text": {"type": "plain_text", "text": change_details}, 

206 "accessory": { 

207 "type": "button", 

208 "text": { 

209 "type": "plain_text", 

210 "text": "View", 

211 }, 

212 "value": "view-change", 

213 "url": get_site_url(change.get_absolute_url()), 

214 }, 

215 } 

216 ) 

217 return payload