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

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.models import ( 

14 GroupChat, 

15 GroupChatSubscription, 

16 HostingStatus, 

17 MeetupStatus, 

18 NotificationDeliveryType, 

19 User, 

20) 

21from couchers.notifications import settings 

22from couchers.notifications.utils import enum_from_topic_action 

23from couchers.proto import conversations_pb2, requests_pb2 

24from couchers.proto.internal import unsubscribe_pb2 

25from couchers.servicers.requests import Requests 

26from couchers.sql import where_moderated_content_visible 

27 

28logger = logging.getLogger(__name__) 

29 

30 

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

32 """ 

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

34 """ 

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

36 

37 if payload.HasField("do_not_email"): 

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

39 user.do_not_email = True 

40 user.hosting_status = HostingStatus.cant_host 

41 user.meetup_status = MeetupStatus.does_not_want_to_meetup 

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

43 

44 if payload.HasField("topic_action"): 

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

46 topic = payload.topic_action.topic 

47 action = payload.topic_action.action 

48 topic_action = enum_from_topic_action[topic, action] 

49 # disable emails for this type 

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

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

52 

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

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

55 topic = payload.topic_key.topic 

56 key = payload.topic_key.key 

57 # a bunch of manual stuff 

58 if topic == "chat": 

59 group_chat_id = int(key) 

60 subscription = session.execute( 

61 where_moderated_content_visible( 

62 select(GroupChatSubscription).join( 

63 GroupChat, GroupChat.conversation_id == GroupChatSubscription.group_chat_id 

64 ), 

65 context, 

66 GroupChat, 

67 is_list_operation=False, 

68 ) 

69 .where(GroupChatSubscription.group_chat_id == group_chat_id) 

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

71 .where(GroupChatSubscription.left == None) 

72 ).scalar_one_or_none() 

73 

74 if subscription is None: 

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

76 

77 subscription.muted_until = DATETIME_INFINITY 

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

79 else: 

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

81 

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

83 Requests().RespondHostRequest( 

84 request=requests_pb2.RespondHostRequestReq( 

85 host_request_id=payload.host_request_quick_decline.host_request_id, 

86 status=conversations_pb2.HOST_REQUEST_STATUS_REJECTED, 

87 ), 

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

89 session=session, 

90 ) 

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

92 

93 raise Exception("Unhandled quick link type")