Coverage for app/backend/src/tests/fixtures/misc.py: 98%
133 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +0000
1from dataclasses import dataclass
2from datetime import datetime, timedelta
3from typing import Any
4from unittest.mock import patch
6from sqlalchemy.orm import Session
8from couchers.config import config
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 couchers.proto.internal import jobs_pb2
14from couchers.servicers.threads import unpack_thread_id
15from couchers.utils import now
16from tests.fixtures.sessions import real_moderation_session
19def process_jobs() -> None:
20 while process_job():
21 pass
24def now_5_min_in_future() -> datetime:
25 return now() + timedelta(minutes=5)
28class EmailCollector:
29 """Intercepts emails so they can be verified by tests."""
31 def __init__(self) -> None:
32 # Collected emails by recipient address, chronologically.
33 self.by_recipient: dict[str, list[jobs_pb2.SendEmailPayload]] = {}
34 self._patch = patch("couchers.email.queuing._queue_email", self._mock_queue_email)
36 def _mock_queue_email(self, session: Session, payload: jobs_pb2.SendEmailPayload) -> None:
37 if payload.recipient not in self.by_recipient:
38 self.by_recipient[payload.recipient] = []
39 self.by_recipient[payload.recipient].append(payload)
41 def __enter__(self):
42 process_jobs() # Flush any emails prior to this point
43 self.by_recipient.clear()
44 self._patch.start()
45 return self
47 def __exit__(self, exc_type, exc_val, exc_tb):
48 self._patch.stop()
49 return False # Let any exception propagate
51 def count(self) -> int:
52 process_jobs()
53 return sum(len(v) for v in self.by_recipient.values())
55 def count_for_recipient(self, recipient: str) -> int:
56 process_jobs()
57 return len(self.by_recipient.get(recipient, []))
59 def count_for_mods(self) -> int:
60 return self.count_for_recipient(config.MODS_EMAIL_RECIPIENT)
62 def count_for_reports(self) -> int:
63 return self.count_for_recipient(config.REPORTS_EMAIL_RECIPIENT)
65 def pop_for_recipient(self, recipient: str, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
66 """
67 Removes and returns the oldest email queued to a given recipient,
68 optionally asserting that it is the last one.
69 """
70 process_jobs()
71 emails = self.by_recipient.get(recipient)
72 assert emails, f"No emails to pop for recipient {recipient}."
73 if last:
74 assert len(emails) == 1, f"Expected a single email for recipient {recipient}."
75 return emails.pop(0)
77 def pop_for_mods(self, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
78 return self.pop_for_recipient(config.MODS_EMAIL_RECIPIENT, last=last)
80 def pop_for_reports(self, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
81 return self.pop_for_recipient(config.REPORTS_EMAIL_RECIPIENT, last=last)
84@dataclass(frozen=True, slots=True, kw_only=True)
85class Push:
86 topic_action: str
87 content: PushNotificationContent
88 key: str | None = None
89 ttl: int | None = None
92class PushCollector:
93 """Captures push notifications and allows inspecting them."""
95 def __init__(self) -> None:
96 # Collected notifications by user id, chronologically.
97 self.by_user: dict[int, list[Push]] = {}
98 self._patch = patch("couchers.notifications.push._push_to_user", self._mock_push_to_user)
100 def _mock_push_to_user(self, session: Session, user_id: int, **kwargs: Any) -> None:
101 if user_id not in self.by_user:
102 self.by_user[user_id] = []
103 self.by_user[user_id].append(Push(**kwargs))
105 def __enter__(self):
106 process_jobs() # Flush any push notifications prior to this point
107 self.by_user.clear()
108 self._patch.start()
109 return self
111 def __exit__(self, exc_type, exc_val, exc_tb):
112 self._patch.stop()
113 return False # Let any exception propagate
115 def count_for_user(self, user_id: int) -> int:
116 process_jobs()
117 return len(self.by_user.get(user_id, []))
119 def pop_for_user(self, user_id: int, *, last: bool = False) -> Push:
120 """
121 Removes and returns the oldest push notification received by the given user,
122 optionally asserting that it is the last one.
123 """
124 process_jobs()
125 pushes = self.by_user.get(user_id)
126 assert pushes, f"No notifications to pop for user {user_id}."
127 if last:
128 assert len(pushes) == 1, f"Expected a single notification for user {user_id}."
129 return pushes.pop(0)
132class Moderator:
133 """
134 A test fixture that provides a moderator user and methods to exercise the moderation API.
136 Usage:
137 def test_example(db, moderator):
138 user, token = generate_user()
139 # ... create a host request ...
140 moderator.approve_host_request(host_request_id)
141 """
143 def __init__(self, user: User, token: str):
144 self.user = user
145 self.token = token
147 def approve_host_request(self, host_request_id: int, reason: str = "Test approval") -> None:
148 """
149 Approve a host request using the moderation API.
151 Args:
152 host_request_id: The conversation_id of the host request
153 reason: Optional reason for approval
154 """
155 with real_moderation_session(self.token) as api:
156 state_res = api.GetModerationState(
157 moderation_pb2.GetModerationStateReq(
158 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST,
159 object_id=host_request_id,
160 )
161 )
162 api.ModerateContent(
163 moderation_pb2.ModerateContentReq(
164 moderation_state_id=state_res.moderation_state.moderation_state_id,
165 action=moderation_pb2.MODERATION_ACTION_APPROVE,
166 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
167 reason=reason,
168 clear_flags=True,
169 )
170 )
172 def approve_group_chat(self, group_chat_id: int, reason: str = "Test approval") -> None:
173 """
174 Approve a group chat using the moderation API.
176 Args:
177 group_chat_id: The conversation_id of the group chat
178 reason: Optional reason for approval
179 """
180 with real_moderation_session(self.token) as api:
181 state_res = api.GetModerationState(
182 moderation_pb2.GetModerationStateReq(
183 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT,
184 object_id=group_chat_id,
185 )
186 )
187 api.ModerateContent(
188 moderation_pb2.ModerateContentReq(
189 moderation_state_id=state_res.moderation_state.moderation_state_id,
190 action=moderation_pb2.MODERATION_ACTION_APPROVE,
191 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
192 reason=reason,
193 clear_flags=True,
194 )
195 )
197 def approve_friend_request(self, friend_request_id: int, reason: str = "Test approval") -> None:
198 """
199 Approve a friend request using the moderation API.
201 Args:
202 friend_request_id: The ID of the friend request (FriendRelationship.id)
203 reason: Optional reason for approval
204 """
205 with real_moderation_session(self.token) as api:
206 state_res = api.GetModerationState(
207 moderation_pb2.GetModerationStateReq(
208 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST,
209 object_id=friend_request_id,
210 )
211 )
212 api.ModerateContent(
213 moderation_pb2.ModerateContentReq(
214 moderation_state_id=state_res.moderation_state.moderation_state_id,
215 action=moderation_pb2.MODERATION_ACTION_APPROVE,
216 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
217 reason=reason,
218 clear_flags=True,
219 )
220 )
222 def approve_event_occurrence(self, occurrence_id: int, reason: str = "Test approval") -> None:
223 """
224 Approve an event occurrence using the moderation API.
226 Args:
227 occurrence_id: The ID of the EventOccurrence (what the proto calls event_id)
228 reason: Optional reason for approval
229 """
230 with real_moderation_session(self.token) as api:
231 state_res = api.GetModerationState(
232 moderation_pb2.GetModerationStateReq(
233 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE,
234 object_id=occurrence_id,
235 )
236 )
237 api.ModerateContent(
238 moderation_pb2.ModerateContentReq(
239 moderation_state_id=state_res.moderation_state.moderation_state_id,
240 action=moderation_pb2.MODERATION_ACTION_APPROVE,
241 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
242 reason=reason,
243 clear_flags=True,
244 )
245 )
247 def approve_comment(self, comment_id: int, reason: str = "Test approval") -> None:
248 """Approve a Comment using the moderation API. comment_id is the database id of the Comment."""
249 with real_moderation_session(self.token) as api:
250 state_res = api.GetModerationState(
251 moderation_pb2.GetModerationStateReq(
252 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT,
253 object_id=comment_id,
254 )
255 )
256 api.ModerateContent(
257 moderation_pb2.ModerateContentReq(
258 moderation_state_id=state_res.moderation_state.moderation_state_id,
259 action=moderation_pb2.MODERATION_ACTION_APPROVE,
260 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
261 reason=reason,
262 clear_flags=True,
263 )
264 )
266 def approve_reply(self, reply_id: int, reason: str = "Test approval") -> None:
267 """Approve a Reply using the moderation API. reply_id is the database id of the Reply."""
268 with real_moderation_session(self.token) as api:
269 state_res = api.GetModerationState(
270 moderation_pb2.GetModerationStateReq(
271 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_REPLY,
272 object_id=reply_id,
273 )
274 )
275 api.ModerateContent(
276 moderation_pb2.ModerateContentReq(
277 moderation_state_id=state_res.moderation_state.moderation_state_id,
278 action=moderation_pb2.MODERATION_ACTION_APPROVE,
279 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
280 reason=reason,
281 clear_flags=True,
282 )
283 )
285 def approve_discussion(self, discussion_id: int, reason: str = "Test approval") -> None:
286 """Approve a Discussion using the moderation API. discussion_id is the database id of the Discussion."""
287 with real_moderation_session(self.token) as api:
288 state_res = api.GetModerationState(
289 moderation_pb2.GetModerationStateReq(
290 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION,
291 object_id=discussion_id,
292 )
293 )
294 api.ModerateContent(
295 moderation_pb2.ModerateContentReq(
296 moderation_state_id=state_res.moderation_state.moderation_state_id,
297 action=moderation_pb2.MODERATION_ACTION_APPROVE,
298 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
299 reason=reason,
300 clear_flags=True,
301 )
302 )
304 def approve_reference(self, reference_id: int, reason: str = "Test approval") -> None:
305 """Approve a Reference using the moderation API."""
306 with real_moderation_session(self.token) as api:
307 state_res = api.GetModerationState(
308 moderation_pb2.GetModerationStateReq(
309 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE,
310 object_id=reference_id,
311 )
312 )
313 api.ModerateContent(
314 moderation_pb2.ModerateContentReq(
315 moderation_state_id=state_res.moderation_state.moderation_state_id,
316 action=moderation_pb2.MODERATION_ACTION_APPROVE,
317 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
318 reason=reason,
319 clear_flags=True,
320 )
321 )
323 def approve_public_trip(self, public_trip_id: int, reason: str = "Test approval") -> None:
324 """Approve a PublicTrip using the moderation API."""
325 with real_moderation_session(self.token) as api:
326 state_res = api.GetModerationState(
327 moderation_pb2.GetModerationStateReq(
328 object_type=moderation_pb2.MODERATION_OBJECT_TYPE_PUBLIC_TRIP,
329 object_id=public_trip_id,
330 )
331 )
332 api.ModerateContent(
333 moderation_pb2.ModerateContentReq(
334 moderation_state_id=state_res.moderation_state.moderation_state_id,
335 action=moderation_pb2.MODERATION_ACTION_APPROVE,
336 visibility=moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
337 reason=reason,
338 clear_flags=True,
339 )
340 )
342 def approve_thread_post(self, packed_thread_id: int, reason: str = "Test approval") -> None:
343 """Approve whichever of Comment/Reply the packed thread_id refers to."""
344 database_id, depth = unpack_thread_id(packed_thread_id)
345 if depth == 1:
346 self.approve_comment(database_id, reason=reason)
347 elif depth == 2: 347 ↛ 350line 347 didn't jump to line 350 because the condition on line 347 was always true
348 self.approve_reply(database_id, reason=reason)
349 else:
350 raise ValueError(f"approve_thread_post: thread_id {packed_thread_id} has unsupported depth {depth}")