Coverage for app/backend/src/couchers/servicers/auth_unsubscribe.py: 69%

1""" 

2Implements the "quick links" that are included in emails, for unsubscribing without logging in. 

3""" 

4 

5import logging 

6 

7import grpc 

8from sqlalchemy import select 

9from sqlalchemy.orm import Session 

10 

11from couchers.constants import DATETIME_INFINITY 

12from couchers.context import CouchersContext, make_one_off_interactive_user_context 

13from couchers.helpers.hosting_meetup_status import record_hosting_meetup_status 

14from couchers.models import ( 

15 GroupChat, 

16 GroupChatSubscription, 

17 HostingMeetupStatusSource, 

18 HostingStatus, 

19 MeetupStatus, 

20 NotificationDeliveryType, 

21 User, 

22) 

23from couchers.notifications import settings 

24from couchers.notifications.utils import enum_from_topic_action 

25from couchers.proto import messages_pb2, requests_pb2 

26from couchers.proto.internal import unsubscribe_pb2 

27from couchers.servicers.requests import Requests 

28from couchers.sql import where_moderated_content_visible 

29 

30logger = logging.getLogger(__name__) 

31 

32 

33def handle_unsubscribe(payload: unsubscribe_pb2.UnsubscribePayload, context: CouchersContext, session: Session) -> str: 

34 """ 

35 Returns a response string or uses context.abort upon error 

36 """ 

37 user = session.execute(select(User).where(User.id == payload.user_id)).scalar_one() 

38 

39 if payload.HasField("do_not_email"): 

40 logger.info(f"User {user.name} turning of emails") 

41 user.do_not_email = True 

42 user.hosting_status = HostingStatus.cant_host 

43 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

44 record_hosting_meetup_status(session, user, HostingMeetupStatusSource.unsubscribe_link) 

45 return context.localization.localize_string("quick_links.do_not_email") 

46 

47 if payload.HasField("topic_action"): 

48 logger.info(f"User {user.name} unsubscribing from topic_action") 

49 topic = payload.topic_action.topic 

50 action = payload.topic_action.action 

51 topic_action = enum_from_topic_action[topic, action] 

52 # disable emails for this type 

53 settings.set_preference(session, user.id, topic_action, NotificationDeliveryType.email, False) 

54 return context.localization.localize_string("quick_links.topic_action") 

55 

56 if payload.HasField("topic_key"): 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 logger.info(f"User {user.name} unsubscribing from topic_key") 

58 topic = payload.topic_key.topic 

59 key = payload.topic_key.key 

60 # a bunch of manual stuff 

61 if topic == "chat": 

62 group_chat_id = int(key) 

63 subscription = session.execute( 

64 where_moderated_content_visible( 

65 select(GroupChatSubscription).join( 

66 GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id 

67 ), 

68 context, 

69 GroupChat, 

70 is_list_operation=False, 

71 ) 

72 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

73 .where(GroupChatSubscription.user_id == user.id) 

74 .where(GroupChatSubscription.left == None) 

75 ).scalar_one_or_none() 

76 

77 if subscription is None: 

78 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "chat_not_found") 

79 

80 subscription.muted_until = DATETIME_INFINITY 

81 return context.localization.localize_string("quick_links.chat_unsub") 

82 else: 

83 context.abort_with_error_code(grpc.StatusCode.UNIMPLEMENTED, "cant_unsub_topic") 

84 

85 if payload.HasField("host_request_quick_decline"): 85 ↛ 96line 85 didn't jump to line 96 because the condition on line 85 was always true

86 Requests().RespondHostRequest( 

87 request=requests_pb2.RespondHostRequestReq( 

88 host_request_id=payload.host_request_quick_decline.host_request_id, 

89 status=messages_pb2.HOST_REQUEST_STATUS_REJECTED, 

90 ), 

91 context=make_one_off_interactive_user_context(couchers_context=context, user_id=payload.user_id), 

92 session=session, 

93 ) 

94 return context.localization.localize_string("quick_links.host_request_quick_decline") 

95 

96 raise Exception("Unhandled quick link type")