Coverage for app/backend/src/tests/fixtures/misc.py: 96%
125 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +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 import query_log
15from tests.fixtures.sessions import real_moderation_session
18def process_jobs() -> None:
19 # One span for the whole drain, not one per job type: Job is a frozen dataclass deriving its name and payload
20 # type from the handler's __name__ and type hints, so wrapping handlers to name them breaks get_type_hints.
21 # Splitting these out wants a span alongside the existing tracer span in worker.process_job.
22 with query_log.span("job", "process_jobs"):
23 while process_job():
24 pass
27class EmailCollector:
28 """Intercepts emails so they can be verified by tests."""
30 def __init__(self) -> None:
31 # Collected emails by recipient address, chronologically.
32 self.by_recipient: dict[str, list[jobs_pb2.SendEmailPayload]] = {}
33 self._patch = patch("couchers.email.queuing._queue_email", self._mock_queue_email)
35 def _mock_queue_email(self, session: Session, payload: jobs_pb2.SendEmailPayload) -> None:
36 if payload.recipient not in self.by_recipient:
37 self.by_recipient[payload.recipient] = []
38 self.by_recipient[payload.recipient].append(payload)
40 def __enter__(self):
41 process_jobs() # Flush any emails prior to this point
42 self.by_recipient.clear()
43 self._patch.start()
44 return self
46 def __exit__(self, exc_type, exc_val, exc_tb):
47 self._patch.stop()
48 return False # Let any exception propagate
50 def count(self) -> int:
51 process_jobs()
52 return sum(len(v) for v in self.by_recipient.values())
54 def count_for_recipient(self, recipient: str) -> int:
55 process_jobs()
56 return len(self.by_recipient.get(recipient, []))
58 def count_for_mods(self) -> int:
59 return self.count_for_recipient(config.MODS_EMAIL_RECIPIENT)
61 def count_for_reports(self) -> int:
62 return self.count_for_recipient(config.REPORTS_EMAIL_RECIPIENT)
64 def pop_for_recipient(self, recipient: str, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
65 """
66 Removes and returns the oldest email queued to a given recipient,
67 optionally asserting that it is the last one.
68 """
69 process_jobs()
70 emails = self.by_recipient.get(recipient)
71 assert emails, f"No emails to pop for recipient {recipient}."
72 if last:
73 assert len(emails) == 1, f"Expected a single email for recipient {recipient}."
74 return emails.pop(0)
76 def pop_for_mods(self, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
77 return self.pop_for_recipient(config.MODS_EMAIL_RECIPIENT, last=last)
79 def pop_for_reports(self, *, last: bool = False) -> jobs_pb2.SendEmailPayload:
80 return self.pop_for_recipient(config.REPORTS_EMAIL_RECIPIENT, last=last)
83@dataclass(frozen=True, slots=True, kw_only=True)
84class Push:
85 topic_action: str
86 content: PushNotificationContent
87 key: str | None = None
88 ttl: int | None = None
91class PushCollector:
92 """Captures push notifications and allows inspecting them."""
94 def __init__(self) -> None:
95 # Collected notifications by user id, chronologically.
96 self.by_user: dict[int, list[Push]] = {}
97 self._patch = patch("couchers.notifications.push._push_to_user", self._mock_push_to_user)
99 def _mock_push_to_user(self, session: Session, user_id: int, **kwargs: Any) -> None:
100 if user_id not in self.by_user:
101 self.by_user[user_id] = []
102 self.by_user[user_id].append(Push(**kwargs))
104 def __enter__(self):
105 process_jobs() # Flush any push notifications prior to this point
106 self.by_user.clear()
107 self._patch.start()
108 return self
110 def __exit__(self, exc_type, exc_val, exc_tb):
111 self._patch.stop()
112 return False # Let any exception propagate
114 def count_for_user(self, user_id: int) -> int:
115 process_jobs()
116 return len(self.by_user.get(user_id, []))
118 def pop_for_user(self, user_id: int, *, last: bool = False) -> Push:
119 """
120 Removes and returns the oldest push notification received by the given user,
121 optionally asserting that it is the last one.
122 """
123 process_jobs()
124 pushes = self.by_user.get(user_id)
125 assert pushes, f"No notifications to pop for user {user_id}."
126 if last:
127 assert len(pushes) == 1, f"Expected a single notification for user {user_id}."
128 return pushes.pop(0)
131class Moderator:
132 """
133 A test fixture that provides a moderator user and methods to exercise the moderation API.
135 Usage:
136 def test_example(db, moderator):
137 user, token = generate_user()
138 # ... create a host request ...
139 moderator.approve_host_request(host_request_id)
140 """
142 def __init__(self, user: User, token: str):
143 self.user = user
144 self.token = token
146 def set_visibility(
147 self,
148 object_type: moderation_pb2.ModerationObjectType.ValueType,
149 object_id: int,
150 visibility: moderation_pb2.ModerationVisibility.ValueType,
151 reason: str = "Test moderation",
152 ) -> None:
153 """Move a piece of moderated content to the given visibility through the moderation API."""
154 raising = visibility in (
155 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
156 moderation_pb2.MODERATION_VISIBILITY_UNLISTED,
157 )
158 with real_moderation_session(self.token) as api:
159 state_res = api.GetModerationState(
160 moderation_pb2.GetModerationStateReq(object_type=object_type, object_id=object_id)
161 )
162 api.ModerateContent(
163 moderation_pb2.ModerateContentReq(
164 moderation_state_id=state_res.moderation_state.moderation_state_id,
165 action=(
166 moderation_pb2.MODERATION_ACTION_APPROVE if raising else moderation_pb2.MODERATION_ACTION_HIDE
167 ),
168 visibility=visibility,
169 reason=reason,
170 clear_flags=True,
171 )
172 )
174 def approve_host_request(self, host_request_id: int, reason: str = "Test approval") -> None:
175 """host_request_id is the conversation_id of the host request."""
176 self.set_visibility(
177 moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST,
178 host_request_id,
179 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
180 reason,
181 )
183 def hide_host_request(self, host_request_id: int, reason: str = "Test hide") -> None:
184 """host_request_id is the conversation_id of the host request."""
185 self.set_visibility(
186 moderation_pb2.MODERATION_OBJECT_TYPE_HOST_REQUEST,
187 host_request_id,
188 moderation_pb2.MODERATION_VISIBILITY_HIDDEN,
189 reason,
190 )
192 def approve_group_chat(self, group_chat_id: int, reason: str = "Test approval") -> None:
193 """group_chat_id is the conversation_id of the group chat."""
194 self.set_visibility(
195 moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT,
196 group_chat_id,
197 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
198 reason,
199 )
201 def set_group_chat_visibility(
202 self,
203 group_chat_id: int,
204 visibility: moderation_pb2.ModerationVisibility.ValueType,
205 reason: str = "Test moderation",
206 ) -> None:
207 self.set_visibility(moderation_pb2.MODERATION_OBJECT_TYPE_GROUP_CHAT, group_chat_id, visibility, reason)
209 def approve_friend_request(self, friend_request_id: int, reason: str = "Test approval") -> None:
210 """friend_request_id is the FriendRelationship id."""
211 self.set_visibility(
212 moderation_pb2.MODERATION_OBJECT_TYPE_FRIEND_REQUEST,
213 friend_request_id,
214 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
215 reason,
216 )
218 def approve_event_occurrence(self, occurrence_id: int, reason: str = "Test approval") -> None:
219 """occurrence_id is the EventOccurrence id, which is what the proto calls event_id."""
220 self.set_visibility(
221 moderation_pb2.MODERATION_OBJECT_TYPE_EVENT_OCCURRENCE,
222 occurrence_id,
223 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
224 reason,
225 )
227 def approve_comment(self, comment_id: int, reason: str = "Test approval") -> None:
228 """comment_id is the database id of the Comment."""
229 self.set_visibility(
230 moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT,
231 comment_id,
232 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
233 reason,
234 )
236 def approve_reply(self, reply_id: int, reason: str = "Test approval") -> None:
237 """reply_id is the database id of the Reply."""
238 self.set_visibility(
239 moderation_pb2.MODERATION_OBJECT_TYPE_REPLY,
240 reply_id,
241 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
242 reason,
243 )
245 def approve_discussion(self, discussion_id: int, reason: str = "Test approval") -> None:
246 """discussion_id is the database id of the Discussion."""
247 self.set_visibility(
248 moderation_pb2.MODERATION_OBJECT_TYPE_DISCUSSION,
249 discussion_id,
250 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
251 reason,
252 )
254 def approve_reference(self, reference_id: int, reason: str = "Test approval") -> None:
255 self.set_visibility(
256 moderation_pb2.MODERATION_OBJECT_TYPE_REFERENCE,
257 reference_id,
258 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
259 reason,
260 )
262 def approve_public_trip(self, public_trip_id: int, reason: str = "Test approval") -> None:
263 self.set_visibility(
264 moderation_pb2.MODERATION_OBJECT_TYPE_PUBLIC_TRIP,
265 public_trip_id,
266 moderation_pb2.MODERATION_VISIBILITY_VISIBLE,
267 reason,
268 )
270 def approve_thread_post(self, packed_thread_id: int, reason: str = "Test approval") -> None:
271 """Approve whichever of Comment/Reply the packed thread_id refers to."""
272 self.set_thread_post_visibility(packed_thread_id, moderation_pb2.MODERATION_VISIBILITY_VISIBLE, reason)
274 def set_thread_post_visibility(
275 self,
276 packed_thread_id: int,
277 visibility: moderation_pb2.ModerationVisibility.ValueType,
278 reason: str = "Test moderation",
279 ) -> None:
280 """Moderate whichever of Comment/Reply the packed thread_id refers to."""
281 database_id, depth = unpack_thread_id(packed_thread_id)
282 if depth == 1:
283 object_type = moderation_pb2.MODERATION_OBJECT_TYPE_COMMENT
284 elif depth == 2: 284 ↛ 287line 284 didn't jump to line 287 because the condition on line 284 was always true
285 object_type = moderation_pb2.MODERATION_OBJECT_TYPE_REPLY
286 else:
287 raise ValueError(f"thread_id {packed_thread_id} has unsupported depth {depth}")
288 self.set_visibility(object_type, database_id, visibility, reason)