Coverage for app/backend/src/tests/fixtures/misc.py: 98%
125 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-14 11:44 +0000
1from dataclasses import dataclass
2from typing import Any
3from unittest.mock import patch
5from sqlalchemy.orm import Session
7from couchers.config import config
8from couchers.jobs.worker import process_job
9from couchers.models import User
10from couchers.notifications.push import PushNotificationContent
11from couchers.proto import moderation_pb2
12from couchers.proto.internal import jobs_pb2
13from couchers.servicers.threads import unpack_thread_id
14from tests.fixtures.sessions import real_moderation_session
17def process_jobs() -> None:
18 while process_job():
19 pass
22class EmailCollector:
23 """Intercepts emails so they can be verified by tests."""
25 def __init__(self) -> None:
26 # Collected emails by recipient address, chronologically.
27 self.by_recipient: dict[str, list[jobs_pb2.SendEmailPayload]] = {}
28 self._patch = patch("couchers.email.queuing._queue_email", self._mock_queue_email)
30 def _mock_queue_email(self, session: Session, payload: jobs_pb2.SendEmailPayload) -> None:
31 if payload.recipient not in self.by_recipient:
32 self.by_recipient[payload.recipient] = []
33 self.by_recipient[payload.recipient].append(payload)
35 def __enter__(self):
36 process_jobs() # Flush any emails prior to this point
37 self.by_recipient.clear()
38 self._patch.start()
39 return self
41 def __exit__(self, exc_type, exc_val, exc_tb):
42 self._patch.stop()
43 return False # Let any exception propagate
45 def count(self) -> int:
46 process_jobs()
47 return sum(len(v) for v in self.by_recipient.values())
49 def count_for_recipient(self, recipient: str) -> int:
50 process_jobs()
51 return len(self.by_recipient.get(recipient, []))
53 def count_for_mods(self) -> int:
54 return self.count_for_recipient(config.MODS_EMAIL_RECIPIENT)
56 def count_for_reports(self) -> int:
57 return self.count_for_recipient(config.REPORTS_EMAIL_RECIPIENT)
59 def pop_for_recipient(self, recipient: str, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
60 """
61 Removes and returns the oldest email queued to a given recipient,
62 optionally asserting that it is the last one.
63 """
64 process_jobs()
65 emails = self.by_recipient.get(recipient)
66 assert emails, f"No emails to pop for recipient {recipient}."
67 if last:
68 assert len(emails) == 1, f"Expected a single email for recipient {recipient}."
69 return emails.pop(0)
71 def pop_for_mods(self, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
72 return self.pop_for_recipient(config.MODS_EMAIL_RECIPIENT, last=last)
74 def pop_for_reports(self, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
75 return self.pop_for_recipient(config.REPORTS_EMAIL_RECIPIENT, last=last)
78@dataclass(frozen=True, slots=True, kw_only=True)
79class Push:
80 topic_action: str
81 content: PushNotificationContent
82 key: str | None = None
83 ttl: int | None = None
86class PushCollector:
87 """Captures push notifications and allows inspecting them."""
89 def __init__(self) -> None:
90 # Collected notifications by user id, chronologically.
91 self.by_user: dict[int, list[Push]] = {}
92 self._patch = patch("couchers.notifications.push._push_to_user", self._mock_push_to_user)
94 def _mock_push_to_user(self, session: Session, user_id: int, **kwargs: Any) -> None:
95 if user_id not in self.by_user:
96 self.by_user[user_id] = []
97 self.by_user[user_id].append(Push(**kwargs))
99 def __enter__(self):
100 process_jobs() # Flush any push notifications prior to this point
101 self.by_user.clear()
102 self._patch.start()
103 return self
105 def __exit__(self, exc_type, exc_val, exc_tb):
106 self._patch.stop()
107 return False # Let any exception propagate
109 def count_for_user(self, user_id: int) -> int:
110 process_jobs()
111 return len(self.by_user.get(user_id, []))
113 def pop_for_user(self, user_id: int, *, last: bool = False) -> Push:
114 """
115 Removes and returns the oldest push notification received by the given user,
116 optionally asserting that it is the last one.
117 """
118 process_jobs()
119 pushes = self.by_user.get(user_id)
120 assert pushes, f"No notifications to pop for user {user_id}."
121 if last:
122 assert len(pushes) == 1, f"Expected a single notification for user {user_id}."
123 return pushes.pop(0)
126class Moderator:
127 """
128 A test fixture that provides a moderator user and methods to exercise the moderation API.
130 Usage:
131 def test_example(db, moderator):
132 user, token = generate_user()
133 # ... create a host request ...
134 moderator.approve_host_request(host_request_id)
135 """
137 def __init__(self, user: User, token: str):
138 self.user = user
139 self.token = token
141 def approve_host_request(self, host_request_id: int, reason: str = "Test approval") -> None:
142 """
143 Approve a host request using the moderation API.
145 Args:
146 host_request_id: The conversation_id of the host request
147 reason: Optional reason for approval
148 """
149 with real_moderation_session(self.token) as api:
150 state_res = api.GetModerationState(
151 moderation_pb2.GetModerationStateReq(
152 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST,
153 object_id=host_request_id,
154 )
155 )
156 api.ModerateContent(
157 moderation_pb2.ModerateContentReq(
158 moderation_state_id=state_res.moderation_state.moderation_state_id,
159 action=moderation_pb2.MODERATION_ACTION_APPROVE,
160 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
161 reason=reason,
162 )
163 )
165 def approve_group_chat(self, group_chat_id: int, reason: str = "Test approval") -> None:
166 """
167 Approve a group chat using the moderation API.
169 Args:
170 group_chat_id: The conversation_id of the group chat
171 reason: Optional reason for approval
172 """
173 with real_moderation_session(self.token) as api:
174 state_res = api.GetModerationState(
175 moderation_pb2.GetModerationStateReq(
176 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT,
177 object_id=group_chat_id,
178 )
179 )
180 api.ModerateContent(
181 moderation_pb2.ModerateContentReq(
182 moderation_state_id=state_res.moderation_state.moderation_state_id,
183 action=moderation_pb2.MODERATION_ACTION_APPROVE,
184 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
185 reason=reason,
186 )
187 )
189 def approve_friend_request(self, friend_request_id: int, reason: str = "Test approval") -> None:
190 """
191 Approve a friend request using the moderation API.
193 Args:
194 friend_request_id: The ID of the friend request (FriendRelationship.id)
195 reason: Optional reason for approval
196 """
197 with real_moderation_session(self.token) as api:
198 state_res = api.GetModerationState(
199 moderation_pb2.GetModerationStateReq(
200 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST,
201 object_id=friend_request_id,
202 )
203 )
204 api.ModerateContent(
205 moderation_pb2.ModerateContentReq(
206 moderation_state_id=state_res.moderation_state.moderation_state_id,
207 action=moderation_pb2.MODERATION_ACTION_APPROVE,
208 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
209 reason=reason,
210 )
211 )
213 def approve_event_occurrence(self, occurrence_id: int, reason: str = "Test approval") -> None:
214 """
215 Approve an event occurrence using the moderation API.
217 Args:
218 occurrence_id: The ID of the EventOccurrence (what the proto calls event_id)
219 reason: Optional reason for approval
220 """
221 with real_moderation_session(self.token) as api:
222 state_res = api.GetModerationState(
223 moderation_pb2.GetModerationStateReq(
224 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE,
225 object_id=occurrence_id,
226 )
227 )
228 api.ModerateContent(
229 moderation_pb2.ModerateContentReq(
230 moderation_state_id=state_res.moderation_state.moderation_state_id,
231 action=moderation_pb2.MODERATION_ACTION_APPROVE,
232 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
233 reason=reason,
234 )
235 )
237 def approve_comment(self, comment_id: int, reason: str = "Test approval") -> None:
238 """Approve a Comment using the moderation API. comment_id is the database id of the Comment."""
239 with real_moderation_session(self.token) as api:
240 state_res = api.GetModerationState(
241 moderation_pb2.GetModerationStateReq(
242 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT,
243 object_id=comment_id,
244 )
245 )
246 api.ModerateContent(
247 moderation_pb2.ModerateContentReq(
248 moderation_state_id=state_res.moderation_state.moderation_state_id,
249 action=moderation_pb2.MODERATION_ACTION_APPROVE,
250 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
251 reason=reason,
252 )
253 )
255 def approve_reply(self, reply_id: int, reason: str = "Test approval") -> None:
256 """Approve a Reply using the moderation API. reply_id is the database id of the Reply."""
257 with real_moderation_session(self.token) as api:
258 state_res = api.GetModerationState(
259 moderation_pb2.GetModerationStateReq(
260 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_REPLY,
261 object_id=reply_id,
262 )
263 )
264 api.ModerateContent(
265 moderation_pb2.ModerateContentReq(
266 moderation_state_id=state_res.moderation_state.moderation_state_id,
267 action=moderation_pb2.MODERATION_ACTION_APPROVE,
268 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
269 reason=reason,
270 )
271 )
273 def approve_discussion(self, discussion_id: int, reason: str = "Test approval") -> None:
274 """Approve a Discussion using the moderation API. discussion_id is the database id of the Discussion."""
275 with real_moderation_session(self.token) as api:
276 state_res = api.GetModerationState(
277 moderation_pb2.GetModerationStateReq(
278 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION,
279 object_id=discussion_id,
280 )
281 )
282 api.ModerateContent(
283 moderation_pb2.ModerateContentReq(
284 moderation_state_id=state_res.moderation_state.moderation_state_id,
285 action=moderation_pb2.MODERATION_ACTION_APPROVE,
286 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
287 reason=reason,
288 )
289 )
291 def approve_reference(self, reference_id: int, reason: str = "Test approval") -> None:
292 """Approve a Reference using the moderation API."""
293 with real_moderation_session(self.token) as api:
294 state_res = api.GetModerationState(
295 moderation_pb2.GetModerationStateReq(
296 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE,
297 object_id=reference_id,
298 )
299 )
300 api.ModerateContent(
301 moderation_pb2.ModerateContentReq(
302 moderation_state_id=state_res.moderation_state.moderation_state_id,
303 action=moderation_pb2.MODERATION_ACTION_APPROVE,
304 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
305 reason=reason,
306 )
307 )
309 def approve_thread_post(self, packed_thread_id: int, reason: str = "Test approval") -> None:
310 """Approve whichever of Comment/Reply the packed thread_id refers to."""
311 database_id, depth = unpack_thread_id(packed_thread_id)
312 if depth == 1:
313 self.approve_comment(database_id, reason=reason)
314 elif depth == 2: 314 ↛ 317line 314 didn't jump to line 317 because the condition on line 314 was always true
315 self.approve_reply(database_id, reason=reason)
316 else:
317 raise ValueError(f"approve_thread_post: thread_id {packed_thread_id} has unsupported depth {depth}")