Coverage for middle_layer/common/application_layer/services/notification_sender_service.py: 84.00%

1# Copyright 2024 Associated Universities, Inc. 

2# 

3# This file is part of Telescope Time Allocation Tools (TTAT). 

4# 

5# TTAT is free software: you can redistribute it and/or modify 

6# it under the terms of the GNU General Public License as published by 

7# the Free Software Foundation, either version 3 of the License, or 

8# any later version. 

9# 

10# TTAT is distributed in the hope that it will be useful, 

11# but WITHOUT ANY WARRANTY; without even the implied warranty of 

12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

13# GNU General Public License for more details. 

14# 

15# You should have received a copy of the GNU General Public License 

16# along with TTAT. If not, see <https://www.gnu.org/licenses/>. 

17 

18from time import sleep 

19 

20import requests 

21 

22from common.utils.gitlab_secret_loader import NOTIFICATION_SERVICE_URL 

23 

24 

25def send_notification(template: str, parameters: dict[str, str] | None = None) -> None: 

26 """Send an email notification through Workspaces' notification service 

27 

28 :param template: Name of mustache template for email, must correspond to entry in Workspaces' template database 

29 :param parameters: Parameters for given template 

30 :raises RuntimeError: When the notification service fails to send the email 

31 """ 

32 if parameters is None: 

33 parameters = dict() 

34 url = NOTIFICATION_SERVICE_URL 

35 url = f"{url}/notify/{template}/send" 

36 response = requests.post(url, json=parameters) 

37 if response.status_code != 200: 

38 # wait 5 seconds and try again: There is an issue with message collisions in the notification system, 

39 # this should be unnecessary once it it resolved. 

40 sleep(5) 

41 response = requests.post(url, json=parameters) 

42 if response.status_code != 200: 

43 raise RuntimeError( 

44 f"Notification failure: Expected 200 (OK) response, " 

45 f"got {response.status_code} response instead: {response.text}" 

46 ) 

47 try: 

48 json = response.json() 

49 except requests.JSONDecodeError: 

50 raise RuntimeError(f"Notification failure: Expected JSON response, instead got: {response.text}") 

51 if json["errors"] or json["message"] != "Email sent": 

52 raise RuntimeError(f"Notification failure: Expected response with no errors, instead got: {json}") 

53 

54 

55def send_basic_notification( 

56 message: str, subject: str, primary_recipient: str, cc_recipient: str | None = None 

57) -> None: 

58 """Send a basic email notification via Workspaces' notification service 

59 

60 :param message: Body of the email 

61 :param subject: Subject of the email 

62 :param primary_recipient: Primary recipient of the email 

63 :param cc_recipient: Optional CC recipient for the email 

64 :raises RuntimeError: When the notification service fails to send the email 

65 """ 

66 params = { 

67 "message": message, 

68 "subject": subject, 

69 "destination_email": primary_recipient, 

70 } 

71 if cc_recipient is not None: 

72 params["cc_email"] = cc_recipient 

73 send_notification(template="email", parameters=params)