Coverage for app / backend / src / tests / fixtures / misc.py: 100%

1from collections.abc import Generator 

2from contextlib import contextmanager 

3from dataclasses import dataclass 

4from typing import Any 

5from unittest.mock import Mock, patch 

6 

7from sqlalchemy.orm import Session 

8 

9from couchers.jobs.worker import process_job 

10from couchers.models import User 

11from couchers.notifications.push import PushNotificationContent 

12from couchers.proto import moderation_pb2 

13from tests.fixtures.sessions import real_moderation_session 

14 

15 

16def process_jobs() -> None: 

17 while process_job(): 

18 pass 

19 

20 

21@contextmanager 

22def mock_notification_email() -> Generator[Mock]: 

23 with patch("couchers.email._queue_email") as mock: 

24 yield mock 

25 process_jobs() 

26 

27 

28@dataclass 

29class EmailData: 

30 sender_name: str 

31 sender_email: str 

32 recipient: str 

33 subject: str 

34 plain: str 

35 html: str 

36 source_data: str 

37 list_unsubscribe_header: str 

38 

39 

40def email_fields(mock: Mock, call_ix: int = 0) -> EmailData: 

41 _, kw = mock.call_args_list[call_ix] 

42 return EmailData( 

43 sender_name=kw.get("sender_name"), 

44 sender_email=kw.get("sender_email"), 

45 recipient=kw.get("recipient"), 

46 subject=kw.get("subject"), 

47 plain=kw.get("plain"), 

48 html=kw.get("html"), 

49 source_data=kw.get("source_data"), 

50 list_unsubscribe_header=kw.get("list_unsubscribe_header"), 

51 ) 

52 

53 

54@dataclass(frozen=True, slots=True, kw_only=True) 

55class Push: 

56 topic_action: str 

57 content: PushNotificationContent 

58 key: str | None = None 

59 ttl: int | None = None 

60 

61 

62class PushCollector: 

63 def __init__(self): 

64 self.by_user: dict[int, list[Push]] = {} 

65 """Collected notifications by user id, chronologically.""" 

66 

67 def push_to_user(self, session: Session, user_id: int, **kwargs: Any) -> None: 

68 if user_id not in self.by_user: 

69 self.by_user[user_id] = [] 

70 self.by_user[user_id].append(Push(**kwargs)) 

71 

72 def count_for_user(self, user_id: int) -> int: 

73 return len(self.by_user.get(user_id, [])) 

74 

75 def pop_for_user(self, user_id: int, last: bool = False) -> Push: 

76 """ 

77 Removes and returns the oldest push notification received by the given user, 

78 optionally asserting that it is the last one. 

79 """ 

80 pushes = self.by_user.get(user_id) 

81 assert pushes, f"No notifications to pop for user {user_id}." 

82 if last: 

83 assert len(pushes) == 1, f"Expected a single notification for user {user_id}." 

84 return pushes.pop(0) 

85 

86 

87class Moderator: 

88 """ 

89 A test fixture that provides a moderator user and methods to exercise the moderation API. 

90 

91 Usage: 

92 def test_example(db, moderator): 

93 user, token = generate_user() 

94 # ... create a host request ... 

95 moderator.approve_host_request(host_request_id) 

96 """ 

97 

98 def __init__(self, user: User, token: str): 

99 self.user = user 

100 self.token = token 

101 

102 def approve_host_request(self, host_request_id: int, reason: str = "Test approval") -> None: 

103 """ 

104 Approve a host request using the moderation API. 

105 

106 Args: 

107 host_request_id: The conversation_id of the host request 

108 reason: Optional reason for approval 

109 """ 

110 with real_moderation_session(self.token) as api: 

111 state_res = api.GetModerationState( 

112 moderation_pb2.GetModerationStateReq( 

113 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST, 

114 object_id=host_request_id, 

115 ) 

116 ) 

117 api.ModerateContent( 

118 moderation_pb2.ModerateContentReq( 

119 moderation_state_id=state_res.moderation_state.moderation_state_id, 

120 action=moderation_pb2.MODERATION_ACTION_APPROVE, 

121 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE, 

122 reason=reason, 

123 ) 

124 ) 

125 

126 def approve_group_chat(self, group_chat_id: int, reason: str = "Test approval") -> None: 

127 """ 

128 Approve a group chat using the moderation API. 

129 

130 Args: 

131 group_chat_id: The conversation_id of the group chat 

132 reason: Optional reason for approval 

133 """ 

134 with real_moderation_session(self.token) as api: 

135 state_res = api.GetModerationState( 

136 moderation_pb2.GetModerationStateReq( 

137 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT, 

138 object_id=group_chat_id, 

139 ) 

140 ) 

141 api.ModerateContent( 

142 moderation_pb2.ModerateContentReq( 

143 moderation_state_id=state_res.moderation_state.moderation_state_id, 

144 action=moderation_pb2.MODERATION_ACTION_APPROVE, 

145 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE, 

146 reason=reason, 

147 ) 

148 )