Coverage for app/backend/src/tests/test_admin.py: 100%
1195 statements
« prev ^ index » next coverage.py v7.17.0, created at 2026-11-03 11:56 +0000
1import json
2from datetime import UTC, date, datetime, timedelta
3from unittest.mock import patch
5import grpc
6import pytest
7from google.protobuf import wrappers_pb2
8from sqlalchemy import select, update
9from sqlalchemy.sql import func
11from couchers.db import session_scope
12from couchers.models import (
13 AccountDeletionToken,
14 ContentReport,
15 EventOccurrence,
16 FriendRelationship,
17 FriendStatus,
18 HostRequest,
19 ModerationObjectType,
20 ModerationState,
21 ModerationUserList,
22 ModerationVisibility,
23 NonvisibleUserAccess,
24 NonvisibleUserAccessType,
25 NonvisibleUserState,
26 PhotoGallery,
27 PhotoGalleryItem,
28 Reference,
29 Upload,
30 User,
31 UserActivity,
32 UserSession,
33)
34from couchers.proto import (
35 account_pb2,
36 admin_pb2,
37 auth_pb2,
38 blocking_pb2,
39 references_pb2,
40 reporting_pb2,
41 requests_pb2,
42)
43from couchers.utils import Timestamp_from_datetime, date_to_api, now, parse_date
44from tests.fixtures.api import events as events_api
45from tests.fixtures.db import (
46 add_users_to_new_moderation_list,
47 backdate_conversations,
48 generate_user,
49 make_friends,
50)
51from tests.fixtures.misc import EmailCollector, Moderator, PushCollector
52from tests.fixtures.sessions import (
53 account_session,
54 auth_api_session,
55 blocking_session,
56 real_admin_session,
57 references_session,
58 reporting_session,
59 requests_session,
60)
61from tests.test_communities import create_community
62from tests.test_references import create_host_reference
63from tests.test_requests import valid_request_text
66def test_access_by_normal_user(db):
67 normal_user, normal_token = generate_user()
69 with real_admin_session(normal_token) as api:
70 # all requests to the admin servicer should break when done by a non-super_user
71 with pytest.raises(grpc.RpcError) as e:
72 api.GetUserDetails(
73 admin_pb2.GetUserDetailsReq(
74 user=str(normal_user.id),
75 )
76 )
77 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
80def test_GetNonvisibleUserAccessLog(db):
81 super_user, super_token = generate_user(is_superuser=True)
82 target, _ = generate_user(username="target")
83 viewer, _ = generate_user(username="viewer")
85 with session_scope() as session:
86 session.add(
87 NonvisibleUserAccess(
88 access_type=NonvisibleUserAccessType.login_attempt,
89 target_user_id=target.id,
90 target_state=NonvisibleUserState.banned,
91 actor_user_id=target.id,
92 ip_address="1.2.3.4",
93 sofa="device-cookie",
94 )
95 )
96 session.add(
97 NonvisibleUserAccess(
98 access_type=NonvisibleUserAccessType.ghost_served,
99 target_user_id=target.id,
100 target_state=NonvisibleUserState.banned,
101 actor_user_id=viewer.id,
102 )
103 )
104 session.add(
105 NonvisibleUserAccess(
106 access_type=NonvisibleUserAccessType.ghost_served,
107 target_user_id=target.id,
108 target_state=NonvisibleUserState.banned,
109 actor_user_id=None,
110 )
111 )
113 with real_admin_session(super_token) as api:
114 res = api.GetNonvisibleUserAccessLog(admin_pb2.GetNonvisibleUserAccessLogReq(user="target"))
116 assert len(res.entries) == 3
117 for entry in res.entries:
118 assert entry.target_user_id == target.id
119 assert entry.target_state == admin_pb2.NONVISIBLE_USER_STATE_BANNED
121 login = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_LOGIN_ATTEMPT]
122 views = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_GHOST_SERVED]
123 assert len(login) == 1
124 assert len(views) == 2
126 assert login[0].actor_user_id.value == target.id
127 assert login[0].actor_username == "target"
128 assert login[0].ip_address == "1.2.3.4"
129 assert login[0].sofa == "device-cookie"
131 logged_in_view = [e for e in views if e.actor_username == "viewer"]
132 logged_out_view = [e for e in views if not e.actor_username]
133 assert len(logged_in_view) == 1
134 assert logged_in_view[0].actor_user_id.value == viewer.id
135 assert len(logged_out_view) == 1
136 assert not logged_out_view[0].HasField("actor_user_id")
139def test_GetUser(db):
140 super_user, super_token = generate_user(is_superuser=True)
141 normal_user, normal_token = generate_user()
143 with real_admin_session(super_token) as api:
144 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
145 assert res.user_id == normal_user.id
146 assert res.username == normal_user.username
148 with real_admin_session(super_token) as api:
149 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note="Testing banning"))
151 with real_admin_session(super_token) as api:
152 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
153 assert res.user_id == normal_user.id
154 assert res.username == normal_user.username
157def test_GetUserDetails(db):
158 super_user, super_token = generate_user(is_superuser=True)
159 normal_user, normal_token = generate_user()
161 with real_admin_session(super_token) as api:
162 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=str(normal_user.id)))
163 assert res.user_id == normal_user.id
164 assert res.username == normal_user.username
165 assert res.email == normal_user.email
166 assert res.gender == normal_user.gender
167 assert parse_date(res.birthdate) == normal_user.birthdate
168 assert not res.banned
169 assert not res.deleted
171 with real_admin_session(super_token) as api:
172 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
173 assert res.user_id == normal_user.id
174 assert res.username == normal_user.username
175 assert res.email == normal_user.email
176 assert res.gender == normal_user.gender
177 assert parse_date(res.birthdate) == normal_user.birthdate
178 assert not res.banned
179 assert not res.deleted
181 with real_admin_session(super_token) as api:
182 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.email))
183 assert res.user_id == normal_user.id
184 assert res.username == normal_user.username
185 assert res.email == normal_user.email
186 assert res.gender == normal_user.gender
187 assert parse_date(res.birthdate) == normal_user.birthdate
188 assert not res.banned
189 assert not res.deleted
192def test_ChangeUserGender(db, email_collector: EmailCollector, push_collector: PushCollector):
193 super_user, super_token = generate_user(is_superuser=True)
194 normal_user, normal_token = generate_user()
196 with real_admin_session(super_token) as api:
197 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
198 assert res.user_id == normal_user.id
199 assert res.username == normal_user.username
200 assert res.email == normal_user.email
201 assert res.gender == "Machine"
202 assert parse_date(res.birthdate) == normal_user.birthdate
203 assert not res.banned
204 assert not res.deleted
206 email = email_collector.pop_for_recipient(normal_user.email, last=True)
207 assert email.subject == "[TEST] Your gender was changed"
208 assert email.recipient == normal_user.email
209 assert "Machine" in email.plain
210 assert "Machine" in email.html
212 push = push_collector.pop_for_user(normal_user.id, last=True)
213 assert push.content.title == "Gender changed"
214 assert push.content.body == "An admin changed your gender to Machine."
217def test_ChangeUserBirthdate(db, email_collector: EmailCollector, push_collector: PushCollector):
218 super_user, super_token = generate_user(is_superuser=True)
219 normal_user, normal_token = generate_user(birthdate=date(year=2000, month=1, day=1))
221 with real_admin_session(super_token) as api:
222 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
223 assert parse_date(res.birthdate) == date(year=2000, month=1, day=1)
225 res = api.ChangeUserBirthdate(
226 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-05-25")
227 )
229 assert res.user_id == normal_user.id
230 assert res.username == normal_user.username
231 assert res.email == normal_user.email
232 assert res.birthdate == "1990-05-25"
233 assert res.gender == normal_user.gender
234 assert not res.banned
235 assert not res.deleted
237 email = email_collector.pop_for_recipient(normal_user.email, last=True)
238 assert email.subject == "[TEST] Your date of birth was changed"
239 assert email.recipient == normal_user.email
240 assert "1990" in email.plain
241 assert "1990" in email.html
243 push = push_collector.pop_for_user(normal_user.id, last=True)
244 assert push.content.title == "Birthdate changed"
245 assert push.content.body == "An admin changed your date of birth to May 25, 1990."
248def test_BanUser(db):
249 super_user, super_token = generate_user(is_superuser=True)
250 normal_user, _ = generate_user()
251 admin_note = "A good reason"
253 with real_admin_session(super_token) as api:
254 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note=admin_note))
255 assert res.user_id == normal_user.id
256 assert res.username == normal_user.username
257 assert res.email == normal_user.email
258 assert res.gender == normal_user.gender
259 assert parse_date(res.birthdate) == normal_user.birthdate
260 assert res.banned
261 assert not res.deleted
262 assert len(res.admin_actions) == 1
263 assert res.admin_actions[0].action_type == "ban"
264 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
265 assert res.admin_actions[0].note == admin_note
266 assert res.admin_actions[0].admin_user_id == super_user.id
267 assert res.admin_actions[0].admin_username == super_user.username
270def test_UnbanUser(db):
271 super_user, super_token = generate_user(is_superuser=True)
272 normal_user, _ = generate_user()
273 admin_note = "A good reason"
275 with real_admin_session(super_token) as api:
276 res = api.UnbanUser(admin_pb2.UnbanUserReq(user=normal_user.username, admin_note=admin_note))
277 assert res.user_id == normal_user.id
278 assert res.username == normal_user.username
279 assert res.email == normal_user.email
280 assert res.gender == normal_user.gender
281 assert parse_date(res.birthdate) == normal_user.birthdate
282 assert not res.banned
283 assert not res.deleted
284 assert len(res.admin_actions) == 1
285 assert res.admin_actions[0].action_type == "unban"
286 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
289def test_ShadowUser(db, moderator: Moderator):
290 super_user, super_token = generate_user(is_superuser=True)
291 surfer, surfer_token = generate_user()
292 host, _ = generate_user()
293 admin_note = "Spammer"
295 # Create a host request from `surfer` and approve its moderation state to VISIBLE so we can verify the cascade
296 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
297 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
298 with requests_session(surfer_token) as api:
299 host_request_id = api.CreateHostRequest(
300 requests_pb2.CreateHostRequestReq(
301 host_user_id=host.id,
302 from_date=today_plus_2,
303 to_date=today_plus_3,
304 text=valid_request_text(),
305 )
306 ).host_request_id
307 moderator.approve_host_request(host_request_id)
309 with real_admin_session(super_token) as api:
310 res = api.ShadowUser(admin_pb2.ShadowUserReq(user=surfer.username, admin_note=admin_note))
311 assert res.user_id == surfer.id
312 assert res.shadowed
313 assert not res.banned
314 assert not res.deleted
315 assert len(res.admin_actions) == 1
316 assert res.admin_actions[0].action_type == "shadow"
317 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
318 assert res.admin_actions[0].note == admin_note
320 # The previously-visible host request is now shadowed
321 with session_scope() as session:
322 state = session.execute(
323 select(ModerationState)
324 .where(ModerationState.object_type == ModerationObjectType.host_request)
325 .where(ModerationState.object_id == host_request_id)
326 ).scalar_one()
327 assert state.visibility == ModerationVisibility.shadowed
330def test_UnshadowUser(db, moderator: Moderator):
331 super_user, super_token = generate_user(is_superuser=True)
332 surfer, surfer_token = generate_user()
333 host, _ = generate_user()
335 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
336 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
337 with requests_session(surfer_token) as api:
338 shadow_cascade_request_id = api.CreateHostRequest(
339 requests_pb2.CreateHostRequestReq(
340 host_user_id=host.id,
341 from_date=today_plus_2,
342 to_date=today_plus_3,
343 text=valid_request_text(),
344 )
345 ).host_request_id
346 backdate_conversations()
347 admin_hidden_request_id = api.CreateHostRequest(
348 requests_pb2.CreateHostRequestReq(
349 host_user_id=host.id,
350 from_date=today_plus_2,
351 to_date=today_plus_3,
352 text=valid_request_text(),
353 )
354 ).host_request_id
356 moderator.hide_host_request(admin_hidden_request_id)
357 with session_scope() as session:
358 session.execute(select(User).where(User.id == surfer.id)).scalar_one().shadowed_at = now()
360 with real_admin_session(super_token) as api:
361 res = api.UnshadowUser(admin_pb2.UnshadowUserReq(user=surfer.username, admin_note="rehabilitated"))
362 assert not res.shadowed
363 assert len(res.admin_actions) == 1
364 assert res.admin_actions[0].action_type == "unshadow"
365 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
367 with session_scope() as session:
368 assert (
369 session.execute(
370 select(ModerationState)
371 .where(ModerationState.object_type == ModerationObjectType.host_request)
372 .where(ModerationState.object_id == shadow_cascade_request_id)
373 )
374 .scalar_one()
375 .visibility
376 == ModerationVisibility.visible
377 )
378 assert (
379 session.execute(
380 select(ModerationState)
381 .where(ModerationState.object_type == ModerationObjectType.host_request)
382 .where(ModerationState.object_id == admin_hidden_request_id)
383 )
384 .scalar_one()
385 .visibility
386 == ModerationVisibility.hidden
387 )
390def test_SetModerationAutoApproveDeadline(db):
391 super_user, super_token = generate_user(is_superuser=True)
392 normal_user, _ = generate_user()
394 with real_admin_session(super_token) as api:
395 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
396 assert not res.HasField("moderation_auto_approve_seconds")
398 res = api.SetModerationAutoApproveDeadline(
399 admin_pb2.SetModerationAutoApproveDeadlineReq(
400 user=normal_user.username, auto_approve_seconds=wrappers_pb2.Int64Value(value=0)
401 )
402 )
403 assert res.moderation_auto_approve_seconds.value == 0
405 res = api.SetModerationAutoApproveDeadline(
406 admin_pb2.SetModerationAutoApproveDeadlineReq(
407 user=normal_user.username, auto_approve_seconds=wrappers_pb2.Int64Value(value=48 * 3600)
408 )
409 )
410 assert res.moderation_auto_approve_seconds.value == 48 * 3600
412 res = api.SetModerationAutoApproveDeadline(
413 admin_pb2.SetModerationAutoApproveDeadlineReq(user=normal_user.username)
414 )
415 assert not res.HasField("moderation_auto_approve_seconds")
417 for seconds in [-1, 48 * 3600 + 1, 2**63 - 1]:
418 with pytest.raises(grpc.RpcError) as e:
419 api.SetModerationAutoApproveDeadline(
420 admin_pb2.SetModerationAutoApproveDeadlineReq(
421 user=normal_user.username, auto_approve_seconds=wrappers_pb2.Int64Value(value=seconds)
422 )
423 )
424 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
426 assert [(a.action_type, a.note) for a in res.admin_actions] == [
427 ("set_moderation_auto_approve_deadline", "Changed from default to never"),
428 ("set_moderation_auto_approve_deadline", "Changed from never to 172800 seconds"),
429 ("set_moderation_auto_approve_deadline", "Changed from 172800 seconds to default"),
430 ]
433def test_ShadowUser_blank_note(db):
434 super_user, super_token = generate_user(is_superuser=True)
435 normal_user, _ = generate_user()
437 with real_admin_session(super_token) as api:
438 with pytest.raises(grpc.RpcError) as e:
439 api.ShadowUser(admin_pb2.ShadowUserReq(user=normal_user.username, admin_note=" \t "))
440 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
443def test_AddAdminNote(db):
444 super_user, super_token = generate_user(is_superuser=True)
445 normal_user, _ = generate_user()
446 admin_note1 = "User reported strange behavior"
447 admin_note2 = "Insert private information here"
449 with real_admin_session(super_token) as api:
450 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note1))
451 assert res.user_id == normal_user.id
452 assert res.username == normal_user.username
453 assert res.email == normal_user.email
454 assert res.gender == normal_user.gender
455 assert parse_date(res.birthdate) == normal_user.birthdate
456 assert not res.banned
457 assert not res.deleted
458 assert len(res.admin_actions) == 1
459 assert res.admin_actions[0].action_type == "note"
460 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
461 assert res.admin_actions[0].note == admin_note1
463 with real_admin_session(super_token) as api:
464 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note2))
465 assert len(res.admin_actions) == 2
466 assert res.admin_actions[0].note == admin_note1
467 assert res.admin_actions[1].note == admin_note2
470def test_AddAdminNote_blank(db):
471 super_user, super_token = generate_user(is_superuser=True)
472 normal_user, _ = generate_user()
473 empty_admin_note = " \t \n "
475 with real_admin_session(super_token) as api:
476 with pytest.raises(grpc.RpcError) as e:
477 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=empty_admin_note))
478 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
479 assert e.value.details() == "Provide exactly one of admin_note or data."
482def test_AddAdminNote_data(db):
483 super_user, super_token = generate_user(is_superuser=True)
484 normal_user, _ = generate_user()
485 payload = '{"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}'
487 with real_admin_session(super_token) as api:
488 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data=payload))
489 assert len(res.admin_actions) == 1
490 assert res.admin_actions[0].action_type == "note"
491 assert res.admin_actions[0].note == ""
492 assert json.loads(res.admin_actions[0].data) == {"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}
495def test_AddAdminNote_both_note_and_data(db):
496 super_user, super_token = generate_user(is_superuser=True)
497 normal_user, _ = generate_user()
499 with real_admin_session(super_token) as api:
500 with pytest.raises(grpc.RpcError) as e:
501 api.AddAdminNote(
502 admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="note text", data='{"x": 1}')
503 )
504 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
505 assert e.value.details() == "Provide exactly one of admin_note or data."
508def test_AddAdminNote_neither(db):
509 super_user, super_token = generate_user(is_superuser=True)
510 normal_user, _ = generate_user()
512 with real_admin_session(super_token) as api:
513 with pytest.raises(grpc.RpcError) as e:
514 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username))
515 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
516 assert e.value.details() == "Provide exactly one of admin_note or data."
519def test_AddAdminNote_invalid_json(db):
520 super_user, super_token = generate_user(is_superuser=True)
521 normal_user, _ = generate_user()
523 with real_admin_session(super_token) as api:
524 with pytest.raises(grpc.RpcError) as e:
525 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data="{not valid json"))
526 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
527 assert e.value.details() == "The admin note data must be valid JSON."
530def test_admin_content_reports(db):
531 super_user, super_token = generate_user(is_superuser=True)
532 normal_user, token = generate_user()
533 bad_user1, _ = generate_user()
534 bad_user2, _ = generate_user()
536 with reporting_session(token) as api:
537 api.Report(
538 reporting_pb2.ReportReq(
539 reason="spam",
540 description="r1",
541 content_ref="comment/123",
542 author_user=bad_user1.username,
543 user_agent="n/a",
544 page="https://couchers.org/comment/123",
545 )
546 )
547 api.Report(
548 reporting_pb2.ReportReq(
549 reason="spam",
550 description="r2",
551 content_ref="comment/124",
552 author_user=bad_user2.username,
553 user_agent="n/a",
554 page="https://couchers.org/comment/124",
555 )
556 )
557 api.Report(
558 reporting_pb2.ReportReq(
559 reason="something else",
560 description="r3",
561 content_ref="page/321",
562 author_user=bad_user1.username,
563 user_agent="n/a",
564 page="https://couchers.org/page/321",
565 )
566 )
568 with session_scope() as session:
569 id_by_description: dict[str, int] = dict(
570 session.execute(select(ContentReport.description, ContentReport.id)).all() # type: ignore[arg-type]
571 )
573 with real_admin_session(super_token) as api:
574 with pytest.raises(grpc.RpcError) as e:
575 api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=-1))
576 assert e.value.code() == grpc.StatusCode.NOT_FOUND
577 assert e.value.details() == "Content report not found."
579 res = api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=id_by_description["r2"]))
580 rep = res.content_report
581 assert rep.content_report_id == id_by_description["r2"]
582 assert rep.reporting_user_id == normal_user.id
583 assert rep.author_user_id == bad_user2.id
584 assert rep.reason == "spam"
585 assert rep.description == "r2"
586 assert rep.content_ref == "comment/124"
587 assert rep.user_agent == "n/a"
588 assert rep.page == "https://couchers.org/comment/124"
590 res = api.GetContentReportsForAuthor(admin_pb2.GetContentReportsForAuthorReq(user=bad_user1.username))
591 assert res.content_reports[0].content_report_id == id_by_description["r3"]
592 assert res.content_reports[1].content_report_id == id_by_description["r1"]
595def test_DeleteUser(db):
596 super_user, super_token = generate_user(is_superuser=True)
597 normal_user, normal_token = generate_user()
599 with real_admin_session(super_token) as api:
600 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
601 assert res.user_id == normal_user.id
602 assert res.username == normal_user.username
603 assert res.email == normal_user.email
604 assert res.gender == normal_user.gender
605 assert parse_date(res.birthdate) == normal_user.birthdate
606 assert not res.banned
607 assert res.deleted
609 with real_admin_session(super_token) as api:
610 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
611 assert res.user_id == normal_user.id
612 assert res.username == normal_user.username
613 assert res.email == normal_user.email
614 assert res.gender == normal_user.gender
615 assert parse_date(res.birthdate) == normal_user.birthdate
616 assert not res.banned
617 assert not res.deleted
620def test_RecoverDeletedUser_after_user_initiated_deletion(db, push_collector: PushCollector):
621 """
622 When a user deletes their account through the normal flow (ConfirmDeleteAccount),
623 undelete_token and undelete_until are set. The admin RecoverDeletedUser must clear
624 these fields to satisfy the undelete_nullity database constraint.
625 """
626 super_user, super_token = generate_user(is_superuser=True)
627 normal_user, normal_token = generate_user()
628 user_id = normal_user.id
630 # User initiates account deletion
631 with account_session(normal_token) as account:
632 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
634 # Get the deletion confirmation token
635 with session_scope() as session:
636 deletion_token = session.execute(select(AccountDeletionToken)).scalar_one().token
638 # User confirms account deletion (this sets undelete_token and undelete_until)
639 with auth_api_session() as (auth_api, metadata_interceptor):
640 auth_api.ConfirmDeleteAccount(auth_pb2.ConfirmDeleteAccountReq(token=deletion_token))
642 # Verify the user is deleted and has undelete fields set
643 with session_scope() as session:
644 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
645 assert user.deleted_at is not None
646 assert user.undelete_token is not None
647 assert user.undelete_until is not None
649 # Admin recovers the user
650 with real_admin_session(super_token) as api:
651 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
652 assert res.user_id == user_id
653 assert not res.deleted
655 # Verify undelete fields are cleared
656 with session_scope() as session:
657 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
658 assert user.deleted_at is None
659 assert user.undelete_token is None
660 assert user.undelete_until is None
663def test_RecoverDeletedUser_email_reused(db):
664 super_user, super_token = generate_user(is_superuser=True)
665 normal_user, _ = generate_user()
667 with session_scope() as session:
668 session.execute(update(User).where(User.id == normal_user.id).values(deleted_at=func.now()))
670 new_user, _ = generate_user(email=normal_user.email)
672 with real_admin_session(super_token) as api:
673 with pytest.raises(grpc.RpcError) as e:
674 api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.email))
675 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
676 assert (
677 e.value.details()
678 == "That email address is used by more than one user, look them up by username or ID instead."
679 )
681 with pytest.raises(grpc.RpcError) as e:
682 api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
683 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
686def test_SearchUsers_exact_email(db):
687 super_user, super_token = generate_user(is_superuser=True)
688 deleted_user, _ = generate_user(email="first_last@couchers.org.invalid", delete_user=True)
689 live_user, _ = generate_user(email="first_last@couchers.org.invalid")
690 wildcard_match, _ = generate_user(email="firstxlast@couchers.org.invalid")
692 with real_admin_session(super_token) as api:
693 res = api.SearchUsers(admin_pb2.SearchUsersReq(email="first_last@couchers.org.invalid"))
694 assert {u.user_id for u in res.users} == {deleted_user.id, live_user.id, wildcard_match.id}
696 res = api.SearchUsers(admin_pb2.SearchUsersReq(exact_email="First_Last@couchers.org.invalid"))
697 assert {u.user_id for u in res.users} == {deleted_user.id, live_user.id}
700def test_CreateApiKey(db, email_collector: EmailCollector, push_collector: PushCollector):
701 with session_scope() as session:
702 super_user, super_token = generate_user(is_superuser=True)
703 normal_user, normal_token = generate_user()
705 assert (
706 session.execute(
707 select(func.count())
708 .select_from(UserSession)
709 .where(UserSession.is_api_key == True)
710 .where(UserSession.user_id == normal_user.id)
711 ).scalar_one()
712 == 0
713 )
715 with real_admin_session(super_token) as api:
716 res = api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=normal_user.username))
718 email = email_collector.pop_for_recipient(normal_user.email, last=True)
719 assert email.subject == "[TEST] Your API key for Couchers.org"
721 with session_scope() as session:
722 token = session.execute(
723 select(UserSession.token)
724 .where(UserSession.is_valid)
725 .where(UserSession.is_api_key == True)
726 .where(UserSession.user_id == normal_user.id)
727 ).scalar_one()
729 assert token in email.plain
730 assert token in email.html
732 assert email.recipient == normal_user.email
733 assert "api key" in email.subject.lower()
734 unique_string = "We've issued you with the following API key:"
735 assert unique_string in email.plain
736 assert unique_string in email.html
737 assert "support@couchers.org" in email.plain
738 assert "support@couchers.org" in email.html
740 push = push_collector.pop_for_user(normal_user.id, last=True)
741 assert push.content.title == "API key created"
742 assert push.content.body == "Details were sent to you via email."
745def test_GetChats(db):
746 super_user, super_token = generate_user(is_superuser=True)
747 normal_user, normal_token = generate_user()
749 with real_admin_session(super_token) as api:
750 res = api.GetChats(admin_pb2.GetChatsReq(user=normal_user.username))
751 assert res.user.user_id == normal_user.id
752 assert res.user.username == normal_user.username
753 assert res.user.name == normal_user.name
754 # New user should have no chats
755 assert len(res.host_requests) == 0
756 assert len(res.group_chats) == 0
759def test_badges(db, email_collector: EmailCollector, push_collector: PushCollector):
760 super_user, super_token = generate_user(is_superuser=True)
761 normal_user, normal_token = generate_user()
763 with real_admin_session(super_token) as api:
764 # can add a badge
765 assert "swagster" not in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
766 res = api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
767 assert "swagster" in res.badges
769 # badge emails are disabled by default
770 assert email_collector.count_for_recipient(normal_user.email) == 0
772 push = push_collector.pop_for_user(normal_user.id, last=True)
773 assert push.content.title == "New profile badge: Swagster"
774 assert push.content.body == "The Swagster badge was added to your profile."
776 # can't add/edit special tags
777 with pytest.raises(grpc.RpcError) as e:
778 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="founder"))
779 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
780 assert e.value.details() == "Admins cannot edit that badge."
782 # double add badge
783 with pytest.raises(grpc.RpcError) as e:
784 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
785 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
786 assert e.value.details() == "The user already has that badge."
788 # can remove badge
789 assert "swagster" in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
790 res = api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
791 assert "swagster" not in res.badges
793 # badge emails are disabled by default
794 assert email_collector.count_for_recipient(normal_user.email) == 0
796 push = push_collector.pop_for_user(normal_user.id, last=True)
797 assert push.content.title == "Profile badge removed"
798 assert push.content.body == "The Swagster badge was removed from your profile."
800 # not found on user
801 with pytest.raises(grpc.RpcError) as e:
802 api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
803 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
804 assert e.value.details() == "The user does not have that badge."
806 # not found in general
807 with pytest.raises(grpc.RpcError) as e:
808 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="nonexistentbadge"))
809 assert e.value.code() == grpc.StatusCode.NOT_FOUND
810 assert e.value.details() == "Badge not found."
813def test_DeleteEvent(db):
814 super_user, super_token = generate_user(is_superuser=True)
815 normal_user, normal_token = generate_user()
817 with session_scope() as session:
818 create_community(session, 0, 2, "Community", [normal_user], [], None)
820 res = events_api.create_event(token=normal_token)
821 event_id = res.event_id
822 assert not res.is_deleted
824 with session_scope() as session:
825 with real_admin_session(super_token) as api:
826 api.DeleteEvent(
827 admin_pb2.DeleteEventReq(
828 event_id=event_id,
829 )
830 )
831 occurrence = session.get_one(EventOccurrence, ident=event_id)
832 assert occurrence.is_deleted
835def test_ListUserIds(db):
836 super_user, super_token = generate_user(is_superuser=True)
837 normal_user, normal_token = generate_user()
839 with real_admin_session(super_token) as api:
840 res = api.ListUserIds(
841 admin_pb2.ListUserIdsReq(
842 start_time=Timestamp_from_datetime(datetime(2000, 1, 1, tzinfo=UTC)),
843 end_time=Timestamp_from_datetime(now()),
844 )
845 )
846 assert len(res.user_ids) == 2
847 assert sorted(res.user_ids) == sorted([super_user.id, normal_user.id])
849 with real_admin_session(super_token) as api:
850 res = api.ListUserIds(
851 admin_pb2.ListUserIdsReq(start_time=Timestamp_from_datetime(now()), end_time=Timestamp_from_datetime(now()))
852 )
853 assert res.user_ids == []
856def test_EditReferenceText(db):
857 super_user, super_token = generate_user(is_superuser=True)
858 test_new_text = "New Text"
860 user1, user1_token = generate_user()
861 user2, user2_token = generate_user()
862 make_friends(user1, user2)
864 with session_scope() as session:
865 with references_session(user1_token) as api:
866 reference = api.WriteFriendReference(
867 references_pb2.WriteFriendReferenceReq(
868 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
869 )
870 )
872 with real_admin_session(super_token) as admin_api:
873 admin_api.EditReferenceText(
874 admin_pb2.EditReferenceTextReq(reference_id=reference.reference_id, new_text=test_new_text)
875 )
877 session.expire_all()
879 modified_reference = session.execute(
880 select(Reference).where(Reference.id == reference.reference_id)
881 ).scalar_one()
882 assert modified_reference.text == test_new_text
885def test_DeleteReference_deprecated(db):
886 """DeleteReference is deprecated; admins should hide via UMS instead."""
887 super_user, super_token = generate_user(is_superuser=True)
889 user1, user1_token = generate_user()
890 user2, user2_token = generate_user()
891 make_friends(user1, user2)
893 with references_session(user1_token) as api:
894 reference = api.WriteFriendReference(
895 references_pb2.WriteFriendReferenceReq(
896 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
897 )
898 )
900 with real_admin_session(super_token) as admin_api:
901 with pytest.raises(grpc.RpcError) as e:
902 admin_api.DeleteReference(admin_pb2.DeleteReferenceReq(reference_id=reference.reference_id))
903 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
906def test_GetUserReferences(db):
907 super_user, super_token = generate_user(is_superuser=True)
909 user1, user1_token = generate_user()
910 user2, user2_token = generate_user()
911 user3, user3_token = generate_user()
912 make_friends(user1, user2)
913 make_friends(user1, user3)
914 make_friends(user2, user3)
916 # user1 writes reference about user2
917 with references_session(user1_token) as api:
918 ref1 = api.WriteFriendReference(
919 references_pb2.WriteFriendReferenceReq(
920 to_user_id=user2.id,
921 text="Reference from user1 to user2",
922 private_text="",
923 was_appropriate=True,
924 rating=1,
925 )
926 )
928 # user2 writes reference about user1
929 with references_session(user2_token) as api:
930 ref2 = api.WriteFriendReference(
931 references_pb2.WriteFriendReferenceReq(
932 to_user_id=user1.id,
933 text="Reference from user2 to user1",
934 private_text="Private note",
935 was_appropriate=True,
936 rating=0.8,
937 )
938 )
940 # user3 writes reference about user1
941 with references_session(user3_token) as api:
942 ref3 = api.WriteFriendReference(
943 references_pb2.WriteFriendReferenceReq(
944 to_user_id=user1.id,
945 text="Reference from user3 to user1",
946 private_text="",
947 was_appropriate=False,
948 rating=0.5,
949 )
950 )
952 # Test GetUserReferences for user1 (admin view shows everything regardless of UMS state).
953 with real_admin_session(super_token) as admin_api:
954 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
956 # user1 wrote 1 reference
957 assert len(res.references_from) == 1
958 assert res.references_from[0].reference_id == ref1.reference_id
959 assert res.references_from[0].from_user_id == user1.id
960 assert res.references_from[0].to_user_id == user2.id
961 assert res.references_from[0].from_user.user_id == user1.id
962 assert res.references_from[0].from_user.username == user1.username
963 assert res.references_from[0].to_user.user_id == user2.id
964 assert res.references_from[0].to_user.username == user2.username
965 assert res.references_from[0].reference_type == "friend"
966 assert res.references_from[0].text == "Reference from user1 to user2"
967 # freshly written content starts out shadowed
968 assert res.references_from[0].moderation_visibility == "shadowed"
969 assert res.references_from[0].host_request_id == 0
970 assert res.references_from[0].hosting_city == ""
971 assert res.references_from[0].from_date == ""
972 assert res.references_from[0].to_date == ""
973 assert res.references_from[0].status == ""
975 # user1 received 2 references
976 assert len(res.references_to) == 2
977 # Ordered by id descending, so ref3 comes first
978 assert res.references_to[0].reference_id == ref3.reference_id
979 assert res.references_to[0].was_appropriate is False
981 assert res.references_to[1].reference_id == ref2.reference_id
982 assert res.references_to[1].private_text == "Private note"
983 assert res.references_to[1].rating == 0.8
986def test_GetUserReferences_host_request(db):
987 super_user, super_token = generate_user(is_superuser=True)
989 user1, _ = generate_user()
990 user2, _ = generate_user()
992 with session_scope() as session:
993 surfed_ref_id, host_request_id = create_host_reference(session, user1.id, user2.id, timedelta(days=2))
994 hosted_ref_id, _ = create_host_reference(
995 session, user2.id, user1.id, timedelta(days=2), host_request_id=host_request_id
996 )
997 host_request = session.execute(
998 select(HostRequest).where(HostRequest.conversation_id == host_request_id)
999 ).scalar_one()
1000 from_date = date_to_api(host_request.from_date)
1001 to_date = date_to_api(host_request.to_date)
1003 with real_admin_session(super_token) as admin_api:
1004 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
1006 assert len(res.references_from) == 1
1007 surfed_ref = res.references_from[0]
1008 assert surfed_ref.reference_id == surfed_ref_id
1009 assert surfed_ref.reference_type == "surfed"
1010 assert surfed_ref.from_user.user_id == user1.id
1011 assert surfed_ref.from_user.username == user1.username
1012 assert surfed_ref.to_user.user_id == user2.id
1013 assert surfed_ref.to_user.username == user2.username
1014 assert surfed_ref.moderation_visibility == "visible"
1015 assert surfed_ref.host_request_id == host_request_id
1016 assert surfed_ref.hosting_city == "Test City"
1017 assert surfed_ref.from_date == from_date
1018 assert surfed_ref.to_date == to_date
1019 assert surfed_ref.status == "confirmed"
1021 assert len(res.references_to) == 1
1022 hosted_ref = res.references_to[0]
1023 assert hosted_ref.reference_id == hosted_ref_id
1024 assert hosted_ref.reference_type == "hosted"
1025 assert hosted_ref.from_user.user_id == user2.id
1026 assert hosted_ref.to_user.user_id == user1.id
1027 assert hosted_ref.host_request_id == host_request_id
1028 assert hosted_ref.hosting_city == "Test City"
1029 assert hosted_ref.from_date == from_date
1030 assert hosted_ref.to_date == to_date
1031 assert hosted_ref.status == "confirmed"
1034def test_GetUserReferences_not_found(db):
1035 super_user, super_token = generate_user(is_superuser=True)
1037 with real_admin_session(super_token) as admin_api:
1038 with pytest.raises(grpc.RpcError) as e:
1039 admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user="nonexistent"))
1040 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1043def test_GetFriendRequests(db):
1044 super_user, super_token = generate_user(is_superuser=True)
1046 user1, _ = generate_user()
1047 user2, _ = generate_user()
1048 user3, _ = generate_user()
1049 user4, _ = generate_user()
1051 # Create a mix of friend requests directly so we control the state
1052 def _add_friend_request(from_user_id, to_user_id, status, visibility, time_responded=None):
1053 with session_scope() as session:
1054 mod_state = ModerationState(
1055 object_type=ModerationObjectType.friend_request,
1056 object_id=0,
1057 visibility=visibility,
1058 )
1059 session.add(mod_state)
1060 session.flush()
1061 rel = FriendRelationship(
1062 from_user_id=from_user_id,
1063 to_user_id=to_user_id,
1064 status=status,
1065 moderation_state_id=mod_state.id,
1066 time_responded=time_responded,
1067 )
1068 session.add(rel)
1069 session.flush()
1070 mod_state.object_id = rel.id
1072 # user1 -> user2: pending, shadowed
1073 _add_friend_request(user1.id, user2.id, FriendStatus.pending, ModerationVisibility.shadowed)
1074 # user1 -> user3: accepted, visible
1075 _add_friend_request(user1.id, user3.id, FriendStatus.accepted, ModerationVisibility.visible, time_responded=now())
1076 # user4 -> user1: rejected, visible
1077 _add_friend_request(user4.id, user1.id, FriendStatus.rejected, ModerationVisibility.visible, time_responded=now())
1079 with real_admin_session(super_token) as admin_api:
1080 res = admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user=user1.username))
1082 # user1 sent two: to user2 (pending) and to user3 (accepted), ordered by id desc
1083 assert len(res.sent) == 2
1084 assert res.sent[0].from_user.user_id == user1.id
1085 assert res.sent[0].to_user.user_id == user3.id
1086 assert res.sent[0].status == "accepted"
1087 assert res.sent[0].HasField("time_responded")
1088 assert res.sent[0].moderation_visibility == "visible"
1090 assert res.sent[1].from_user.user_id == user1.id
1091 assert res.sent[1].to_user.user_id == user2.id
1092 assert res.sent[1].status == "pending"
1093 assert not res.sent[1].HasField("time_responded")
1094 assert res.sent[1].moderation_visibility == "shadowed"
1096 # user1 received one: from user4 (rejected)
1097 assert len(res.received) == 1
1098 assert res.received[0].from_user.user_id == user4.id
1099 assert res.received[0].to_user.user_id == user1.id
1100 assert res.received[0].status == "rejected"
1103def test_GetFriendRequests_not_found(db):
1104 super_user, super_token = generate_user(is_superuser=True)
1106 with real_admin_session(super_token) as admin_api:
1107 with pytest.raises(grpc.RpcError) as e:
1108 admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user="nonexistent"))
1109 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1112def test_GetUserBlocks(db):
1113 super_user, super_token = generate_user(is_superuser=True)
1115 user1, token1 = generate_user()
1116 user2, _ = generate_user()
1117 user3, token3 = generate_user()
1118 user4, token4 = generate_user()
1120 with blocking_session(token1) as user_blocks:
1121 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user2.username))
1123 with blocking_session(token3) as user_blocks:
1124 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user1.username))
1126 with blocking_session(token4) as user_blocks:
1127 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user1.username))
1129 with real_admin_session(super_token) as admin_api:
1130 res = admin_api.GetUserBlocks(admin_pb2.GetUserBlocksReq(user=user1.username))
1132 assert len(res.blocked_users) == 1
1133 assert res.blocked_users[0].user.user_id == user2.id
1134 assert res.blocked_users[0].user.username == user2.username
1135 assert res.blocked_users[0].HasField("time_blocked")
1137 # most recently blocked first
1138 assert [block.user.user_id for block in res.blocking_users] == [user4.id, user3.id]
1140 with real_admin_session(super_token) as admin_api:
1141 res = admin_api.GetUserBlocks(admin_pb2.GetUserBlocksReq(user=user2.username))
1143 assert len(res.blocked_users) == 0
1144 assert [block.user.user_id for block in res.blocking_users] == [user1.id]
1147def test_GetUserBlocks_not_found(db):
1148 super_user, super_token = generate_user(is_superuser=True)
1150 with real_admin_session(super_token) as admin_api:
1151 with pytest.raises(grpc.RpcError) as e:
1152 admin_api.GetUserBlocks(admin_pb2.GetUserBlocksReq(user="nonexistent"))
1153 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1156def test_AddUsersToModerationUserList(db):
1157 super_user, super_token = generate_user(is_superuser=True)
1158 user1, _ = generate_user()
1159 user2, _ = generate_user()
1160 user3, _ = generate_user()
1161 user4, _ = generate_user()
1162 user5, _ = generate_user()
1163 moderation_list_id = add_users_to_new_moderation_list([user1])
1165 with session_scope() as session:
1166 with real_admin_session(super_token) as api:
1167 # Test adding users to a non-existent moderation list (should raise an error)
1168 with pytest.raises(grpc.RpcError) as e:
1169 api.AddUsersToModerationUserList(
1170 admin_pb2.AddUsersToModerationUserListReq(users=[user2.username], moderation_list_id=999),
1171 )
1172 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1173 assert "Moderation user list not found." == e.value.details()
1175 # Test with non-existent user (should raise an error)
1176 with pytest.raises(grpc.RpcError) as e:
1177 api.AddUsersToModerationUserList(
1178 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, "nonexistent"]),
1179 )
1180 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1181 assert "Couldn't find that user." == e.value.details()
1183 # Test successful creation of new moderation list (no moderation_list_id provided)
1184 res = api.AddUsersToModerationUserList(
1185 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, user2.username, user3.username]),
1186 )
1187 assert res.moderation_list_id > 0
1188 with session_scope() as session:
1189 moderation_user_list = session.get(ModerationUserList, res.moderation_list_id)
1190 assert moderation_user_list is not None
1191 assert len(moderation_user_list.users) == 3
1192 assert {user1.id, user2.id, user3.id}.issubset({user.id for user in moderation_user_list.users})
1194 # Test list endpoint returns same moderation list with same members not repeated
1195 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1196 assert len(listRes.moderation_lists) == 1
1197 assert listRes.moderation_lists[0].moderation_list_id == res.moderation_list_id
1198 assert len(listRes.moderation_lists[0].members) == 3
1199 assert {user1.id, user2.id, user3.id}.issubset({m.user_id for m in listRes.moderation_lists[0].members})
1201 # Test user can be in multiple moderation lists
1202 listRes3 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1203 assert len(listRes3.moderation_lists) == 2
1205 # Test adding users to an existing moderation list
1206 res2 = api.AddUsersToModerationUserList(
1207 admin_pb2.AddUsersToModerationUserListReq(
1208 users=[user4.username, user5.username], moderation_list_id=moderation_list_id
1209 ),
1210 )
1211 assert res2.moderation_list_id == moderation_list_id
1212 with session_scope() as session:
1213 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1214 assert len(moderation_user_list.users) == 3
1215 assert {user1.id, user4.id, user5.id}.issubset({user.id for user in moderation_user_list.users})
1217 # Test list user moderation lists endpoint returns the right moderation list
1218 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user5.username))
1219 assert len(listRes2.moderation_lists) == 1
1220 assert listRes2.moderation_lists[0].moderation_list_id == moderation_list_id
1221 assert len(listRes2.moderation_lists[0].members) == 3
1222 assert {user1.id, user4.id, user5.id}.issubset({m.user_id for m in listRes2.moderation_lists[0].members})
1225def test_RemoveUserFromModerationUserList(db):
1226 super_user, super_token = generate_user(is_superuser=True)
1227 user1, _ = generate_user()
1228 user2, _ = generate_user()
1229 user3, _ = generate_user()
1230 moderation_list_id = add_users_to_new_moderation_list([user1, user2])
1232 with real_admin_session(super_token) as api:
1233 # Test with non-existent user (should raise error)
1234 with pytest.raises(grpc.RpcError) as e:
1235 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user="nonexistent"))
1236 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1237 assert "Couldn't find that user." == e.value.details()
1239 # Test without providing moderation list id (should raise error)
1240 with pytest.raises(grpc.RpcError) as e:
1241 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username))
1242 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1243 assert "Missing moderation user list id." == e.value.details()
1245 # Test removing user that's not in the provided moderation list (should raise error)
1246 with pytest.raises(grpc.RpcError) as e:
1247 api.RemoveUserFromModerationUserList(
1248 admin_pb2.RemoveUserFromModerationUserListReq(
1249 user=user3.username, moderation_list_id=moderation_list_id
1250 )
1251 )
1252 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1253 assert "User is not in the moderation user list." == e.value.details()
1255 # Test successful removal
1256 api.RemoveUserFromModerationUserList(
1257 admin_pb2.RemoveUserFromModerationUserListReq(user=user1.username, moderation_list_id=moderation_list_id)
1258 )
1259 with session_scope() as session:
1260 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1261 assert user1.id not in {user.id for user in moderation_user_list.users}
1262 assert user2.id in {user.id for user in moderation_user_list.users}
1264 # Test list user moderation lists endpoint returns right number of moderation lists
1265 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1266 assert len(listRes.moderation_lists) == 0
1267 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1268 assert len(listRes2.moderation_lists) == 1
1270 # Test removing all users from moderation list should also delete the moderation list
1271 api.RemoveUserFromModerationUserList(
1272 admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username, moderation_list_id=moderation_list_id)
1273 )
1274 with session_scope() as session:
1275 assert session.get(ModerationUserList, moderation_list_id) is None
1278def test_admin_delete_account_url(db, email_collector: EmailCollector, push_collector: PushCollector):
1279 super_user, super_token = generate_user(is_superuser=True)
1281 user, token = generate_user()
1282 user_id = user.id
1284 with real_admin_session(super_token) as admin_api:
1285 url = admin_api.CreateAccountDeletionLink(
1286 admin_pb2.CreateAccountDeletionLinkReq(user=user.username)
1287 ).account_deletion_confirm_url
1289 assert push_collector.count_for_user(user_id) == 0
1291 with session_scope() as session:
1292 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
1293 token = token_o.token
1294 assert token_o.user.id == user_id
1295 assert url == f"http://localhost:3000/delete-account?token={token}"
1297 with auth_api_session() as (auth_api, metadata_interceptor):
1298 auth_api.ConfirmDeleteAccount(
1299 auth_pb2.ConfirmDeleteAccountReq(
1300 token=token,
1301 )
1302 )
1304 push = push_collector.pop_for_user(user_id, last=True)
1305 assert push.content.title == "Account deleted"
1306 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
1307 email_collector.pop_for_recipient(user.email, last=True)
1310def test_AccessStats(db):
1311 super_user, super_token = generate_user(is_superuser=True)
1312 normal_user, normal_token = generate_user()
1314 # Insert UserActivity rows: a couple inside the default 90-day window, one well
1315 # outside it, and one with NULL ip_address / user_agent. The INET column is
1316 # returned by psycopg3 as an IPv4Address/IPv6Address object, which used to
1317 # crash the proto string assignment.
1318 in_window_1 = now() - timedelta(days=1)
1319 in_window_2 = now() - timedelta(days=10)
1320 out_of_window = now() - timedelta(days=200)
1321 with session_scope() as session:
1322 session.add(
1323 UserActivity(
1324 user_id=normal_user.id, period=in_window_1, ip_address="1.2.3.4", user_agent="ua-a", api_calls=5
1325 )
1326 )
1327 session.add(
1328 UserActivity(
1329 user_id=normal_user.id, period=in_window_2, ip_address="2001:db8::1", user_agent="ua-b", api_calls=3
1330 )
1331 )
1332 session.add(
1333 UserActivity(
1334 user_id=normal_user.id, period=out_of_window, ip_address="9.9.9.9", user_agent="ua-old", api_calls=99
1335 )
1336 )
1337 session.add(UserActivity(user_id=normal_user.id, period=in_window_1, api_calls=1))
1339 with real_admin_session(super_token) as api:
1340 res = api.AccessStats(admin_pb2.AccessStatsReq(user=normal_user.username))
1342 by_ip = {s.ip_address: s for s in res.stats}
1343 assert "1.2.3.4" in by_ip
1344 assert by_ip["1.2.3.4"].api_call_count == 5
1345 assert by_ip["1.2.3.4"].user_agent == "ua-a"
1346 assert "2001:db8::1" in by_ip
1347 assert by_ip["2001:db8::1"].api_call_count == 3
1348 # NULL ip_address row produces an empty-string ip_address in the proto
1349 assert "" in by_ip
1350 assert by_ip[""].api_call_count == 1
1351 # out-of-window row is excluded by the 90-day default
1352 assert "9.9.9.9" not in by_ip
1354 # explicit end_time should bound the upper end of the window (regression: was >=)
1355 with real_admin_session(super_token) as api:
1356 res = api.AccessStats(
1357 admin_pb2.AccessStatsReq(
1358 user=normal_user.username,
1359 start_time=Timestamp_from_datetime(now() - timedelta(days=5)),
1360 end_time=Timestamp_from_datetime(now()),
1361 )
1362 )
1363 ips = {s.ip_address for s in res.stats}
1364 assert ips == {"1.2.3.4", ""}
1367def test_SetLastDonated(db):
1368 super_user, super_token = generate_user(is_superuser=True)
1369 normal_user, normal_token = generate_user(last_donated=None)
1371 with real_admin_session(super_token) as api:
1372 # user starts with no last_donated
1373 with session_scope() as session:
1374 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1375 assert user.last_donated is None
1377 # can set last_donated
1378 donation_time = now() - timedelta(days=30)
1379 res = api.SetLastDonated(
1380 admin_pb2.SetLastDonatedReq(
1381 user=normal_user.username,
1382 last_donated=Timestamp_from_datetime(donation_time),
1383 )
1384 )
1386 with session_scope() as session:
1387 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1388 assert user.last_donated is not None
1389 # check timestamp is close (within a second)
1390 assert abs((user.last_donated - donation_time).total_seconds()) < 1
1392 # can clear last_donated by not setting the field
1393 res = api.SetLastDonated(admin_pb2.SetLastDonatedReq(user=normal_user.username))
1395 with session_scope() as session:
1396 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1397 assert user.last_donated is None
1399 # user not found
1400 with pytest.raises(grpc.RpcError) as e:
1401 api.SetLastDonated(admin_pb2.SetLastDonatedReq(user="nonexistent"))
1402 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1403 assert e.value.details() == "Couldn't find that user."
1406def test_admin_actions_level(db):
1407 super_user, super_token = generate_user(is_superuser=True)
1408 normal_user, _ = generate_user()
1410 with real_admin_session(super_token) as api:
1411 # Default level is NORMAL
1412 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="normal note"))
1413 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
1415 # Explicitly set to DEBUG
1416 res = api.AddAdminNote(
1417 admin_pb2.AddAdminNoteReq(
1418 user=normal_user.username,
1419 admin_note="debug note",
1420 level=admin_pb2.ADMIN_ACTION_LEVEL_DEBUG,
1421 )
1422 )
1423 assert len(res.admin_actions) == 2
1424 assert res.admin_actions[1].level == admin_pb2.ADMIN_ACTION_LEVEL_DEBUG
1426 # Explicitly set to HIGH
1427 res = api.AddAdminNote(
1428 admin_pb2.AddAdminNoteReq(
1429 user=normal_user.username,
1430 admin_note="high note",
1431 level=admin_pb2.ADMIN_ACTION_LEVEL_HIGH,
1432 )
1433 )
1434 assert len(res.admin_actions) == 3
1435 assert res.admin_actions[2].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
1437 # Explicitly set to TRACE
1438 res = api.AddAdminNote(
1439 admin_pb2.AddAdminNoteReq(
1440 user=normal_user.username,
1441 admin_note="trace note",
1442 level=admin_pb2.ADMIN_ACTION_LEVEL_TRACE,
1443 )
1444 )
1445 assert len(res.admin_actions) == 4
1446 assert res.admin_actions[3].level == admin_pb2.ADMIN_ACTION_LEVEL_TRACE
1449def test_admin_actions_on_mutations(db, push_collector: PushCollector):
1450 super_user, super_token = generate_user(is_superuser=True)
1451 normal_user, _ = generate_user()
1453 original_gender = normal_user.gender
1454 original_birthdate = normal_user.birthdate
1456 with real_admin_session(super_token) as api:
1457 # ChangeUserGender
1458 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
1459 assert any(
1460 a.action_type == "change_gender" and a.note == f"Changed from '{original_gender}' to 'Machine'"
1461 for a in res.admin_actions
1462 )
1464 # ChangeUserBirthdate
1465 res = api.ChangeUserBirthdate(
1466 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-01-01")
1467 )
1468 assert any(
1469 a.action_type == "change_birthdate" and a.note == f"Changed from {original_birthdate} to 1990-01-01"
1470 for a in res.admin_actions
1471 )
1473 # SetPassportSexGenderException
1474 res = api.SetPassportSexGenderException(
1475 admin_pb2.SetPassportSexGenderExceptionReq(user=normal_user.username, passport_sex_gender_exception=True)
1476 )
1477 assert any(
1478 a.action_type == "set_passport_sex_gender_exception" and a.note == "Changed from False to True"
1479 for a in res.admin_actions
1480 )
1482 # SendModNote with notify
1483 res = api.SendModNote(
1484 admin_pb2.SendModNoteReq(
1485 user=normal_user.username,
1486 content="Please update your profile",
1487 internal_id="test1",
1488 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY,
1489 )
1490 )
1491 assert any(
1492 a.action_type == "send_mod_note" and a.note == "Notify user: Yes\n\nPlease update your profile"
1493 for a in res.admin_actions
1494 )
1496 # SendModNote without notifying
1497 res = api.SendModNote(
1498 admin_pb2.SendModNoteReq(
1499 user=normal_user.username,
1500 content="Silent note",
1501 internal_id="test2",
1502 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NONE,
1503 )
1504 )
1505 assert any(
1506 a.action_type == "send_mod_note" and a.note == "Notify user: No\n\nSilent note" for a in res.admin_actions
1507 )
1509 # SendModNote including the note text in the email
1510 res = api.SendModNote(
1511 admin_pb2.SendModNoteReq(
1512 user=normal_user.username,
1513 content="Emailed note",
1514 internal_id="test3",
1515 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT,
1516 )
1517 )
1518 assert any(
1519 a.action_type == "send_mod_note" and a.note == "Notify user: Yes, including the note text\n\nEmailed note"
1520 for a in res.admin_actions
1521 )
1523 # DeleteUser
1524 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
1525 assert any(a.action_type == "delete_user" for a in res.admin_actions)
1526 assert any(
1527 a.action_type == "delete_user" and a.level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH for a in res.admin_actions
1528 )
1530 # RecoverDeletedUser
1531 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
1532 assert any(a.action_type == "recover_user" for a in res.admin_actions)
1534 # MarkUserNeedsLocationUpdate
1535 res = api.MarkUserNeedsLocationUpdate(admin_pb2.MarkUserNeedsLocationUpdateReq(user=normal_user.username))
1536 assert any(
1537 a.action_type == "mark_needs_location_update" and a.note == "Marked user as needing location update"
1538 for a in res.admin_actions
1539 )
1541 # SetLastDonated
1542 res = api.SetLastDonated(
1543 admin_pb2.SetLastDonatedReq(
1544 user=normal_user.username,
1545 last_donated=Timestamp_from_datetime(now()),
1546 )
1547 )
1548 assert any(a.action_type == "set_last_donated" for a in res.admin_actions)
1551def test_create_admin_tag(db):
1552 super_user, super_token = generate_user(is_superuser=True)
1554 with real_admin_session(super_token) as api:
1555 res = api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1556 assert res.tag == "test-tag"
1557 assert res.admin_tag_id > 0
1560def test_create_admin_tag_duplicate(db):
1561 super_user, super_token = generate_user(is_superuser=True)
1563 with real_admin_session(super_token) as api:
1564 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1565 with pytest.raises(grpc.RpcError) as e:
1566 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1567 assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
1568 assert e.value.details() == "That admin tag already exists."
1571def test_create_admin_tag_empty(db):
1572 super_user, super_token = generate_user(is_superuser=True)
1574 with real_admin_session(super_token) as api:
1575 with pytest.raises(grpc.RpcError) as e:
1576 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=""))
1577 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1578 assert e.value.details() == "The admin tag cannot be empty."
1580 with pytest.raises(grpc.RpcError) as e:
1581 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=" "))
1582 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1583 assert e.value.details() == "The admin tag cannot be empty."
1586def test_list_admin_tags(db):
1587 super_user, super_token = generate_user(is_superuser=True)
1589 with real_admin_session(super_token) as api:
1590 # Empty initially
1591 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1592 assert len(res.tags) == 0
1594 # Add some tags
1595 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="bravo"))
1596 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="alpha"))
1598 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1599 assert len(res.tags) == 2
1600 # Ordered alphabetically
1601 assert res.tags[0].tag == "alpha"
1602 assert res.tags[1].tag == "bravo"
1605def test_add_admin_tag_to_user(db):
1606 super_user, super_token = generate_user(is_superuser=True)
1607 normal_user, _ = generate_user()
1609 with real_admin_session(super_token) as api:
1610 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1612 res = api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1613 assert "vip" in res.admin_tags
1614 assert any(a.action_type == "add_tag" and a.tag == "vip" for a in res.admin_actions)
1617def test_add_admin_tag_to_user_duplicate(db):
1618 super_user, super_token = generate_user(is_superuser=True)
1619 normal_user, _ = generate_user()
1621 with real_admin_session(super_token) as api:
1622 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1623 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1625 with pytest.raises(grpc.RpcError) as e:
1626 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1627 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1628 assert e.value.details() == "The user already has that admin tag."
1631def test_add_admin_tag_to_user_tag_not_found(db):
1632 super_user, super_token = generate_user(is_superuser=True)
1633 normal_user, _ = generate_user()
1635 with real_admin_session(super_token) as api:
1636 with pytest.raises(grpc.RpcError) as e:
1637 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="nonexistent"))
1638 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1639 assert e.value.details() == "Admin tag not found."
1642def test_remove_admin_tag_from_user(db):
1643 super_user, super_token = generate_user(is_superuser=True)
1644 normal_user, _ = generate_user()
1646 with real_admin_session(super_token) as api:
1647 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1648 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1650 res = api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1651 assert "vip" not in res.admin_tags
1652 assert any(a.action_type == "remove_tag" and a.tag == "vip" for a in res.admin_actions)
1655def test_remove_admin_tag_from_user_not_assigned(db):
1656 super_user, super_token = generate_user(is_superuser=True)
1657 normal_user, _ = generate_user()
1659 with real_admin_session(super_token) as api:
1660 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1662 with pytest.raises(grpc.RpcError) as e:
1663 api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1664 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1665 assert e.value.details() == "The user does not have that admin tag."
1668def test_search_users_by_admin_tag(db):
1669 super_user, super_token = generate_user(is_superuser=True)
1670 user1, _ = generate_user()
1671 user2, _ = generate_user()
1672 user3, _ = generate_user()
1674 with real_admin_session(super_token) as api:
1675 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1676 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="flagged"))
1678 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user1.username, tag="vip"))
1679 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="vip"))
1680 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="flagged"))
1682 # Search for users with "vip" tag
1683 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip"]))
1684 user_ids = {u.user_id for u in res.users}
1685 assert user1.id in user_ids
1686 assert user2.id in user_ids
1687 assert user3.id not in user_ids
1689 # Search for users with both "vip" AND "flagged" tags (AND logic)
1690 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip", "flagged"]))
1691 user_ids = {u.user_id for u in res.users}
1692 assert user2.id in user_ids
1693 assert user1.id not in user_ids
1695 # Search for non-existent tag returns no results
1696 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["nonexistent"]))
1697 assert len(res.users) == 0
1700def test_search_users_by_admin_note(db):
1701 super_user, super_token = generate_user(is_superuser=True)
1702 user1, _ = generate_user()
1703 user2, _ = generate_user()
1705 with real_admin_session(super_token) as api:
1706 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="suspicious activity"))
1707 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="normal user"))
1709 # Search by admin action log content (ilike)
1710 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_action_log="%suspicious%"))
1711 user_ids = {u.user_id for u in res.users}
1712 assert user1.id in user_ids
1713 assert user2.id not in user_ids
1716def test_ListAdminActions_empty(db):
1717 super_user, super_token = generate_user(is_superuser=True)
1719 with real_admin_session(super_token) as api:
1720 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1721 assert len(res.admin_actions) == 0
1722 assert res.next_page_token == ""
1725def test_ListAdminActions_returns_newest_first_with_target_info(db):
1726 super_user, super_token = generate_user(is_superuser=True)
1727 user1, _ = generate_user()
1728 user2, _ = generate_user()
1730 with real_admin_session(super_token) as api:
1731 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="first note"))
1732 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="second note"))
1733 api.BanUser(admin_pb2.BanUserReq(user=user1.username, admin_note="ban reason"))
1735 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1737 assert len(res.admin_actions) == 3
1738 # Newest first
1739 assert res.admin_actions[0].action_type == "ban"
1740 assert res.admin_actions[0].target_user_id == user1.id
1741 assert res.admin_actions[0].target_username == user1.username
1742 assert res.admin_actions[0].admin_user_id == super_user.id
1743 assert res.admin_actions[0].admin_username == super_user.username
1744 assert res.admin_actions[1].action_type == "note"
1745 assert res.admin_actions[1].target_user_id == user2.id
1746 assert res.admin_actions[2].action_type == "note"
1747 assert res.admin_actions[2].target_user_id == user1.id
1750def test_ListAdminActions_filter_by_admin_and_target(db):
1751 super1, super1_token = generate_user(is_superuser=True)
1752 super2, super2_token = generate_user(is_superuser=True)
1753 user1, _ = generate_user()
1754 user2, _ = generate_user()
1756 with real_admin_session(super1_token) as api:
1757 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super1 to user1"))
1758 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="from super1 to user2"))
1759 with real_admin_session(super2_token) as api:
1760 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super2 to user1"))
1762 with real_admin_session(super1_token) as api:
1763 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id))
1764 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super1 to user2"}
1766 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(target_user_id=user1.id))
1767 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super2 to user1"}
1769 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id, target_user_id=user1.id))
1770 assert [a.note for a in res.admin_actions] == ["from super1 to user1"]
1773def test_ListAdminActions_pagination(db):
1774 super_user, super_token = generate_user(is_superuser=True)
1775 user, _ = generate_user()
1777 with real_admin_session(super_token) as api:
1778 for i in range(3):
1779 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user.username, admin_note=f"note {i}"))
1781 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2))
1782 assert len(res.admin_actions) == 2
1783 assert res.next_page_token != ""
1784 first_page_notes = [a.note for a in res.admin_actions]
1786 res2 = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2, page_token=res.next_page_token))
1787 assert len(res2.admin_actions) == 1
1788 assert res2.next_page_token == ""
1790 all_notes = first_page_notes + [a.note for a in res2.admin_actions]
1791 assert set(all_notes) == {"note 0", "note 1", "note 2"}
1794def test_ListUserUploads(db):
1795 super_user, super_token = generate_user(is_superuser=True)
1796 user, _ = generate_user(complete_profile=False)
1797 other_user, _ = generate_user()
1799 with session_scope() as session:
1800 for i in range(3):
1801 session.add(
1802 Upload(
1803 key=f"key{i}",
1804 filename=f"photo{i}.jpg",
1805 creator_user_id=user.id,
1806 credit=f"credit {i}" if i == 0 else None,
1807 )
1808 )
1809 session.add(Upload(key="other_key", filename="other.jpg", creator_user_id=other_user.id))
1811 with real_admin_session(super_token) as api:
1812 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1814 assert len(res.uploads) == 3
1815 assert res.next_page_token == ""
1816 assert {u.filename for u in res.uploads} == {"photo0.jpg", "photo1.jpg", "photo2.jpg"}
1818 upload0 = next(u for u in res.uploads if u.key == "key0")
1819 assert upload0.credit == "credit 0"
1820 assert upload0.full_url.endswith("/img/full/photo0.jpg")
1821 assert upload0.thumbnail_url.endswith("/img/thumbnail/photo0.jpg")
1822 assert upload0.HasField("created")
1825def test_ListUserUploads_metadata(db):
1826 super_user, super_token = generate_user(is_superuser=True)
1827 user, _ = generate_user(complete_profile=False)
1829 with session_scope() as session:
1830 session.add(
1831 Upload(
1832 key="with_metadata",
1833 filename="photo.jpg",
1834 creator_user_id=user.id,
1835 metadata_exif=b"\x01\x02\x03",
1836 metadata_parsed={"Image Make": "Canon", "EXIF LensModel": "50mm"},
1837 original_filename="IMG_1234.HEIC",
1838 original_format="heif",
1839 original_size=123456,
1840 original_width=4032,
1841 original_height=3024,
1842 )
1843 )
1844 session.add(Upload(key="without_metadata", filename="old.jpg", creator_user_id=user.id))
1846 with real_admin_session(super_token) as api:
1847 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1849 uploads = {u.key: u for u in res.uploads}
1850 assert uploads["without_metadata"].metadata == admin_pb2.UploadMetadata()
1852 metadata = uploads["with_metadata"].metadata
1853 assert json.loads(metadata.parsed_json) == {"Image Make": "Canon", "EXIF LensModel": "50mm"}
1854 assert metadata.parse_error == ""
1855 assert metadata.original_filename == "IMG_1234.HEIC"
1856 assert metadata.original_format == "heif"
1857 assert metadata.original_size == 123456
1858 assert metadata.original_width == 4032
1859 assert metadata.original_height == 3024
1862def test_ListUserUploads_pagination(db):
1863 super_user, super_token = generate_user(is_superuser=True)
1864 user, _ = generate_user(complete_profile=False)
1866 with session_scope() as session:
1867 for i in range(3):
1868 session.add(Upload(key=f"key{i}", filename=f"photo{i}.jpg", creator_user_id=user.id))
1870 with real_admin_session(super_token) as api:
1871 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username, page_size=2))
1872 assert len(res.uploads) == 2
1873 assert res.next_page_token != ""
1874 first_page_keys = [u.key for u in res.uploads]
1876 res2 = api.ListUserUploads(
1877 admin_pb2.ListUserUploadsReq(user=user.username, page_size=2, page_token=res.next_page_token)
1878 )
1879 assert len(res2.uploads) == 1
1880 assert res2.next_page_token == ""
1882 all_keys = first_page_keys + [u.key for u in res2.uploads]
1883 assert set(all_keys) == {"key0", "key1", "key2"}
1886def test_ListUserUploads_uses(db):
1887 super_user, super_token = generate_user(is_superuser=True)
1888 user, _ = generate_user(complete_profile=False)
1890 with session_scope() as session:
1891 session.add(Upload(key="used_key", filename="used.jpg", creator_user_id=user.id))
1892 session.add(Upload(key="unused_key", filename="unused.jpg", creator_user_id=user.id))
1893 gallery = PhotoGallery(owner_user_id=user.id)
1894 session.add(gallery)
1895 session.flush()
1896 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="used_key", position=1.0))
1898 with real_admin_session(super_token) as api:
1899 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1901 uploads = {u.key: u for u in res.uploads}
1902 assert list(uploads["unused_key"].uses) == []
1904 used_uses = uploads["used_key"].uses
1905 assert len(used_uses) == 1
1906 assert used_uses[0].type == admin_pb2.UPLOAD_USE_TYPE_PROFILE_GALLERY_PHOTO_AVATAR
1907 assert used_uses[0].is_current
1908 assert used_uses[0].user_id == user.id
1911def test_ListUserUploads_not_found(db):
1912 super_user, super_token = generate_user(is_superuser=True)
1914 with real_admin_session(super_token) as api:
1915 with pytest.raises(grpc.RpcError) as e:
1916 api.ListUserUploads(admin_pb2.ListUserUploadsReq(user="nonexistent"))
1917 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1920# community invite feature tested in test_events.py
1921# SendBlogPostNotification tested in test_notifications.py
1922# MarkUserNeedsLocationUpdate tested in test_jail.py
1925def _ota_manifest(*, version, fingerprint, created_at="2026-05-31T00:00:00.000Z"):
1926 return {
1927 "id": f"id-{version}",
1928 "createdAt": created_at,
1929 "runtimeVersion": fingerprint,
1930 "launchAsset": {"key": "bundle", "url": f"https://cdn.testing.invalid/{version}/bundle.hbc"},
1931 "assets": [],
1932 "metadata": {},
1933 "extra": {},
1934 }
1937def _ota_signed_multipart(manifest):
1938 # Mimics the signed multipart body the CDN holds (signature header omitted; we only read the JSON).
1939 boundary = "COUCHERS_OTA_BOUNDARY"
1941 def part(name, body, content_type):
1942 return f'--{boundary}\r\ncontent-disposition: form-data; name="{name}"\r\ncontent-type: {content_type}\r\n\r\n{body}\r\n'
1944 body = (
1945 part("manifest", json.dumps(manifest), "application/json; charset=utf-8")
1946 + part("extensions", "{}", "application/json")
1947 + f"--{boundary}--\r\n"
1948 )
1949 return f"multipart/mixed; boundary={boundary}", body.encode()
1952def _patch_ota_cdn(manifests):
1953 # manifests: {version: manifest_dict}. URL is {cdn_root}/{version}/{platform}/manifest.
1954 def fake(url):
1955 version = url.split("/")[-3]
1956 if version not in manifests:
1957 return "multipart/mixed; boundary=COUCHERS_OTA_BOUNDARY", b""
1958 return _ota_signed_multipart(manifests[version])
1960 return patch("couchers.servicers.admin._fetch_signed_manifest", side_effect=fake)
1963def test_CreateOTAPackage(db):
1964 super_user, super_token = generate_user(is_superuser=True)
1966 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1967 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1968 res = api.CreateOTAPackage(
1969 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1970 )
1972 assert res.platform == admin_pb2.OTA_PLATFORM_IOS
1973 assert res.fingerprint == "ios-fp"
1974 assert res.version == "v1.3.1.aaaa"
1975 assert res.manifest_id == "id-v1.3.1.aaaa"
1976 assert res.banned is False
1977 assert res.live is True
1978 assert res.creator_user_id == super_user.id
1981def test_CreateOTAPackage_invalid(db):
1982 _, super_token = generate_user(is_superuser=True)
1984 manifests = {"v-incomplete": {"id": "x"}} # on the CDN but missing runtimeVersion / createdAt
1985 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1986 # missing version
1987 with pytest.raises(grpc.RpcError) as e:
1988 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1989 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1991 # nothing published at this version
1992 with pytest.raises(grpc.RpcError) as e:
1993 api.CreateOTAPackage(
1994 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-missing")
1995 )
1996 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1998 # manifest present but missing required fields
1999 with pytest.raises(grpc.RpcError) as e:
2000 api.CreateOTAPackage(
2001 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-incomplete")
2002 )
2003 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
2006def test_CreateOTAPackage_rejects_duplicate_version(db):
2007 _, super_token = generate_user(is_superuser=True)
2009 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
2010 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
2011 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa"))
2012 with pytest.raises(grpc.RpcError) as e:
2013 api.CreateOTAPackage(
2014 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
2015 )
2016 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
2019def test_ListOTAPackages(db):
2020 _, super_token = generate_user(is_superuser=True)
2022 manifests = {
2023 "v1.3.1.ios": _ota_manifest(version="v1.3.1.ios", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
2024 "v1.3.2.ios": _ota_manifest(version="v1.3.2.ios", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
2025 "v1.3.2.android": _ota_manifest(
2026 version="v1.3.2.android", fingerprint="android-fp", created_at="2026-06-01T00:00:00.000Z"
2027 ),
2028 }
2029 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
2030 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.ios"))
2031 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.ios"))
2032 api.CreateOTAPackage(
2033 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_ANDROID, version="v1.3.2.android")
2034 )
2036 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
2037 # newest (by manifest createdAt) first
2038 assert [p.version for p in res.packages] == ["v1.3.2.android", "v1.3.2.ios", "v1.3.1.ios"]
2039 # only the newest per (platform, fingerprint) is live
2040 live = {p.version: p.live for p in res.packages}
2041 assert live == {"v1.3.2.android": True, "v1.3.2.ios": True, "v1.3.1.ios": False}
2043 ios = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(platform=admin_pb2.OTA_PLATFORM_IOS))
2044 assert [p.version for p in ios.packages] == ["v1.3.2.ios", "v1.3.1.ios"]
2047def test_BanOTAPackage(db):
2048 super_user, super_token = generate_user(is_superuser=True)
2050 manifests = {
2051 "v1.3.1.good": _ota_manifest(
2052 version="v1.3.1.good", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"
2053 ),
2054 "v1.3.2.bad": _ota_manifest(version="v1.3.2.bad", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
2055 }
2056 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
2057 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.good"))
2058 second = api.CreateOTAPackage(
2059 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.bad")
2060 )
2061 assert second.live is True
2063 banned = api.BanOTAPackage(
2064 admin_pb2.BanOTAPackageReq(ota_package_id=second.ota_package_id, reason="bad bundle")
2065 )
2066 assert banned.banned is True
2067 assert banned.banned_reason == "bad bundle"
2068 assert banned.banned_by_user_id == super_user.id
2069 assert banned.live is False
2071 # banning the newest stops new check-ins getting it; the previous one becomes live again
2072 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(include_banned=True))
2073 live = {p.version: p.live for p in res.packages}
2074 assert live == {"v1.3.2.bad": False, "v1.3.1.good": True}
2076 # banned packages are excluded by default
2077 non_banned = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
2078 assert [p.version for p in non_banned.packages] == ["v1.3.1.good"]
2081def test_BanOTAPackage_requires_reason(db):
2082 _, super_token = generate_user(is_superuser=True)
2084 manifests = {
2085 "v1.3.1": _ota_manifest(version="v1.3.1", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
2086 }
2087 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
2088 pkg = api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1"))
2089 with pytest.raises(grpc.RpcError) as e:
2090 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id))
2091 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
2092 with pytest.raises(grpc.RpcError) as e:
2093 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id, reason=" "))
2094 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
2097def test_BanOTAPackage_not_found(db):
2098 _, super_token = generate_user(is_superuser=True)
2100 with real_admin_session(super_token) as api:
2101 with pytest.raises(grpc.RpcError) as e:
2102 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=123456, reason="never mind"))
2103 assert e.value.code() == grpc.StatusCode.NOT_FOUND