Coverage for src / couchers / notifications / push.py: 94%

1from dataclasses import dataclass 

2from typing import ClassVar 

3 

4from sqlalchemy import select 

5from sqlalchemy.orm import Session 

6from sqlalchemy.sql import func 

7 

8from couchers import urls 

9from couchers.config import config 

10from couchers.jobs.enqueue import queue_job 

11from couchers.models import PushNotificationSubscription 

12from couchers.notifications.send_raw_push_notification import send_raw_push_notification_v2 

13from couchers.proto.internal import jobs_pb2 

14 

15 

16@dataclass(frozen=True, slots=True) 

17class PushNotificationContent: 

18 """Defines the user-visible content of a push notification.""" 

19 

20 # Android reference: https://developer.android.com/develop/ui/views/notifications 

21 # iOS reference: https://developer.apple.com/documentation/usernotifications/unnotificationcontent 

22 

23 MAX_TITLE_LENGTH: ClassVar[int] = 500 

24 MAX_BODY_LENGTH: ClassVar[int] = 2000 

25 

26 title: str 

27 """A localized title for the notification, this should be a very short string (2-4 words).""" 

28 body: str 

29 """The localized text of the notification body.""" 

30 action_url: str | None = None 

31 """The URL to open when the notification is clicked. If None, will open the app's main URL.""" 

32 icon_url: str | None = None 

33 """A URL to the icon to show in the notification. If None, will use the default app icon.""" 

34 

35 

36def push_to_subscription( 

37 session: Session, 

38 *, 

39 push_notification_subscription_id: int, 

40 user_id: int, 

41 topic_action: str, 

42 content: PushNotificationContent, 

43 key: str | None = None, 

44 ttl: int = 0, 

45) -> None: 

46 title = config["NOTIFICATION_PREFIX"] + content.title[: PushNotificationContent.MAX_TITLE_LENGTH] 

47 body = content.body[: PushNotificationContent.MAX_BODY_LENGTH] 

48 icon_url = content.icon_url or urls.icon_url() 

49 action_url = content.action_url or "" 

50 queue_job( 

51 session, 

52 job=send_raw_push_notification_v2, 

53 payload=jobs_pb2.SendRawPushNotificationPayloadV2( 

54 push_notification_subscription_id=push_notification_subscription_id, 

55 ttl=ttl, 

56 title=title, 

57 body=body, 

58 icon=icon_url, 

59 url=action_url, 

60 user_id=user_id, 

61 topic_action=topic_action, 

62 key=key or "", 

63 ), 

64 priority=7, 

65 ) 

66 

67 

68def _push_to_user( 

69 session: Session, 

70 user_id: int, 

71 topic_action: str, 

72 content: PushNotificationContent, 

73 key: str | None, 

74 ttl: int, 

75) -> None: 

76 """ 

77 Same as above but for a given user 

78 """ 

79 sub_ids = ( 

80 session.execute( 

81 select(PushNotificationSubscription.id) 

82 .where(PushNotificationSubscription.user_id == user_id) 

83 .where(PushNotificationSubscription.disabled_at > func.now()) 

84 ) 

85 .scalars() 

86 .all() 

87 ) 

88 for sub_id in sub_ids: 88 ↛ 89line 88 didn't jump to line 89 because the loop on line 88 never started

89 push_to_subscription( 

90 session, 

91 push_notification_subscription_id=sub_id, 

92 user_id=user_id, 

93 topic_action=topic_action, 

94 content=content, 

95 key=key, 

96 ttl=ttl, 

97 ) 

98 

99 

100def push_to_user( 

101 session: Session, 

102 *, 

103 user_id: int, 

104 topic_action: str, 

105 content: PushNotificationContent, 

106 key: str | None = None, 

107 ttl: int = 0, 

108) -> None: 

109 """ 

110 This indirection is so that this can be easily mocked. Not sure how to do it better :( 

111 """ 

112 _push_to_user( 

113 session, 

114 user_id=user_id, 

115 topic_action=topic_action, 

116 content=content, 

117 key=key, 

118 ttl=ttl, 

119 )