Coverage for src/comments_views/core/utils.py: 91%

1from json import JSONDecodeError 

2from urllib.parse import urlencode 

3 

4import requests 

5from django.conf import settings 

6from django.contrib import messages 

7from django.contrib.auth import get_user_model 

8from django.http import HttpRequest 

9from django.urls import reverse 

10from django.utils.translation import gettext_lazy as _ 

11from requests import RequestException, Response 

12 

13from comments_api.constants import API_MESSAGE_KEY 

14from comments_api.utils import parse_date 

15 

16from .app_settings import app_settings 

17from .rights import AbstractUserRights 

18 

19# Default timeout (seconds) to be used when performing API requests. 

20DEFAULT_TIMEOUT = 4 

21DEFAULT_API_ERROR_MESSAGE = _("Error: Something went wrong. Please try again.") 

22 

23 

24def comments_server_base_url() -> str: 

25 """ 

26 Returns the base URL of the comment server as defined in settings.py. 

27 Fallbacks to https://comments.centre-mersenne.org 

28 """ 

29 comment_base_url = app_settings.API_BASE_URL 

30 

31 # Remove trailing slash 

32 if comment_base_url[-1] == "/": 

33 comment_base_url = comment_base_url[:-1] 

34 return comment_base_url 

35 

36 

37def comments_server_url( 

38 query_params: dict[str, str] = {}, url_prefix="comments", item_id=None 

39) -> str: 

40 """ 

41 Returns a formatted URL to query the comment server (API). 

42 The default URL corresponds to the list of all available comments. 

43 

44 Parameters: 

45 - `query_params`: A dictionnary of query parameters to be appended to the URL. 

46 They are used to precisely filter the comment dataset. 

47 - `comment_id`: Filters on the given comment ID 

48 """ 

49 base_url = f"{comments_server_base_url()}/api/{url_prefix}/" 

50 if item_id: 

51 base_url += f"{item_id}/" 

52 if query_params: 

53 base_url += f"?{urlencode(query_params)}" 

54 return base_url 

55 

56 

57def comments_credentials() -> tuple[str, str]: 

58 """ 

59 Returns the current server credentials (login, password) required to authenticate 

60 on the comment server. 

61 """ 

62 credentials = app_settings.API_CREDENTIALS 

63 if credentials: 

64 assert isinstance(credentials, tuple) and len(credentials) == 2, ( 

65 "The server has malformed comments server credentials." 

66 "Please provide them as `(login, password)`" 

67 ) 

68 

69 return credentials 

70 

71 raise AttributeError("The server is missing the comments server credentials") 

72 

73 

74def add_api_error_message(initial_message: str, api_json_data: dict) -> str: 

75 """Adds the API error description to the given string.""" 

76 formatted_message = initial_message 

77 api_error = api_json_data.get(API_MESSAGE_KEY) 

78 if api_error and isinstance(api_error, str): 

79 formatted_message = f"{formatted_message}<br><br>" if formatted_message else "" 

80 formatted_message += f'"{api_error}"' 

81 

82 return formatted_message 

83 

84 

85def json_from_response( 

86 response: Response, request_for_message: HttpRequest | None = None 

87) -> tuple[bool, dict]: 

88 """ 

89 Returns the JSON content of a HTTP response and whether an exception was raised 

90 while doing so. 

91 We consider that there's an exception if the response does not contain correctly 

92 formed JSON (except HTTP 204) or if its status code >= 300. 

93 

94 Params: 

95 - `response` The HTTP response to parse. 

96 - `request_for_message` The current HTTP request handled by Django. Used 

97 to add a message (Django) when there's an exception. 

98 Don't provide it if you don't want any messages. 

99 

100 Returns: 

101 - `error` Whether an exception was raised. 

102 - `content` The JSON content of the HTTP response. 

103 """ 

104 try: 

105 data = response.json() 

106 except JSONDecodeError: 

107 # 204 No Content 

108 if response.status_code == 204: 

109 return (False, {}) 

110 if request_for_message is not None: 

111 messages.error(request_for_message, DEFAULT_API_ERROR_MESSAGE) 

112 return (True, {}) 

113 

114 error = False 

115 if response.status_code >= 300: 

116 if request_for_message is not None: 

117 error_message = add_api_error_message(DEFAULT_API_ERROR_MESSAGE, data) 

118 messages.error(request_for_message, error_message) 

119 error = True 

