Coverage for src/couchers/servicers/notifications.py: 93%

71 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 16:44 +0000

1import logging 

2 

3import grpc 

4from google.protobuf import empty_pb2 

5 

6from couchers import errors 

7from couchers.config import config 

8from couchers.db import session_scope 

9from couchers.models import ( 

10 HostingStatus, 

11 MeetupStatus, 

12 Notification, 

13 NotificationDeliveryType, 

14 PushNotificationSubscription, 

15 User, 

16) 

17from couchers.notifications.push import get_vapid_public_key, push_to_subscription, push_to_user 

18from couchers.notifications.push_api import decode_key 

19from couchers.notifications.render import render_notification 

20from couchers.notifications.settings import PreferenceNotUserEditableError, get_user_setting_groups, set_preference 

21from couchers.notifications.utils import enum_from_topic_action 

22from couchers.sql import couchers_select as select 

23from couchers.utils import Timestamp_from_datetime 

24from proto import notifications_pb2, notifications_pb2_grpc 

25 

26logger = logging.getLogger(__name__) 

27MAX_PAGINATION_LENGTH = 100 

28 

29 

30def notification_to_pb(user, notification: Notification): 

31 rendered = render_notification(user, notification) 

32 return notifications_pb2.Notification( 

33 notification_id=notification.id, 

34 created=Timestamp_from_datetime(notification.created), 

35 topic=notification.topic_action.topic, 

36 action=notification.topic_action.action, 

37 key=notification.key, 

38 title=rendered.push_title, 

39 body=rendered.push_body, 

40 icon=rendered.push_icon, 

41 url=rendered.push_url, 

42 ) 

43 

44 

45class Notifications(notifications_pb2_grpc.NotificationsServicer): 

46 def GetNotificationSettings(self, request, context): 

47 with session_scope() as session: 

48 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

49 return notifications_pb2.GetNotificationSettingsRes( 

50 do_not_email_enabled=user.do_not_email, 

51 groups=get_user_setting_groups(user.id), 

52 ) 

53 

54 def SetNotificationSettings(self, request, context): 

55 with session_scope() as session: 

56 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

57 user.do_not_email = request.enable_do_not_email 

58 if request.enable_do_not_email: 

59 user.hosting_status = HostingStatus.cant_host 

60 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

61 for preference in request.preferences: 

62 topic_action = enum_from_topic_action.get((preference.topic, preference.action), None) 

63 if not topic_action: 

64 context.abort(grpc.StatusCode.NOT_FOUND, errors.INVALID_NOTIFICATION_PREFERENCE) 

65 delivery_types = {t.name for t in NotificationDeliveryType} 

66 if preference.delivery_method not in delivery_types: 

67 context.abort(grpc.StatusCode.NOT_FOUND, errors.INVALID_DELIVERY_METHOD) 

68 delivery_type = NotificationDeliveryType[preference.delivery_method] 

69 try: 

70 set_preference(session, user.id, topic_action, delivery_type, preference.enabled) 

71 except PreferenceNotUserEditableError: 

72 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANNOT_EDIT_THAT_NOTIFICATION_PREFERENCE) 

73 return notifications_pb2.GetNotificationSettingsRes( 

74 do_not_email_enabled=user.do_not_email, 

75 groups=get_user_setting_groups(user.id), 

76 ) 

77 

78 def ListNotifications(self, request, context): 

79 with session_scope() as session: 

80 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

81 page_size = min(MAX_PAGINATION_LENGTH, request.page_size or MAX_PAGINATION_LENGTH) 

82 next_notification_id = int(request.page_token) if request.page_token else 2**50 

83 notifications = ( 

84 session.execute( 

85 select(Notification) 

86 .where(Notification.user_id == context.user_id) 

87 .where(Notification.id <= next_notification_id) 

88 .order_by(Notification.id.desc()) 

89 .limit(page_size + 1) 

90 ) 

91 .scalars() 

92 .all() 

93 ) 

94 return notifications_pb2.ListNotificationsRes( 

95 notifications=[notification_to_pb(user, notification) for notification in notifications[:page_size]], 

96 next_page_token=str(notifications[-1].id) if len(notifications) > page_size else None, 

97 ) 

98 

99 def GetVapidPublicKey(self, request, context): 

100 if not config["PUSH_NOTIFICATIONS_ENABLED"]: 

101 context.abort(grpc.StatusCode.UNAVAILABLE, errors.PUSH_NOTIFICATIONS_DISABLED) 

102 

103 return notifications_pb2.GetVapidPublicKeyRes(vapid_public_key=get_vapid_public_key()) 

104 

105 def RegisterPushNotification(self, request, context): 

106 if not config["PUSH_NOTIFICATIONS_ENABLED"]: 

107 context.abort(grpc.StatusCode.UNAVAILABLE, errors.PUSH_NOTIFICATIONS_DISABLED) 

108 

109 with session_scope() as session: 

110 subscription = PushNotificationSubscription( 

111 user_id=context.user_id, 

112 endpoint=request.endpoint, 

113 p256dh_key=decode_key(request.p256dh_key), 

114 auth_key=decode_key(request.auth_key), 

115 full_subscription_info=request.full_subscription_json, 

116 user_agent=request.user_agent, 

117 ) 

118 session.add(subscription) 

119 session.flush() 

120 sub_id = subscription.id 

121 push_to_subscription( 

122 sub_id, 

123 title="Checking push notifications work!", 

124 body="Hi, thanks for enabling push notifications!", 

125 ) 

126 return empty_pb2.Empty() 

127 

128 def SendTestPushNotification(self, request, context): 

129 if not config["PUSH_NOTIFICATIONS_ENABLED"]: 

130 context.abort(grpc.StatusCode.UNAVAILABLE, errors.PUSH_NOTIFICATIONS_DISABLED) 

131 

132 push_to_user( 

133 context.user_id, 

134 title="Checking push notifications work!", 

135 body="If you see this, then it's working :)", 

136 ) 

137 return empty_pb2.Empty()