Coverage for app/backend/src/tests/test_admin.py: 100%
1155 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-19 15:47 +0000
1import json
2from datetime import UTC, date, datetime, timedelta
3from unittest.mock import patch
5import grpc
6import pytest
7from sqlalchemy import select
8from sqlalchemy.sql import func
10from couchers.db import session_scope
11from couchers.models import (
12 AccountDeletionToken,
13 ContentReport,
14 EventOccurrence,
15 FriendRelationship,
16 FriendStatus,
17 HostRequest,
18 ModerationObjectType,
19 ModerationState,
20 ModerationUserList,
21 ModerationVisibility,
22 NonvisibleUserAccess,
23 NonvisibleUserAccessType,
24 NonvisibleUserState,
25 PhotoGallery,
26 PhotoGalleryItem,
27 Reference,
28 Upload,
29 User,
30 UserActivity,
31 UserSession,
32)
33from couchers.proto import (
34 account_pb2,
35 admin_pb2,
36 auth_pb2,
37 blocking_pb2,
38 events_pb2,
39 references_pb2,
40 reporting_pb2,
41 requests_pb2,
42)
43from couchers.utils import Timestamp_from_datetime, date_to_api, datetime_to_iso8601_local, now, parse_date
44from tests.fixtures.db import (
45 add_users_to_new_moderation_list,
46 backdate_conversations,
47 generate_user,
48 make_friends,
49)
50from tests.fixtures.misc import EmailCollector, Moderator, PushCollector
51from tests.fixtures.sessions import (
52 account_session,
53 auth_api_session,
54 blocking_session,
55 events_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_ShadowUser_blank_note(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 with pytest.raises(grpc.RpcError) as e:
396 api.ShadowUser(admin_pb2.ShadowUserReq(user=normal_user.username, admin_note=" \t "))
397 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
400def test_AddAdminNote(db):
401 super_user, super_token = generate_user(is_superuser=True)
402 normal_user, _ = generate_user()
403 admin_note1 = "User reported strange behavior"
404 admin_note2 = "Insert private information here"
406 with real_admin_session(super_token) as api:
407 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note1))
408 assert res.user_id == normal_user.id
409 assert res.username == normal_user.username
410 assert res.email == normal_user.email
411 assert res.gender == normal_user.gender
412 assert parse_date(res.birthdate) == normal_user.birthdate
413 assert not res.banned
414 assert not res.deleted
415 assert len(res.admin_actions) == 1
416 assert res.admin_actions[0].action_type == "note"
417 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
418 assert res.admin_actions[0].note == admin_note1
420 with real_admin_session(super_token) as api:
421 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note2))
422 assert len(res.admin_actions) == 2
423 assert res.admin_actions[0].note == admin_note1
424 assert res.admin_actions[1].note == admin_note2
427def test_AddAdminNote_blank(db):
428 super_user, super_token = generate_user(is_superuser=True)
429 normal_user, _ = generate_user()
430 empty_admin_note = " \t \n "
432 with real_admin_session(super_token) as api:
433 with pytest.raises(grpc.RpcError) as e:
434 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=empty_admin_note))
435 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
436 assert e.value.details() == "Provide exactly one of admin_note or data."
439def test_AddAdminNote_data(db):
440 super_user, super_token = generate_user(is_superuser=True)
441 normal_user, _ = generate_user()
442 payload = '{"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}'
444 with real_admin_session(super_token) as api:
445 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data=payload))
446 assert len(res.admin_actions) == 1
447 assert res.admin_actions[0].action_type == "note"
448 assert res.admin_actions[0].note == ""
449 assert json.loads(res.admin_actions[0].data) == {"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}
452def test_AddAdminNote_both_note_and_data(db):
453 super_user, super_token = generate_user(is_superuser=True)
454 normal_user, _ = generate_user()
456 with real_admin_session(super_token) as api:
457 with pytest.raises(grpc.RpcError) as e:
458 api.AddAdminNote(
459 admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="note text", data='{"x": 1}')
460 )
461 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
462 assert e.value.details() == "Provide exactly one of admin_note or data."
465def test_AddAdminNote_neither(db):
466 super_user, super_token = generate_user(is_superuser=True)
467 normal_user, _ = generate_user()
469 with real_admin_session(super_token) as api:
470 with pytest.raises(grpc.RpcError) as e:
471 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username))
472 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
473 assert e.value.details() == "Provide exactly one of admin_note or data."
476def test_AddAdminNote_invalid_json(db):
477 super_user, super_token = generate_user(is_superuser=True)
478 normal_user, _ = generate_user()
480 with real_admin_session(super_token) as api:
481 with pytest.raises(grpc.RpcError) as e:
482 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data="{not valid json"))
483 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
484 assert e.value.details() == "The admin note data must be valid JSON."
487def test_admin_content_reports(db):
488 super_user, super_token = generate_user(is_superuser=True)
489 normal_user, token = generate_user()
490 bad_user1, _ = generate_user()
491 bad_user2, _ = generate_user()
493 with reporting_session(token) as api:
494 api.Report(
495 reporting_pb2.ReportReq(
496 reason="spam",
497 description="r1",
498 content_ref="comment/123",
499 author_user=bad_user1.username,
500 user_agent="n/a",
501 page="https://couchers.org/comment/123",
502 )
503 )
504 api.Report(
505 reporting_pb2.ReportReq(
506 reason="spam",
507 description="r2",
508 content_ref="comment/124",
509 author_user=bad_user2.username,
510 user_agent="n/a",
511 page="https://couchers.org/comment/124",
512 )
513 )
514 api.Report(
515 reporting_pb2.ReportReq(
516 reason="something else",
517 description="r3",
518 content_ref="page/321",
519 author_user=bad_user1.username,
520 user_agent="n/a",
521 page="https://couchers.org/page/321",
522 )
523 )
525 with session_scope() as session:
526 id_by_description: dict[str, int] = dict(
527 session.execute(select(ContentReport.description, ContentReport.id)).all() # type: ignore[arg-type]
528 )
530 with real_admin_session(super_token) as api:
531 with pytest.raises(grpc.RpcError) as e:
532 api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=-1))
533 assert e.value.code() == grpc.StatusCode.NOT_FOUND
534 assert e.value.details() == "Content report not found."
536 res = api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=id_by_description["r2"]))
537 rep = res.content_report
538 assert rep.content_report_id == id_by_description["r2"]
539 assert rep.reporting_user_id == normal_user.id
540 assert rep.author_user_id == bad_user2.id
541 assert rep.reason == "spam"
542 assert rep.description == "r2"
543 assert rep.content_ref == "comment/124"
544 assert rep.user_agent == "n/a"
545 assert rep.page == "https://couchers.org/comment/124"
547 res = api.GetContentReportsForAuthor(admin_pb2.GetContentReportsForAuthorReq(user=bad_user1.username))
548 assert res.content_reports[0].content_report_id == id_by_description["r3"]
549 assert res.content_reports[1].content_report_id == id_by_description["r1"]
552def test_DeleteUser(db):
553 super_user, super_token = generate_user(is_superuser=True)
554 normal_user, normal_token = generate_user()
556 with real_admin_session(super_token) as api:
557 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
558 assert res.user_id == normal_user.id
559 assert res.username == normal_user.username
560 assert res.email == normal_user.email
561 assert res.gender == normal_user.gender
562 assert parse_date(res.birthdate) == normal_user.birthdate
563 assert not res.banned
564 assert res.deleted
566 with real_admin_session(super_token) as api:
567 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
568 assert res.user_id == normal_user.id
569 assert res.username == normal_user.username
570 assert res.email == normal_user.email
571 assert res.gender == normal_user.gender
572 assert parse_date(res.birthdate) == normal_user.birthdate
573 assert not res.banned
574 assert not res.deleted
577def test_RecoverDeletedUser_after_user_initiated_deletion(db, push_collector: PushCollector):
578 """
579 When a user deletes their account through the normal flow (ConfirmDeleteAccount),
580 undelete_token and undelete_until are set. The admin RecoverDeletedUser must clear
581 these fields to satisfy the undelete_nullity database constraint.
582 """
583 super_user, super_token = generate_user(is_superuser=True)
584 normal_user, normal_token = generate_user()
585 user_id = normal_user.id
587 # User initiates account deletion
588 with account_session(normal_token) as account:
589 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
591 # Get the deletion confirmation token
592 with session_scope() as session:
593 deletion_token = session.execute(select(AccountDeletionToken)).scalar_one().token
595 # User confirms account deletion (this sets undelete_token and undelete_until)
596 with auth_api_session() as (auth_api, metadata_interceptor):
597 auth_api.ConfirmDeleteAccount(auth_pb2.ConfirmDeleteAccountReq(token=deletion_token))
599 # Verify the user is deleted and has undelete fields set
600 with session_scope() as session:
601 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
602 assert user.deleted_at is not None
603 assert user.undelete_token is not None
604 assert user.undelete_until is not None
606 # Admin recovers the user
607 with real_admin_session(super_token) as api:
608 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
609 assert res.user_id == user_id
610 assert not res.deleted
612 # Verify undelete fields are cleared
613 with session_scope() as session:
614 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
615 assert user.deleted_at is None
616 assert user.undelete_token is None
617 assert user.undelete_until is None
620def test_CreateApiKey(db, email_collector: EmailCollector, push_collector: PushCollector):
621 with session_scope() as session:
622 super_user, super_token = generate_user(is_superuser=True)
623 normal_user, normal_token = generate_user()
625 assert (
626 session.execute(
627 select(func.count())
628 .select_from(UserSession)
629 .where(UserSession.is_api_key == True)
630 .where(UserSession.user_id == normal_user.id)
631 ).scalar_one()
632 == 0
633 )
635 with real_admin_session(super_token) as api:
636 res = api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=normal_user.username))
638 email = email_collector.pop_for_recipient(normal_user.email, last=True)
639 assert email.subject == "[TEST] Your API key for Couchers.org"
641 with session_scope() as session:
642 token = session.execute(
643 select(UserSession.token)
644 .where(UserSession.is_valid)
645 .where(UserSession.is_api_key == True)
646 .where(UserSession.user_id == normal_user.id)
647 ).scalar_one()
649 assert token in email.plain
650 assert token in email.html
652 assert email.recipient == normal_user.email
653 assert "api key" in email.subject.lower()
654 unique_string = "We've issued you with the following API key:"
655 assert unique_string in email.plain
656 assert unique_string in email.html
657 assert "support@couchers.org" in email.plain
658 assert "support@couchers.org" in email.html
660 push = push_collector.pop_for_user(normal_user.id, last=True)
661 assert push.content.title == "API key created"
662 assert push.content.body == "Details were sent to you via email."
665def test_GetChats(db):
666 super_user, super_token = generate_user(is_superuser=True)
667 normal_user, normal_token = generate_user()
669 with real_admin_session(super_token) as api:
670 res = api.GetChats(admin_pb2.GetChatsReq(user=normal_user.username))
671 assert res.user.user_id == normal_user.id
672 assert res.user.username == normal_user.username
673 assert res.user.name == normal_user.name
674 # New user should have no chats
675 assert len(res.host_requests) == 0
676 assert len(res.group_chats) == 0
679def test_badges(db, email_collector: EmailCollector, push_collector: PushCollector):
680 super_user, super_token = generate_user(is_superuser=True)
681 normal_user, normal_token = generate_user()
683 with real_admin_session(super_token) as api:
684 # can add a badge
685 assert "swagster" not in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
686 res = api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
687 assert "swagster" in res.badges
689 # badge emails are disabled by default
690 assert email_collector.count_for_recipient(normal_user.email) == 0
692 push = push_collector.pop_for_user(normal_user.id, last=True)
693 assert push.content.title == "New profile badge: Swagster"
694 assert push.content.body == "The Swagster badge was added to your profile."
696 # can't add/edit special tags
697 with pytest.raises(grpc.RpcError) as e:
698 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="founder"))
699 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
700 assert e.value.details() == "Admins cannot edit that badge."
702 # double add badge
703 with pytest.raises(grpc.RpcError) as e:
704 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
705 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
706 assert e.value.details() == "The user already has that badge."
708 # can remove badge
709 assert "swagster" in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
710 res = api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
711 assert "swagster" not in res.badges
713 # badge emails are disabled by default
714 assert email_collector.count_for_recipient(normal_user.email) == 0
716 push = push_collector.pop_for_user(normal_user.id, last=True)
717 assert push.content.title == "Profile badge removed"
718 assert push.content.body == "The Swagster badge was removed from your profile."
720 # not found on user
721 with pytest.raises(grpc.RpcError) as e:
722 api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
723 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
724 assert e.value.details() == "The user does not have that badge."
726 # not found in general
727 with pytest.raises(grpc.RpcError) as e:
728 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="nonexistentbadge"))
729 assert e.value.code() == grpc.StatusCode.NOT_FOUND
730 assert e.value.details() == "Badge not found."
733def test_DeleteEvent(db):
734 super_user, super_token = generate_user(is_superuser=True)
735 normal_user, normal_token = generate_user()
737 with session_scope() as session:
738 create_community(session, 0, 2, "Community", [normal_user], [], None)
740 start_time = now() + timedelta(hours=2)
741 end_time = start_time + timedelta(hours=3)
742 with events_session(normal_token) as api:
743 res = api.CreateEvent(
744 events_pb2.CreateEventReq(
745 title="Dummy Title",
746 content="Dummy content.",
747 photo_key=None,
748 location=events_pb2.EventLocation(
749 address="Near Null Island",
750 lat=0.1,
751 lng=0.2,
752 ),
753 start_datetime_iso8601_local=datetime_to_iso8601_local(start_time),
754 end_datetime_iso8601_local=datetime_to_iso8601_local(end_time),
755 )
756 )
757 event_id = res.event_id
758 assert not res.is_deleted
760 with session_scope() as session:
761 with real_admin_session(super_token) as api:
762 api.DeleteEvent(
763 admin_pb2.DeleteEventReq(
764 event_id=event_id,
765 )
766 )
767 occurrence = session.get_one(EventOccurrence, ident=event_id)
768 assert occurrence.is_deleted
771def test_ListUserIds(db):
772 super_user, super_token = generate_user(is_superuser=True)
773 normal_user, normal_token = generate_user()
775 with real_admin_session(super_token) as api:
776 res = api.ListUserIds(
777 admin_pb2.ListUserIdsReq(
778 start_time=Timestamp_from_datetime(datetime(2000, 1, 1, tzinfo=UTC)),
779 end_time=Timestamp_from_datetime(now()),
780 )
781 )
782 assert len(res.user_ids) == 2
783 assert sorted(res.user_ids) == sorted([super_user.id, normal_user.id])
785 with real_admin_session(super_token) as api:
786 res = api.ListUserIds(
787 admin_pb2.ListUserIdsReq(start_time=Timestamp_from_datetime(now()), end_time=Timestamp_from_datetime(now()))
788 )
789 assert res.user_ids == []
792def test_EditReferenceText(db):
793 super_user, super_token = generate_user(is_superuser=True)
794 test_new_text = "New Text"
796 user1, user1_token = generate_user()
797 user2, user2_token = generate_user()
798 make_friends(user1, user2)
800 with session_scope() as session:
801 with references_session(user1_token) as api:
802 reference = api.WriteFriendReference(
803 references_pb2.WriteFriendReferenceReq(
804 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
805 )
806 )
808 with real_admin_session(super_token) as admin_api:
809 admin_api.EditReferenceText(
810 admin_pb2.EditReferenceTextReq(reference_id=reference.reference_id, new_text=test_new_text)
811 )
813 session.expire_all()
815 modified_reference = session.execute(
816 select(Reference).where(Reference.id == reference.reference_id)
817 ).scalar_one()
818 assert modified_reference.text == test_new_text
821def test_DeleteReference_deprecated(db):
822 """DeleteReference is deprecated; admins should hide via UMS instead."""
823 super_user, super_token = generate_user(is_superuser=True)
825 user1, user1_token = generate_user()
826 user2, user2_token = generate_user()
827 make_friends(user1, user2)
829 with references_session(user1_token) as api:
830 reference = api.WriteFriendReference(
831 references_pb2.WriteFriendReferenceReq(
832 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
833 )
834 )
836 with real_admin_session(super_token) as admin_api:
837 with pytest.raises(grpc.RpcError) as e:
838 admin_api.DeleteReference(admin_pb2.DeleteReferenceReq(reference_id=reference.reference_id))
839 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
842def test_GetUserReferences(db):
843 super_user, super_token = generate_user(is_superuser=True)
845 user1, user1_token = generate_user()
846 user2, user2_token = generate_user()
847 user3, user3_token = generate_user()
848 make_friends(user1, user2)
849 make_friends(user1, user3)
850 make_friends(user2, user3)
852 # user1 writes reference about user2
853 with references_session(user1_token) as api:
854 ref1 = api.WriteFriendReference(
855 references_pb2.WriteFriendReferenceReq(
856 to_user_id=user2.id,
857 text="Reference from user1 to user2",
858 private_text="",
859 was_appropriate=True,
860 rating=1,
861 )
862 )
864 # user2 writes reference about user1
865 with references_session(user2_token) as api:
866 ref2 = api.WriteFriendReference(
867 references_pb2.WriteFriendReferenceReq(
868 to_user_id=user1.id,
869 text="Reference from user2 to user1",
870 private_text="Private note",
871 was_appropriate=True,
872 rating=0.8,
873 )
874 )
876 # user3 writes reference about user1
877 with references_session(user3_token) as api:
878 ref3 = api.WriteFriendReference(
879 references_pb2.WriteFriendReferenceReq(
880 to_user_id=user1.id,
881 text="Reference from user3 to user1",
882 private_text="",
883 was_appropriate=False,
884 rating=0.5,
885 )
886 )
888 # Test GetUserReferences for user1 (admin view shows everything regardless of UMS state).
889 with real_admin_session(super_token) as admin_api:
890 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
892 # user1 wrote 1 reference
893 assert len(res.references_from) == 1
894 assert res.references_from[0].reference_id == ref1.reference_id
895 assert res.references_from[0].from_user_id == user1.id
896 assert res.references_from[0].to_user_id == user2.id
897 assert res.references_from[0].from_user.user_id == user1.id
898 assert res.references_from[0].from_user.username == user1.username
899 assert res.references_from[0].to_user.user_id == user2.id
900 assert res.references_from[0].to_user.username == user2.username
901 assert res.references_from[0].reference_type == "friend"
902 assert res.references_from[0].text == "Reference from user1 to user2"
903 # freshly written content starts out shadowed
904 assert res.references_from[0].moderation_visibility == "shadowed"
905 assert res.references_from[0].host_request_id == 0
906 assert res.references_from[0].hosting_city == ""
907 assert res.references_from[0].from_date == ""
908 assert res.references_from[0].to_date == ""
909 assert res.references_from[0].status == ""
911 # user1 received 2 references
912 assert len(res.references_to) == 2
913 # Ordered by id descending, so ref3 comes first
914 assert res.references_to[0].reference_id == ref3.reference_id
915 assert res.references_to[0].was_appropriate is False
917 assert res.references_to[1].reference_id == ref2.reference_id
918 assert res.references_to[1].private_text == "Private note"
919 assert res.references_to[1].rating == 0.8
922def test_GetUserReferences_host_request(db):
923 super_user, super_token = generate_user(is_superuser=True)
925 user1, _ = generate_user()
926 user2, _ = generate_user()
928 with session_scope() as session:
929 surfed_ref_id, host_request_id = create_host_reference(session, user1.id, user2.id, timedelta(days=2))
930 hosted_ref_id, _ = create_host_reference(
931 session, user2.id, user1.id, timedelta(days=2), host_request_id=host_request_id
932 )
933 host_request = session.execute(
934 select(HostRequest).where(HostRequest.conversation_id == host_request_id)
935 ).scalar_one()
936 from_date = date_to_api(host_request.from_date)
937 to_date = date_to_api(host_request.to_date)
939 with real_admin_session(super_token) as admin_api:
940 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
942 assert len(res.references_from) == 1
943 surfed_ref = res.references_from[0]
944 assert surfed_ref.reference_id == surfed_ref_id
945 assert surfed_ref.reference_type == "surfed"
946 assert surfed_ref.from_user.user_id == user1.id
947 assert surfed_ref.from_user.username == user1.username
948 assert surfed_ref.to_user.user_id == user2.id
949 assert surfed_ref.to_user.username == user2.username
950 assert surfed_ref.moderation_visibility == "visible"
951 assert surfed_ref.host_request_id == host_request_id
952 assert surfed_ref.hosting_city == "Test City"
953 assert surfed_ref.from_date == from_date
954 assert surfed_ref.to_date == to_date
955 assert surfed_ref.status == "confirmed"
957 assert len(res.references_to) == 1
958 hosted_ref = res.references_to[0]
959 assert hosted_ref.reference_id == hosted_ref_id
960 assert hosted_ref.reference_type == "hosted"
961 assert hosted_ref.from_user.user_id == user2.id
962 assert hosted_ref.to_user.user_id == user1.id
963 assert hosted_ref.host_request_id == host_request_id
964 assert hosted_ref.hosting_city == "Test City"
965 assert hosted_ref.from_date == from_date
966 assert hosted_ref.to_date == to_date
967 assert hosted_ref.status == "confirmed"
970def test_GetUserReferences_not_found(db):
971 super_user, super_token = generate_user(is_superuser=True)
973 with real_admin_session(super_token) as admin_api:
974 with pytest.raises(grpc.RpcError) as e:
975 admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user="nonexistent"))
976 assert e.value.code() == grpc.StatusCode.NOT_FOUND
979def test_GetFriendRequests(db):
980 super_user, super_token = generate_user(is_superuser=True)
982 user1, _ = generate_user()
983 user2, _ = generate_user()
984 user3, _ = generate_user()
985 user4, _ = generate_user()
987 # Create a mix of friend requests directly so we control the state
988 def _add_friend_request(from_user_id, to_user_id, status, visibility, time_responded=None):
989 with session_scope() as session:
990 mod_state = ModerationState(
991 object_type=ModerationObjectType.friend_request,
992 object_id=0,
993 visibility=visibility,
994 )
995 session.add(mod_state)
996 session.flush()
997 rel = FriendRelationship(
998 from_user_id=from_user_id,
999 to_user_id=to_user_id,
1000 status=status,
1001 moderation_state_id=mod_state.id,
1002 time_responded=time_responded,
1003 )
1004 session.add(rel)
1005 session.flush()
1006 mod_state.object_id = rel.id
1008 # user1 -> user2: pending, shadowed
1009 _add_friend_request(user1.id, user2.id, FriendStatus.pending, ModerationVisibility.shadowed)
1010 # user1 -> user3: accepted, visible
1011 _add_friend_request(user1.id, user3.id, FriendStatus.accepted, ModerationVisibility.visible, time_responded=now())
1012 # user4 -> user1: rejected, visible
1013 _add_friend_request(user4.id, user1.id, FriendStatus.rejected, ModerationVisibility.visible, time_responded=now())
1015 with real_admin_session(super_token) as admin_api:
1016 res = admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user=user1.username))
1018 # user1 sent two: to user2 (pending) and to user3 (accepted), ordered by id desc
1019 assert len(res.sent) == 2
1020 assert res.sent[0].from_user.user_id == user1.id
1021 assert res.sent[0].to_user.user_id == user3.id
1022 assert res.sent[0].status == "accepted"
1023 assert res.sent[0].HasField("time_responded")
1024 assert res.sent[0].moderation_visibility == "visible"
1026 assert res.sent[1].from_user.user_id == user1.id
1027 assert res.sent[1].to_user.user_id == user2.id
1028 assert res.sent[1].status == "pending"
1029 assert not res.sent[1].HasField("time_responded")
1030 assert res.sent[1].moderation_visibility == "shadowed"
1032 # user1 received one: from user4 (rejected)
1033 assert len(res.received) == 1
1034 assert res.received[0].from_user.user_id == user4.id
1035 assert res.received[0].to_user.user_id == user1.id
1036 assert res.received[0].status == "rejected"
1039def test_GetFriendRequests_not_found(db):
1040 super_user, super_token = generate_user(is_superuser=True)
1042 with real_admin_session(super_token) as admin_api:
1043 with pytest.raises(grpc.RpcError) as e:
1044 admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user="nonexistent"))
1045 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1048def test_GetUserBlocks(db):
1049 super_user, super_token = generate_user(is_superuser=True)
1051 user1, token1 = generate_user()
1052 user2, _ = generate_user()
1053 user3, token3 = generate_user()
1054 user4, token4 = generate_user()
1056 with blocking_session(token1) as user_blocks:
1057 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user2.username))
1059 with blocking_session(token3) as user_blocks:
1060 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user1.username))
1062 with blocking_session(token4) as user_blocks:
1063 user_blocks.BlockUser(blocking_pb2.BlockUserReq(username=user1.username))
1065 with real_admin_session(super_token) as admin_api:
1066 res = admin_api.GetUserBlocks(admin_pb2.GetUserBlocksReq(user=user1.username))
1068 assert len(res.blocked_users) == 1
1069 assert res.blocked_users[0].user.user_id == user2.id
1070 assert res.blocked_users[0].user.username == user2.username
1071 assert res.blocked_users[0].HasField("time_blocked")
1073 # most recently blocked first
1074 assert [block.user.user_id for block in res.blocking_users] == [user4.id, user3.id]
1076 with real_admin_session(super_token) as admin_api:
1077 res = admin_api.GetUserBlocks(admin_pb2.GetUserBlocksReq(user=user2.username))
1079 assert len(res.blocked_users) == 0
1080 assert [block.user.user_id for block in res.blocking_users] == [user1.id]
1083def test_GetUserBlocks_not_found(db):
1084 super_user, super_token = generate_user(is_superuser=True)
1086 with real_admin_session(super_token) as admin_api:
1087 with pytest.raises(grpc.RpcError) as e:
1088 admin_api.GetUserBlocks(admin_pb2.GetUserBlocksReq(user="nonexistent"))
1089 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1092def test_AddUsersToModerationUserList(db):
1093 super_user, super_token = generate_user(is_superuser=True)
1094 user1, _ = generate_user()
1095 user2, _ = generate_user()
1096 user3, _ = generate_user()
1097 user4, _ = generate_user()
1098 user5, _ = generate_user()
1099 moderation_list_id = add_users_to_new_moderation_list([user1])
1101 with session_scope() as session:
1102 with real_admin_session(super_token) as api:
1103 # Test adding users to a non-existent moderation list (should raise an error)
1104 with pytest.raises(grpc.RpcError) as e:
1105 api.AddUsersToModerationUserList(
1106 admin_pb2.AddUsersToModerationUserListReq(users=[user2.username], moderation_list_id=999),
1107 )
1108 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1109 assert "Moderation user list not found." == e.value.details()
1111 # Test with non-existent user (should raise an error)
1112 with pytest.raises(grpc.RpcError) as e:
1113 api.AddUsersToModerationUserList(
1114 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, "nonexistent"]),
1115 )
1116 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1117 assert "Couldn't find that user." == e.value.details()
1119 # Test successful creation of new moderation list (no moderation_list_id provided)
1120 res = api.AddUsersToModerationUserList(
1121 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, user2.username, user3.username]),
1122 )
1123 assert res.moderation_list_id > 0
1124 with session_scope() as session:
1125 moderation_user_list = session.get(ModerationUserList, res.moderation_list_id)
1126 assert moderation_user_list is not None
1127 assert len(moderation_user_list.users) == 3
1128 assert {user1.id, user2.id, user3.id}.issubset({user.id for user in moderation_user_list.users})
1130 # Test list endpoint returns same moderation list with same members not repeated
1131 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1132 assert len(listRes.moderation_lists) == 1
1133 assert listRes.moderation_lists[0].moderation_list_id == res.moderation_list_id
1134 assert len(listRes.moderation_lists[0].members) == 3
1135 assert {user1.id, user2.id, user3.id}.issubset({m.user_id for m in listRes.moderation_lists[0].members})
1137 # Test user can be in multiple moderation lists
1138 listRes3 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1139 assert len(listRes3.moderation_lists) == 2
1141 # Test adding users to an existing moderation list
1142 res2 = api.AddUsersToModerationUserList(
1143 admin_pb2.AddUsersToModerationUserListReq(
1144 users=[user4.username, user5.username], moderation_list_id=moderation_list_id
1145 ),
1146 )
1147 assert res2.moderation_list_id == moderation_list_id
1148 with session_scope() as session:
1149 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1150 assert len(moderation_user_list.users) == 3
1151 assert {user1.id, user4.id, user5.id}.issubset({user.id for user in moderation_user_list.users})
1153 # Test list user moderation lists endpoint returns the right moderation list
1154 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user5.username))
1155 assert len(listRes2.moderation_lists) == 1
1156 assert listRes2.moderation_lists[0].moderation_list_id == moderation_list_id
1157 assert len(listRes2.moderation_lists[0].members) == 3
1158 assert {user1.id, user4.id, user5.id}.issubset({m.user_id for m in listRes2.moderation_lists[0].members})
1161def test_RemoveUserFromModerationUserList(db):
1162 super_user, super_token = generate_user(is_superuser=True)
1163 user1, _ = generate_user()
1164 user2, _ = generate_user()
1165 user3, _ = generate_user()
1166 moderation_list_id = add_users_to_new_moderation_list([user1, user2])
1168 with real_admin_session(super_token) as api:
1169 # Test with non-existent user (should raise error)
1170 with pytest.raises(grpc.RpcError) as e:
1171 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user="nonexistent"))
1172 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1173 assert "Couldn't find that user." == e.value.details()
1175 # Test without providing moderation list id (should raise error)
1176 with pytest.raises(grpc.RpcError) as e:
1177 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username))
1178 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1179 assert "Missing moderation user list id." == e.value.details()
1181 # Test removing user that's not in the provided moderation list (should raise error)
1182 with pytest.raises(grpc.RpcError) as e:
1183 api.RemoveUserFromModerationUserList(
1184 admin_pb2.RemoveUserFromModerationUserListReq(
1185 user=user3.username, moderation_list_id=moderation_list_id
1186 )
1187 )
1188 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1189 assert "User is not in the moderation user list." == e.value.details()
1191 # Test successful removal
1192 api.RemoveUserFromModerationUserList(
1193 admin_pb2.RemoveUserFromModerationUserListReq(user=user1.username, moderation_list_id=moderation_list_id)
1194 )
1195 with session_scope() as session:
1196 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1197 assert user1.id not in {user.id for user in moderation_user_list.users}
1198 assert user2.id in {user.id for user in moderation_user_list.users}
1200 # Test list user moderation lists endpoint returns right number of moderation lists
1201 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1202 assert len(listRes.moderation_lists) == 0
1203 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1204 assert len(listRes2.moderation_lists) == 1
1206 # Test removing all users from moderation list should also delete the moderation list
1207 api.RemoveUserFromModerationUserList(
1208 admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username, moderation_list_id=moderation_list_id)
1209 )
1210 with session_scope() as session:
1211 assert session.get(ModerationUserList, moderation_list_id) is None
1214def test_admin_delete_account_url(db, email_collector: EmailCollector, push_collector: PushCollector):
1215 super_user, super_token = generate_user(is_superuser=True)
1217 user, token = generate_user()
1218 user_id = user.id
1220 with real_admin_session(super_token) as admin_api:
1221 url = admin_api.CreateAccountDeletionLink(
1222 admin_pb2.CreateAccountDeletionLinkReq(user=user.username)
1223 ).account_deletion_confirm_url
1225 assert push_collector.count_for_user(user_id) == 0
1227 with session_scope() as session:
1228 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
1229 token = token_o.token
1230 assert token_o.user.id == user_id
1231 assert url == f"http://localhost:3000/delete-account?token={token}"
1233 with auth_api_session() as (auth_api, metadata_interceptor):
1234 auth_api.ConfirmDeleteAccount(
1235 auth_pb2.ConfirmDeleteAccountReq(
1236 token=token,
1237 )
1238 )
1240 push = push_collector.pop_for_user(user_id, last=True)
1241 assert push.content.title == "Account deleted"
1242 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
1243 email_collector.pop_for_recipient(user.email, last=True)
1246def test_AccessStats(db):
1247 super_user, super_token = generate_user(is_superuser=True)
1248 normal_user, normal_token = generate_user()
1250 # Insert UserActivity rows: a couple inside the default 90-day window, one well
1251 # outside it, and one with NULL ip_address / user_agent. The INET column is
1252 # returned by psycopg3 as an IPv4Address/IPv6Address object, which used to
1253 # crash the proto string assignment.
1254 in_window_1 = now() - timedelta(days=1)
1255 in_window_2 = now() - timedelta(days=10)
1256 out_of_window = now() - timedelta(days=200)
1257 with session_scope() as session:
1258 session.add(
1259 UserActivity(
1260 user_id=normal_user.id, period=in_window_1, ip_address="1.2.3.4", user_agent="ua-a", api_calls=5
1261 )
1262 )
1263 session.add(
1264 UserActivity(
1265 user_id=normal_user.id, period=in_window_2, ip_address="2001:db8::1", user_agent="ua-b", api_calls=3
1266 )
1267 )
1268 session.add(
1269 UserActivity(
1270 user_id=normal_user.id, period=out_of_window, ip_address="9.9.9.9", user_agent="ua-old", api_calls=99
1271 )
1272 )
1273 session.add(UserActivity(user_id=normal_user.id, period=in_window_1, api_calls=1))
1275 with real_admin_session(super_token) as api:
1276 res = api.AccessStats(admin_pb2.AccessStatsReq(user=normal_user.username))
1278 by_ip = {s.ip_address: s for s in res.stats}
1279 assert "1.2.3.4" in by_ip
1280 assert by_ip["1.2.3.4"].api_call_count == 5
1281 assert by_ip["1.2.3.4"].user_agent == "ua-a"
1282 assert "2001:db8::1" in by_ip
1283 assert by_ip["2001:db8::1"].api_call_count == 3
1284 # NULL ip_address row produces an empty-string ip_address in the proto
1285 assert "" in by_ip
1286 assert by_ip[""].api_call_count == 1
1287 # out-of-window row is excluded by the 90-day default
1288 assert "9.9.9.9" not in by_ip
1290 # explicit end_time should bound the upper end of the window (regression: was >=)
1291 with real_admin_session(super_token) as api:
1292 res = api.AccessStats(
1293 admin_pb2.AccessStatsReq(
1294 user=normal_user.username,
1295 start_time=Timestamp_from_datetime(now() - timedelta(days=5)),
1296 end_time=Timestamp_from_datetime(now()),
1297 )
1298 )
1299 ips = {s.ip_address for s in res.stats}
1300 assert ips == {"1.2.3.4", ""}
1303def test_SetLastDonated(db):
1304 super_user, super_token = generate_user(is_superuser=True)
1305 normal_user, normal_token = generate_user(last_donated=None)
1307 with real_admin_session(super_token) as api:
1308 # user starts with no last_donated
1309 with session_scope() as session:
1310 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1311 assert user.last_donated is None
1313 # can set last_donated
1314 donation_time = now() - timedelta(days=30)
1315 res = api.SetLastDonated(
1316 admin_pb2.SetLastDonatedReq(
1317 user=normal_user.username,
1318 last_donated=Timestamp_from_datetime(donation_time),
1319 )
1320 )
1322 with session_scope() as session:
1323 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1324 assert user.last_donated is not None
1325 # check timestamp is close (within a second)
1326 assert abs((user.last_donated - donation_time).total_seconds()) < 1
1328 # can clear last_donated by not setting the field
1329 res = api.SetLastDonated(admin_pb2.SetLastDonatedReq(user=normal_user.username))
1331 with session_scope() as session:
1332 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1333 assert user.last_donated is None
1335 # user not found
1336 with pytest.raises(grpc.RpcError) as e:
1337 api.SetLastDonated(admin_pb2.SetLastDonatedReq(user="nonexistent"))
1338 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1339 assert e.value.details() == "Couldn't find that user."
1342def test_admin_actions_level(db):
1343 super_user, super_token = generate_user(is_superuser=True)
1344 normal_user, _ = generate_user()
1346 with real_admin_session(super_token) as api:
1347 # Default level is NORMAL
1348 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="normal note"))
1349 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
1351 # Explicitly set to DEBUG
1352 res = api.AddAdminNote(
1353 admin_pb2.AddAdminNoteReq(
1354 user=normal_user.username,
1355 admin_note="debug note",
1356 level=admin_pb2.ADMIN_ACTION_LEVEL_DEBUG,
1357 )
1358 )
1359 assert len(res.admin_actions) == 2
1360 assert res.admin_actions[1].level == admin_pb2.ADMIN_ACTION_LEVEL_DEBUG
1362 # Explicitly set to HIGH
1363 res = api.AddAdminNote(
1364 admin_pb2.AddAdminNoteReq(
1365 user=normal_user.username,
1366 admin_note="high note",
1367 level=admin_pb2.ADMIN_ACTION_LEVEL_HIGH,
1368 )
1369 )
1370 assert len(res.admin_actions) == 3
1371 assert res.admin_actions[2].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
1373 # Explicitly set to TRACE
1374 res = api.AddAdminNote(
1375 admin_pb2.AddAdminNoteReq(
1376 user=normal_user.username,
1377 admin_note="trace note",
1378 level=admin_pb2.ADMIN_ACTION_LEVEL_TRACE,
1379 )
1380 )
1381 assert len(res.admin_actions) == 4
1382 assert res.admin_actions[3].level == admin_pb2.ADMIN_ACTION_LEVEL_TRACE
1385def test_admin_actions_on_mutations(db, push_collector: PushCollector):
1386 super_user, super_token = generate_user(is_superuser=True)
1387 normal_user, _ = generate_user()
1389 original_gender = normal_user.gender
1390 original_birthdate = normal_user.birthdate
1392 with real_admin_session(super_token) as api:
1393 # ChangeUserGender
1394 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
1395 assert any(
1396 a.action_type == "change_gender" and a.note == f"Changed from '{original_gender}' to 'Machine'"
1397 for a in res.admin_actions
1398 )
1400 # ChangeUserBirthdate
1401 res = api.ChangeUserBirthdate(
1402 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-01-01")
1403 )
1404 assert any(
1405 a.action_type == "change_birthdate" and a.note == f"Changed from {original_birthdate} to 1990-01-01"
1406 for a in res.admin_actions
1407 )
1409 # SetPassportSexGenderException
1410 res = api.SetPassportSexGenderException(
1411 admin_pb2.SetPassportSexGenderExceptionReq(user=normal_user.username, passport_sex_gender_exception=True)
1412 )
1413 assert any(
1414 a.action_type == "set_passport_sex_gender_exception" and a.note == "Changed from False to True"
1415 for a in res.admin_actions
1416 )
1418 # SendModNote with notify
1419 res = api.SendModNote(
1420 admin_pb2.SendModNoteReq(
1421 user=normal_user.username,
1422 content="Please update your profile",
1423 internal_id="test1",
1424 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY,
1425 )
1426 )
1427 assert any(
1428 a.action_type == "send_mod_note" and a.note == "Notify user: Yes\n\nPlease update your profile"
1429 for a in res.admin_actions
1430 )
1432 # SendModNote without notifying
1433 res = api.SendModNote(
1434 admin_pb2.SendModNoteReq(
1435 user=normal_user.username,
1436 content="Silent note",
1437 internal_id="test2",
1438 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NONE,
1439 )
1440 )
1441 assert any(
1442 a.action_type == "send_mod_note" and a.note == "Notify user: No\n\nSilent note" for a in res.admin_actions
1443 )
1445 # SendModNote including the note text in the email
1446 res = api.SendModNote(
1447 admin_pb2.SendModNoteReq(
1448 user=normal_user.username,
1449 content="Emailed note",
1450 internal_id="test3",
1451 notification=admin_pb2.MOD_NOTE_NOTIFICATION_NOTIFY_WITH_CONTENT,
1452 )
1453 )
1454 assert any(
1455 a.action_type == "send_mod_note" and a.note == "Notify user: Yes, including the note text\n\nEmailed note"
1456 for a in res.admin_actions
1457 )
1459 # DeleteUser
1460 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
1461 assert any(a.action_type == "delete_user" for a in res.admin_actions)
1462 assert any(
1463 a.action_type == "delete_user" and a.level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH for a in res.admin_actions
1464 )
1466 # RecoverDeletedUser
1467 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
1468 assert any(a.action_type == "recover_user" for a in res.admin_actions)
1470 # MarkUserNeedsLocationUpdate
1471 res = api.MarkUserNeedsLocationUpdate(admin_pb2.MarkUserNeedsLocationUpdateReq(user=normal_user.username))
1472 assert any(
1473 a.action_type == "mark_needs_location_update" and a.note == "Marked user as needing location update"
1474 for a in res.admin_actions
1475 )
1477 # SetLastDonated
1478 res = api.SetLastDonated(
1479 admin_pb2.SetLastDonatedReq(
1480 user=normal_user.username,
1481 last_donated=Timestamp_from_datetime(now()),
1482 )
1483 )
1484 assert any(a.action_type == "set_last_donated" for a in res.admin_actions)
1487def test_create_admin_tag(db):
1488 super_user, super_token = generate_user(is_superuser=True)
1490 with real_admin_session(super_token) as api:
1491 res = api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1492 assert res.tag == "test-tag"
1493 assert res.admin_tag_id > 0
1496def test_create_admin_tag_duplicate(db):
1497 super_user, super_token = generate_user(is_superuser=True)
1499 with real_admin_session(super_token) as api:
1500 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1501 with pytest.raises(grpc.RpcError) as e:
1502 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1503 assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
1504 assert e.value.details() == "That admin tag already exists."
1507def test_create_admin_tag_empty(db):
1508 super_user, super_token = generate_user(is_superuser=True)
1510 with real_admin_session(super_token) as api:
1511 with pytest.raises(grpc.RpcError) as e:
1512 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=""))
1513 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1514 assert e.value.details() == "The admin tag cannot be empty."
1516 with pytest.raises(grpc.RpcError) as e:
1517 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=" "))
1518 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1519 assert e.value.details() == "The admin tag cannot be empty."
1522def test_list_admin_tags(db):
1523 super_user, super_token = generate_user(is_superuser=True)
1525 with real_admin_session(super_token) as api:
1526 # Empty initially
1527 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1528 assert len(res.tags) == 0
1530 # Add some tags
1531 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="bravo"))
1532 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="alpha"))
1534 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1535 assert len(res.tags) == 2
1536 # Ordered alphabetically
1537 assert res.tags[0].tag == "alpha"
1538 assert res.tags[1].tag == "bravo"
1541def test_add_admin_tag_to_user(db):
1542 super_user, super_token = generate_user(is_superuser=True)
1543 normal_user, _ = generate_user()
1545 with real_admin_session(super_token) as api:
1546 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1548 res = api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1549 assert "vip" in res.admin_tags
1550 assert any(a.action_type == "add_tag" and a.tag == "vip" for a in res.admin_actions)
1553def test_add_admin_tag_to_user_duplicate(db):
1554 super_user, super_token = generate_user(is_superuser=True)
1555 normal_user, _ = generate_user()
1557 with real_admin_session(super_token) as api:
1558 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1559 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1561 with pytest.raises(grpc.RpcError) as e:
1562 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1563 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1564 assert e.value.details() == "The user already has that admin tag."
1567def test_add_admin_tag_to_user_tag_not_found(db):
1568 super_user, super_token = generate_user(is_superuser=True)
1569 normal_user, _ = generate_user()
1571 with real_admin_session(super_token) as api:
1572 with pytest.raises(grpc.RpcError) as e:
1573 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="nonexistent"))
1574 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1575 assert e.value.details() == "Admin tag not found."
1578def test_remove_admin_tag_from_user(db):
1579 super_user, super_token = generate_user(is_superuser=True)
1580 normal_user, _ = generate_user()
1582 with real_admin_session(super_token) as api:
1583 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1584 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1586 res = api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1587 assert "vip" not in res.admin_tags
1588 assert any(a.action_type == "remove_tag" and a.tag == "vip" for a in res.admin_actions)
1591def test_remove_admin_tag_from_user_not_assigned(db):
1592 super_user, super_token = generate_user(is_superuser=True)
1593 normal_user, _ = generate_user()
1595 with real_admin_session(super_token) as api:
1596 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1598 with pytest.raises(grpc.RpcError) as e:
1599 api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1600 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1601 assert e.value.details() == "The user does not have that admin tag."
1604def test_search_users_by_admin_tag(db):
1605 super_user, super_token = generate_user(is_superuser=True)
1606 user1, _ = generate_user()
1607 user2, _ = generate_user()
1608 user3, _ = generate_user()
1610 with real_admin_session(super_token) as api:
1611 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1612 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="flagged"))
1614 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user1.username, tag="vip"))
1615 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="vip"))
1616 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="flagged"))
1618 # Search for users with "vip" tag
1619 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip"]))
1620 user_ids = {u.user_id for u in res.users}
1621 assert user1.id in user_ids
1622 assert user2.id in user_ids
1623 assert user3.id not in user_ids
1625 # Search for users with both "vip" AND "flagged" tags (AND logic)
1626 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip", "flagged"]))
1627 user_ids = {u.user_id for u in res.users}
1628 assert user2.id in user_ids
1629 assert user1.id not in user_ids
1631 # Search for non-existent tag returns no results
1632 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["nonexistent"]))
1633 assert len(res.users) == 0
1636def test_search_users_by_admin_note(db):
1637 super_user, super_token = generate_user(is_superuser=True)
1638 user1, _ = generate_user()
1639 user2, _ = generate_user()
1641 with real_admin_session(super_token) as api:
1642 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="suspicious activity"))
1643 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="normal user"))
1645 # Search by admin action log content (ilike)
1646 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_action_log="%suspicious%"))
1647 user_ids = {u.user_id for u in res.users}
1648 assert user1.id in user_ids
1649 assert user2.id not in user_ids
1652def test_ListAdminActions_empty(db):
1653 super_user, super_token = generate_user(is_superuser=True)
1655 with real_admin_session(super_token) as api:
1656 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1657 assert len(res.admin_actions) == 0
1658 assert res.next_page_token == ""
1661def test_ListAdminActions_returns_newest_first_with_target_info(db):
1662 super_user, super_token = generate_user(is_superuser=True)
1663 user1, _ = generate_user()
1664 user2, _ = generate_user()
1666 with real_admin_session(super_token) as api:
1667 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="first note"))
1668 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="second note"))
1669 api.BanUser(admin_pb2.BanUserReq(user=user1.username, admin_note="ban reason"))
1671 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1673 assert len(res.admin_actions) == 3
1674 # Newest first
1675 assert res.admin_actions[0].action_type == "ban"
1676 assert res.admin_actions[0].target_user_id == user1.id
1677 assert res.admin_actions[0].target_username == user1.username
1678 assert res.admin_actions[0].admin_user_id == super_user.id
1679 assert res.admin_actions[0].admin_username == super_user.username
1680 assert res.admin_actions[1].action_type == "note"
1681 assert res.admin_actions[1].target_user_id == user2.id
1682 assert res.admin_actions[2].action_type == "note"
1683 assert res.admin_actions[2].target_user_id == user1.id
1686def test_ListAdminActions_filter_by_admin_and_target(db):
1687 super1, super1_token = generate_user(is_superuser=True)
1688 super2, super2_token = generate_user(is_superuser=True)
1689 user1, _ = generate_user()
1690 user2, _ = generate_user()
1692 with real_admin_session(super1_token) as api:
1693 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super1 to user1"))
1694 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="from super1 to user2"))
1695 with real_admin_session(super2_token) as api:
1696 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super2 to user1"))
1698 with real_admin_session(super1_token) as api:
1699 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id))
1700 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super1 to user2"}
1702 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(target_user_id=user1.id))
1703 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super2 to user1"}
1705 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id, target_user_id=user1.id))
1706 assert [a.note for a in res.admin_actions] == ["from super1 to user1"]
1709def test_ListAdminActions_pagination(db):
1710 super_user, super_token = generate_user(is_superuser=True)
1711 user, _ = generate_user()
1713 with real_admin_session(super_token) as api:
1714 for i in range(3):
1715 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user.username, admin_note=f"note {i}"))
1717 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2))
1718 assert len(res.admin_actions) == 2
1719 assert res.next_page_token != ""
1720 first_page_notes = [a.note for a in res.admin_actions]
1722 res2 = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2, page_token=res.next_page_token))
1723 assert len(res2.admin_actions) == 1
1724 assert res2.next_page_token == ""
1726 all_notes = first_page_notes + [a.note for a in res2.admin_actions]
1727 assert set(all_notes) == {"note 0", "note 1", "note 2"}
1730def test_ListUserUploads(db):
1731 super_user, super_token = generate_user(is_superuser=True)
1732 user, _ = generate_user(complete_profile=False)
1733 other_user, _ = generate_user()
1735 with session_scope() as session:
1736 for i in range(3):
1737 session.add(
1738 Upload(
1739 key=f"key{i}",
1740 filename=f"photo{i}.jpg",
1741 creator_user_id=user.id,
1742 credit=f"credit {i}" if i == 0 else None,
1743 )
1744 )
1745 session.add(Upload(key="other_key", filename="other.jpg", creator_user_id=other_user.id))
1747 with real_admin_session(super_token) as api:
1748 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1750 assert len(res.uploads) == 3
1751 assert res.next_page_token == ""
1752 assert {u.filename for u in res.uploads} == {"photo0.jpg", "photo1.jpg", "photo2.jpg"}
1754 upload0 = next(u for u in res.uploads if u.key == "key0")
1755 assert upload0.credit == "credit 0"
1756 assert upload0.full_url.endswith("/img/full/photo0.jpg")
1757 assert upload0.thumbnail_url.endswith("/img/thumbnail/photo0.jpg")
1758 assert upload0.HasField("created")
1761def test_ListUserUploads_metadata(db):
1762 super_user, super_token = generate_user(is_superuser=True)
1763 user, _ = generate_user(complete_profile=False)
1765 with session_scope() as session:
1766 session.add(
1767 Upload(
1768 key="with_metadata",
1769 filename="photo.jpg",
1770 creator_user_id=user.id,
1771 metadata_exif=b"\x01\x02\x03",
1772 metadata_parsed={"Image Make": "Canon", "EXIF LensModel": "50mm"},
1773 original_filename="IMG_1234.HEIC",
1774 original_format="heif",
1775 original_size=123456,
1776 original_width=4032,
1777 original_height=3024,
1778 )
1779 )
1780 session.add(Upload(key="without_metadata", filename="old.jpg", creator_user_id=user.id))
1782 with real_admin_session(super_token) as api:
1783 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1785 uploads = {u.key: u for u in res.uploads}
1786 assert uploads["without_metadata"].metadata == admin_pb2.UploadMetadata()
1788 metadata = uploads["with_metadata"].metadata
1789 assert json.loads(metadata.parsed_json) == {"Image Make": "Canon", "EXIF LensModel": "50mm"}
1790 assert metadata.parse_error == ""
1791 assert metadata.original_filename == "IMG_1234.HEIC"
1792 assert metadata.original_format == "heif"
1793 assert metadata.original_size == 123456
1794 assert metadata.original_width == 4032
1795 assert metadata.original_height == 3024
1798def test_ListUserUploads_pagination(db):
1799 super_user, super_token = generate_user(is_superuser=True)
1800 user, _ = generate_user(complete_profile=False)
1802 with session_scope() as session:
1803 for i in range(3):
1804 session.add(Upload(key=f"key{i}", filename=f"photo{i}.jpg", creator_user_id=user.id))
1806 with real_admin_session(super_token) as api:
1807 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username, page_size=2))
1808 assert len(res.uploads) == 2
1809 assert res.next_page_token != ""
1810 first_page_keys = [u.key for u in res.uploads]
1812 res2 = api.ListUserUploads(
1813 admin_pb2.ListUserUploadsReq(user=user.username, page_size=2, page_token=res.next_page_token)
1814 )
1815 assert len(res2.uploads) == 1
1816 assert res2.next_page_token == ""
1818 all_keys = first_page_keys + [u.key for u in res2.uploads]
1819 assert set(all_keys) == {"key0", "key1", "key2"}
1822def test_ListUserUploads_uses(db):
1823 super_user, super_token = generate_user(is_superuser=True)
1824 user, _ = generate_user(complete_profile=False)
1826 with session_scope() as session:
1827 session.add(Upload(key="used_key", filename="used.jpg", creator_user_id=user.id))
1828 session.add(Upload(key="unused_key", filename="unused.jpg", creator_user_id=user.id))
1829 gallery = PhotoGallery(owner_user_id=user.id)
1830 session.add(gallery)
1831 session.flush()
1832 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="used_key", position=1.0))
1834 with real_admin_session(super_token) as api:
1835 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1837 uploads = {u.key: u for u in res.uploads}
1838 assert list(uploads["unused_key"].uses) == []
1840 used_uses = uploads["used_key"].uses
1841 assert len(used_uses) == 1
1842 assert used_uses[0].type == admin_pb2.UPLOAD_USE_TYPE_PROFILE_GALLERY_PHOTO_AVATAR
1843 assert used_uses[0].is_current
1844 assert used_uses[0].user_id == user.id
1847def test_ListUserUploads_not_found(db):
1848 super_user, super_token = generate_user(is_superuser=True)
1850 with real_admin_session(super_token) as api:
1851 with pytest.raises(grpc.RpcError) as e:
1852 api.ListUserUploads(admin_pb2.ListUserUploadsReq(user="nonexistent"))
1853 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1856# community invite feature tested in test_events.py
1857# SendBlogPostNotification tested in test_notifications.py
1858# MarkUserNeedsLocationUpdate tested in test_jail.py
1861def _ota_manifest(*, version, fingerprint, created_at="2026-05-31T00:00:00.000Z"):
1862 return {
1863 "id": f"id-{version}",
1864 "createdAt": created_at,
1865 "runtimeVersion": fingerprint,
1866 "launchAsset": {"key": "bundle", "url": f"https://cdn.testing.invalid/{version}/bundle.hbc"},
1867 "assets": [],
1868 "metadata": {},
1869 "extra": {},
1870 }
1873def _ota_signed_multipart(manifest):
1874 # Mimics the signed multipart body the CDN holds (signature header omitted; we only read the JSON).
1875 boundary = "COUCHERS_OTA_BOUNDARY"
1877 def part(name, body, content_type):
1878 return f'--{boundary}\r\ncontent-disposition: form-data; name="{name}"\r\ncontent-type: {content_type}\r\n\r\n{body}\r\n'
1880 body = (
1881 part("manifest", json.dumps(manifest), "application/json; charset=utf-8")
1882 + part("extensions", "{}", "application/json")
1883 + f"--{boundary}--\r\n"
1884 )
1885 return f"multipart/mixed; boundary={boundary}", body.encode()
1888def _patch_ota_cdn(manifests):
1889 # manifests: {version: manifest_dict}. URL is {cdn_root}/{version}/{platform}/manifest.
1890 def fake(url):
1891 version = url.split("/")[-3]
1892 if version not in manifests:
1893 return "multipart/mixed; boundary=COUCHERS_OTA_BOUNDARY", b""
1894 return _ota_signed_multipart(manifests[version])
1896 return patch("couchers.servicers.admin._fetch_signed_manifest", side_effect=fake)
1899def test_CreateOTAPackage(db):
1900 super_user, super_token = generate_user(is_superuser=True)
1902 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1903 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1904 res = api.CreateOTAPackage(
1905 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1906 )
1908 assert res.platform == admin_pb2.OTA_PLATFORM_IOS
1909 assert res.fingerprint == "ios-fp"
1910 assert res.version == "v1.3.1.aaaa"
1911 assert res.manifest_id == "id-v1.3.1.aaaa"
1912 assert res.banned is False
1913 assert res.live is True
1914 assert res.creator_user_id == super_user.id
1917def test_CreateOTAPackage_invalid(db):
1918 _, super_token = generate_user(is_superuser=True)
1920 manifests = {"v-incomplete": {"id": "x"}} # on the CDN but missing runtimeVersion / createdAt
1921 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1922 # missing version
1923 with pytest.raises(grpc.RpcError) as e:
1924 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1925 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1927 # nothing published at this version
1928 with pytest.raises(grpc.RpcError) as e:
1929 api.CreateOTAPackage(
1930 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-missing")
1931 )
1932 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1934 # manifest present but missing required fields
1935 with pytest.raises(grpc.RpcError) as e:
1936 api.CreateOTAPackage(
1937 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-incomplete")
1938 )
1939 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1942def test_CreateOTAPackage_rejects_duplicate_version(db):
1943 _, super_token = generate_user(is_superuser=True)
1945 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1946 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1947 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa"))
1948 with pytest.raises(grpc.RpcError) as e:
1949 api.CreateOTAPackage(
1950 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1951 )
1952 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1955def test_ListOTAPackages(db):
1956 _, super_token = generate_user(is_superuser=True)
1958 manifests = {
1959 "v1.3.1.ios": _ota_manifest(version="v1.3.1.ios", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1960 "v1.3.2.ios": _ota_manifest(version="v1.3.2.ios", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1961 "v1.3.2.android": _ota_manifest(
1962 version="v1.3.2.android", fingerprint="android-fp", created_at="2026-06-01T00:00:00.000Z"
1963 ),
1964 }
1965 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1966 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.ios"))
1967 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.ios"))
1968 api.CreateOTAPackage(
1969 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_ANDROID, version="v1.3.2.android")
1970 )
1972 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1973 # newest (by manifest createdAt) first
1974 assert [p.version for p in res.packages] == ["v1.3.2.android", "v1.3.2.ios", "v1.3.1.ios"]
1975 # only the newest per (platform, fingerprint) is live
1976 live = {p.version: p.live for p in res.packages}
1977 assert live == {"v1.3.2.android": True, "v1.3.2.ios": True, "v1.3.1.ios": False}
1979 ios = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1980 assert [p.version for p in ios.packages] == ["v1.3.2.ios", "v1.3.1.ios"]
1983def test_BanOTAPackage(db):
1984 super_user, super_token = generate_user(is_superuser=True)
1986 manifests = {
1987 "v1.3.1.good": _ota_manifest(
1988 version="v1.3.1.good", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"
1989 ),
1990 "v1.3.2.bad": _ota_manifest(version="v1.3.2.bad", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1991 }
1992 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1993 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.good"))
1994 second = api.CreateOTAPackage(
1995 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.bad")
1996 )
1997 assert second.live is True
1999 banned = api.BanOTAPackage(
2000 admin_pb2.BanOTAPackageReq(ota_package_id=second.ota_package_id, reason="bad bundle")
2001 )
2002 assert banned.banned is True
2003 assert banned.banned_reason == "bad bundle"
2004 assert banned.banned_by_user_id == super_user.id
2005 assert banned.live is False
2007 # banning the newest stops new check-ins getting it; the previous one becomes live again
2008 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(include_banned=True))
2009 live = {p.version: p.live for p in res.packages}
2010 assert live == {"v1.3.2.bad": False, "v1.3.1.good": True}
2012 # banned packages are excluded by default
2013 non_banned = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
2014 assert [p.version for p in non_banned.packages] == ["v1.3.1.good"]
2017def test_BanOTAPackage_requires_reason(db):
2018 _, super_token = generate_user(is_superuser=True)
2020 manifests = {
2021 "v1.3.1": _ota_manifest(version="v1.3.1", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
2022 }
2023 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
2024 pkg = api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1"))
2025 with pytest.raises(grpc.RpcError) as e:
2026 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id))
2027 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
2028 with pytest.raises(grpc.RpcError) as e:
2029 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id, reason=" "))
2030 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
2033def test_BanOTAPackage_not_found(db):
2034 _, super_token = generate_user(is_superuser=True)
2036 with real_admin_session(super_token) as api:
2037 with pytest.raises(grpc.RpcError) as e:
2038 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=123456, reason="never mind"))
2039 assert e.value.code() == grpc.StatusCode.NOT_FOUND