120 return (error, data) 

121 

122 

123def format_comment_author_name(comment: dict) -> str: 

124 """ 

125 Formats an author name according to the comments properties. 

126 """ 

127 author_name = comment.get("author_name", "") 

128 

129 hide_name = comment.get("hide_author_name") 

130 article_author = comment.get("article_author_comment") 

131 editorial_team = comment.get("editorial_team_comment") 

132 extra_text = "" 

133 

134 if article_author: 

135 if hide_name: 

136 return _("Auteur(s) de l'article").capitalize() 

137 extra_text = _("Auteur de l'article") 

138 elif editorial_team: 

139 extra_text = _("Équipe éditoriale") 

140 if hide_name: 

141 return extra_text.capitalize() 

142 

143 if not extra_text: 

144 return author_name 

145 

146 return f"{author_name} ({extra_text.lower()})" 

147 

148 

149def format_comment( 

150 comment: dict, rights: AbstractUserRights | None = None, users: dict = {} 

151) -> None: 

152 """ 

153 Format a comment data for display. 

154 Formatted data includes: 

155 - author display name (actual name or function) 

156 - Moderation data (only if `users is provided`) 

157 - comment URL 

158 

159 Params: 

160 - `comment` The comment data (usually a dict constructed from 

161 the JSON data). 

162 - `rights` The user rights used to check access over moderation data. 

163 - `users` The dictionnary of users used to map moderator IDs 

164 """ 

165 # Format author name and parent author name 

166 comment["author_display_name"] = format_comment_author_name(comment) 

167 

168 parent = comment.get("parent") 

169 # The comment data may contain either the parent dictionnary data or 

170 # just the parent ID 

171 if parent and isinstance(parent, dict): 

172 comment["parent"]["author_display_name"] = format_comment_author_name(parent) 

173 

174 # Format author provider link (only ORCID right now) 

175 author_provider = comment.get("author_provider") 

176 author_provider_uid = comment.get("author_provider_uid") 

177 if author_provider and author_provider.lower() == "orcid" and author_provider_uid: 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 comment["author_provider_link"] = f"{app_settings.ORCID_BASE_URL}{author_provider_uid}/" 

179 

180 # Parse dates for display 

181 date_submitted = comment.get("date_submitted") 

182 if date_submitted: 182 ↛ 185line 182 didn't jump to line 185 because the condition on line 182 was always true

183 comment["date_submitted"] = parse_date(date_submitted) 

184 

185 moderation_data = comment.get("moderation_data") 

186 if moderation_data: 186 ↛ 205line 186 didn't jump to line 205 because the condition on line 186 was always true

187 if moderation_data.get("date_moderated"): 187 ↛ 192line 187 didn't jump to line 192 because the condition on line 187 was always true

188 comment["moderation_data"]["date_moderated"] = parse_date( 

189 moderation_data.get("date_moderated") 

190 ) 

191 

192 moderator_id = moderation_data.get("moderator") 

193 if ( 193 ↛ 202line 193 didn't jump to line 202 because the condition on line 193 was never true

194 moderator_id 

195 and moderator_id in users 

196 and rights 

197 and ( 

198 rights.comment_can_manage_moderators(comment) 

199 or getattr(rights.user, "is_superuser", False) 

200 ) 

201 ): 

202 comment["moderation_data"]["moderated_by"] = users[moderator_id] 

203 

204 # Format comment link 

205 base_url = comment.get("base_url") 

206 if base_url: 206 ↛ 220line 206 didn't jump to line 220 because the condition on line 206 was always true

207 # Remove potential trailing slash 

208 if base_url[-1] == "/": 

209 base_url = base_url[:-1] 

210 article_route = reverse("article", kwargs={"aid": comment["doi"]}) 

211 # If a site has an URL prefix, we need to discard it from the article_route, 

212 # it should already be present in base_url. 

213 url_prefix = getattr(settings, "SITE_URL_PREFIX", None) 

214 if url_prefix: 214 ↛ 218line 214 didn't jump to line 218 because the condition on line 214 was always true

215 url_prefix = f"/{url_prefix}" 

216 if article_route.startswith(url_prefix): 

217 article_route = article_route[len(url_prefix) :] 

218 comment["base_url"] = f"{base_url}{article_route}" 

219 

220 if rights: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true

221 comment["can_moderate"] = rights.comment_can_moderate(comment) 

222 

223 # Get the moderators 

224 if rights and users: 224 ↛ 225line 224 didn't jump to line 225 because the condition on line 224 was never true

225 moderators = comment.get("moderators") 

226 if moderators and ( 

227 rights.comment_can_manage_moderators(comment) 

228 or getattr(rights.user, "is_superuser", False) 

229 ): 

230 comment["moderators_processed"] = { 

231 moderator_id: users[moderator_id] 

232 for moderator_id in moderators 

233 if moderator_id in users 

234 } 

235 

236 

237def get_user_dict() -> dict[int, dict[str, str]]: 

238 """ 

239 Return a dict of users indexed by the user's pk. 

240 A value is a dict of the user's `pk`, `first_name`, `last_name` and `email` 

241 """ 

242 users = get_user_model().objects.all().values("pk", "first_name", "last_name", "email") 

243 return {u["pk"]: u for u in users} 

244 

245 

246def api_request_wrapper(http_method: str, url: str, **kwargs) -> Response | None: 

247 """ 

248 Light wrapper around `get`, `post`, and `put` methods of the `requests` library: 

249 - Catch any `RequestException`. 

250 - Add a default timeout to the request if no `timeout` was provided. 

251 

252 Params: 

253 - `http_method` The HTTP method to use for the request 

254 (get, post, put, head or delete). 

255 

256 Returns: 

257 - `response` The `Response` object if no `RequestException` was raised, 

258 else `None`. 

259 """ 

260 if "timeout" not in kwargs: 260 ↛ 263line 260 didn't jump to line 263 because the condition on line 260 was always true

261 # (connect timeout, read timeout) 

262 kwargs["timeout"] = (4, 7) 

263 http_method = http_method.upper() 

264 try: 

265 if http_method == "GET": 

266 response = requests.get(url, **kwargs) 

267 elif http_method == "POST": 

268 response = requests.post(url, **kwargs) 

269 elif http_method == "PUT": 

270 response = requests.put(url, **kwargs) 

271 # Other HTTP methods are not used. They can be added above 

272 # The timeout kwarg is only available for HEAD and DELETE 

273 else: 

274 raise ValueError(f"Wrong argument for 'http_method': '{http_method}'") 

275 except RequestException: 

276 response = None 

277 

278 return response 

279 

280 

281def make_api_request( 

282 http_method: str, url: str, request_for_message: HttpRequest | None = None, **kwargs 

283) -> tuple[bool, dict]: 

284 """ 

285 Performs the HTTP request to the provided URL and return whether an exception was 

286 raised and the JSON content of the response. \\ 

287 Catched exceptions are any `RequestException` or `JSONDecodeError`. \\ 

288 Any additional kwargs are passed directly to the `requests` library. 

289 

290 Params: 

291 - `http_method` The HTTP method to use for the request 

292 (get, post, put, head or delete). 

293 - `url` The URL to request. 

294 - `request_for_message` The current HTTP request handled by Django. Used 

295 to add a message (Django) when there's an exception. 

296 Don't provide it if you don't want any messages. 

297 

298 Returns: 

299 - `error` Whether an exception was raised. 

300 - `content` The JSON content of the HTTP response. 

301 """ 

302 response = api_request_wrapper(http_method, url, **kwargs) 

303 

304 if response is None: 

305 if request_for_message is not None: 

306 messages.error(request_for_message, DEFAULT_API_ERROR_MESSAGE) 

307 return True, {} 

308 

309 return json_from_response(response, request_for_message) 

310 

311 

312def get_comment( 

313 query_params: dict[str, str], 

314 id: int, 

315 request_for_message: HttpRequest | None = None, 

316) -> tuple[bool, dict]: 

317 """ 

318 Make an HTTP request to retrieve the requested comment. Commodity wrapper 

319 around `make_api_request`. 

320 

321 Params: 

322 - `query_params` Query parameters for the API URL. 

323 - `id` The comment's ID. 

324 - `request_for_message` The current HTTP request handled by Django. Used 

325 to add a message (Django) when there's an exception. 

326 Don't provide it if you don't want any messages. 

327 

328 Returns: 

329 - `error` Whether an exception was raised. 

330 - `content` The JSON content of the HTTP response. 

331 """ 

332 return make_api_request( 

333 "GET", 

334 comments_server_url(query_params, item_id=id), 

335 request_for_message, 

336 auth=comments_credentials(), 

337 )