Coverage for app/backend/src/tests/test_admin.py: 100%
1063 statements
« prev ^ index » next coverage.py v7.15.3, created at 2026-08-04 22:32 +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 ModerationObjectType,
18 ModerationState,
19 ModerationUserList,
20 ModerationVisibility,
21 NonvisibleUserAccess,
22 NonvisibleUserAccessType,
23 NonvisibleUserState,
24 PhotoGallery,
25 PhotoGalleryItem,
26 Reference,
27 Upload,
28 User,
29 UserActivity,
30 UserSession,
31)
32from couchers.proto import (
33 account_pb2,
34 admin_pb2,
35 auth_pb2,
36 events_pb2,
37 references_pb2,
38 reporting_pb2,
39 requests_pb2,
40)
41from couchers.utils import Timestamp_from_datetime, datetime_to_iso8601_local, now, parse_date
42from tests.fixtures.db import (
43 add_users_to_new_moderation_list,
44 backdate_conversations,
45 generate_user,
46 make_friends,
47)
48from tests.fixtures.misc import EmailCollector, PushCollector
49from tests.fixtures.sessions import (
50 account_session,
51 auth_api_session,
52 events_session,
53 real_admin_session,
54 references_session,
55 reporting_session,
56 requests_session,
57)
58from tests.test_communities import create_community
59from tests.test_requests import valid_request_text
62@pytest.fixture(autouse=True)
63def _(testconfig):
64 pass
67def test_access_by_normal_user(db):
68 normal_user, normal_token = generate_user()
70 with real_admin_session(normal_token) as api:
71 # all requests to the admin servicer should break when done by a non-super_user
72 with pytest.raises(grpc.RpcError) as e:
73 api.GetUserDetails(
74 admin_pb2.GetUserDetailsReq(
75 user=str(normal_user.id),
76 )
77 )
78 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
81def test_GetNonvisibleUserAccessLog(db):
82 super_user, super_token = generate_user(is_superuser=True)
83 target, _ = generate_user(username="target")
84 viewer, _ = generate_user(username="viewer")
86 with session_scope() as session:
87 session.add(
88 NonvisibleUserAccess(
89 access_type=NonvisibleUserAccessType.login_attempt,
90 target_user_id=target.id,
91 target_state=NonvisibleUserState.banned,
92 actor_user_id=target.id,
93 ip_address="1.2.3.4",
94 sofa="device-cookie",
95 )
96 )
97 session.add(
98 NonvisibleUserAccess(
99 access_type=NonvisibleUserAccessType.ghost_served,
100 target_user_id=target.id,
101 target_state=NonvisibleUserState.banned,
102 actor_user_id=viewer.id,
103 )
104 )
105 session.add(
106 NonvisibleUserAccess(
107 access_type=NonvisibleUserAccessType.ghost_served,
108 target_user_id=target.id,
109 target_state=NonvisibleUserState.banned,
110 actor_user_id=None,
111 )
112 )
114 with real_admin_session(super_token) as api:
115 res = api.GetNonvisibleUserAccessLog(admin_pb2.GetNonvisibleUserAccessLogReq(user="target"))
117 assert len(res.entries) == 3
118 for entry in res.entries:
119 assert entry.target_user_id == target.id
120 assert entry.target_state == admin_pb2.NONVISIBLE_USER_STATE_BANNED
122 login = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_LOGIN_ATTEMPT]
123 views = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_GHOST_SERVED]
124 assert len(login) == 1
125 assert len(views) == 2
127 assert login[0].actor_user_id.value == target.id
128 assert login[0].actor_username == "target"
129 assert login[0].ip_address == "1.2.3.4"
130 assert login[0].sofa == "device-cookie"
132 logged_in_view = [e for e in views if e.actor_username == "viewer"]
133 logged_out_view = [e for e in views if not e.actor_username]
134 assert len(logged_in_view) == 1
135 assert logged_in_view[0].actor_user_id.value == viewer.id
136 assert len(logged_out_view) == 1
137 assert not logged_out_view[0].HasField("actor_user_id")
140def test_GetUser(db):
141 super_user, super_token = generate_user(is_superuser=True)
142 normal_user, normal_token = generate_user()
144 with real_admin_session(super_token) as api:
145 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
146 assert res.user_id == normal_user.id
147 assert res.username == normal_user.username
149 with real_admin_session(super_token) as api:
150 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note="Testing banning"))
152 with real_admin_session(super_token) as api:
153 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
154 assert res.user_id == normal_user.id
155 assert res.username == normal_user.username
158def test_GetUserDetails(db):
159 super_user, super_token = generate_user(is_superuser=True)
160 normal_user, normal_token = generate_user()
162 with real_admin_session(super_token) as api:
163 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=str(normal_user.id)))
164 assert res.user_id == normal_user.id
165 assert res.username == normal_user.username
166 assert res.email == normal_user.email
167 assert res.gender == normal_user.gender
168 assert parse_date(res.birthdate) == normal_user.birthdate
169 assert not res.banned
170 assert not res.deleted
172 with real_admin_session(super_token) as api:
173 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
174 assert res.user_id == normal_user.id
175 assert res.username == normal_user.username
176 assert res.email == normal_user.email
177 assert res.gender == normal_user.gender
178 assert parse_date(res.birthdate) == normal_user.birthdate
179 assert not res.banned
180 assert not res.deleted
182 with real_admin_session(super_token) as api:
183 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.email))
184 assert res.user_id == normal_user.id
185 assert res.username == normal_user.username
186 assert res.email == normal_user.email
187 assert res.gender == normal_user.gender
188 assert parse_date(res.birthdate) == normal_user.birthdate
189 assert not res.banned
190 assert not res.deleted
193def test_ChangeUserGender(db, email_collector: EmailCollector, push_collector: PushCollector):
194 super_user, super_token = generate_user(is_superuser=True)
195 normal_user, normal_token = generate_user()
197 with real_admin_session(super_token) as api:
198 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
199 assert res.user_id == normal_user.id
200 assert res.username == normal_user.username
201 assert res.email == normal_user.email
202 assert res.gender == "Machine"
203 assert parse_date(res.birthdate) == normal_user.birthdate
204 assert not res.banned
205 assert not res.deleted
207 email = email_collector.pop_for_recipient(normal_user.email, last=True)
208 assert email.subject == "[TEST] Your gender was changed"
209 assert email.recipient == normal_user.email
210 assert "Machine" in email.plain
211 assert "Machine" in email.html
213 push = push_collector.pop_for_user(normal_user.id, last=True)
214 assert push.content.title == "Gender changed"
215 assert push.content.body == "An admin changed your gender to Machine."
218def test_ChangeUserBirthdate(db, email_collector: EmailCollector, push_collector: PushCollector):
219 super_user, super_token = generate_user(is_superuser=True)
220 normal_user, normal_token = generate_user(birthdate=date(year=2000, month=1, day=1))
222 with real_admin_session(super_token) as api:
223 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
224 assert parse_date(res.birthdate) == date(year=2000, month=1, day=1)
226 res = api.ChangeUserBirthdate(
227 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-05-25")
228 )
230 assert res.user_id == normal_user.id
231 assert res.username == normal_user.username
232 assert res.email == normal_user.email
233 assert res.birthdate == "1990-05-25"
234 assert res.gender == normal_user.gender
235 assert not res.banned
236 assert not res.deleted
238 email = email_collector.pop_for_recipient(normal_user.email, last=True)
239 assert email.subject == "[TEST] Your date of birth was changed"
240 assert email.recipient == normal_user.email
241 assert "1990" in email.plain
242 assert "1990" in email.html
244 push = push_collector.pop_for_user(normal_user.id, last=True)
245 assert push.content.title == "Birthdate changed"
246 assert push.content.body == "An admin changed your date of birth to May 25, 1990."
249def test_BanUser(db):
250 super_user, super_token = generate_user(is_superuser=True)
251 normal_user, _ = generate_user()
252 admin_note = "A good reason"
254 with real_admin_session(super_token) as api:
255 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note=admin_note))
256 assert res.user_id == normal_user.id
257 assert res.username == normal_user.username
258 assert res.email == normal_user.email
259 assert res.gender == normal_user.gender
260 assert parse_date(res.birthdate) == normal_user.birthdate
261 assert res.banned
262 assert not res.deleted
263 assert len(res.admin_actions) == 1
264 assert res.admin_actions[0].action_type == "ban"
265 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
266 assert res.admin_actions[0].note == admin_note
267 assert res.admin_actions[0].admin_user_id == super_user.id
268 assert res.admin_actions[0].admin_username == super_user.username
271def test_UnbanUser(db):
272 super_user, super_token = generate_user(is_superuser=True)
273 normal_user, _ = generate_user()
274 admin_note = "A good reason"
276 with real_admin_session(super_token) as api:
277 res = api.UnbanUser(admin_pb2.UnbanUserReq(user=normal_user.username, admin_note=admin_note))
278 assert res.user_id == normal_user.id
279 assert res.username == normal_user.username
280 assert res.email == normal_user.email
281 assert res.gender == normal_user.gender
282 assert parse_date(res.birthdate) == normal_user.birthdate
283 assert not res.banned
284 assert not res.deleted
285 assert len(res.admin_actions) == 1
286 assert res.admin_actions[0].action_type == "unban"
287 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
290def test_ShadowUser(db):
291 super_user, super_token = generate_user(is_superuser=True)
292 surfer, surfer_token = generate_user()
293 host, _ = generate_user()
294 admin_note = "Spammer"
296 # Create a host request from `surfer` and approve its moderation state to VISIBLE so we can verify the cascade
297 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
298 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
299 with requests_session(surfer_token) as api:
300 host_request_id = api.CreateHostRequest(
301 requests_pb2.CreateHostRequestReq(
302 host_user_id=host.id,
303 from_date=today_plus_2,
304 to_date=today_plus_3,
305 text=valid_request_text(),
306 )
307 ).host_request_id
308 with session_scope() as session:
309 state = session.execute(
310 select(ModerationState)
311 .where(ModerationState.object_type == ModerationObjectType.host_request)
312 .where(ModerationState.object_id == host_request_id)
313 ).scalar_one()
314 state.visibility = ModerationVisibility.visible
316 with real_admin_session(super_token) as api:
317 res = api.ShadowUser(admin_pb2.ShadowUserReq(user=surfer.username, admin_note=admin_note))
318 assert res.user_id == surfer.id
319 assert res.shadowed
320 assert not res.banned
321 assert not res.deleted
322 assert len(res.admin_actions) == 1
323 assert res.admin_actions[0].action_type == "shadow"
324 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
325 assert res.admin_actions[0].note == admin_note
327 # The previously-visible host request is now shadowed
328 with session_scope() as session:
329 state = session.execute(
330 select(ModerationState)
331 .where(ModerationState.object_type == ModerationObjectType.host_request)
332 .where(ModerationState.object_id == host_request_id)
333 ).scalar_one()
334 assert state.visibility == ModerationVisibility.shadowed
337def test_UnshadowUser(db):
338 super_user, super_token = generate_user(is_superuser=True)
339 surfer, surfer_token = generate_user()
340 host, _ = generate_user()
342 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
343 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
344 with requests_session(surfer_token) as api:
345 shadow_cascade_request_id = api.CreateHostRequest(
346 requests_pb2.CreateHostRequestReq(
347 host_user_id=host.id,
348 from_date=today_plus_2,
349 to_date=today_plus_3,
350 text=valid_request_text(),
351 )
352 ).host_request_id
353 backdate_conversations()
354 admin_hidden_request_id = api.CreateHostRequest(
355 requests_pb2.CreateHostRequestReq(
356 host_user_id=host.id,
357 from_date=today_plus_2,
358 to_date=today_plus_3,
359 text=valid_request_text(),
360 )
361 ).host_request_id
363 with session_scope() as session:
364 session.execute(select(User).where(User.id == surfer.id)).scalar_one().shadowed_at = now()
365 session.execute(
366 select(ModerationState)
367 .where(ModerationState.object_type == ModerationObjectType.host_request)
368 .where(ModerationState.object_id == shadow_cascade_request_id)
369 ).scalar_one().visibility = ModerationVisibility.shadowed
370 session.execute(
371 select(ModerationState)
372 .where(ModerationState.object_type == ModerationObjectType.host_request)
373 .where(ModerationState.object_id == admin_hidden_request_id)
374 ).scalar_one().visibility = ModerationVisibility.hidden
376 with real_admin_session(super_token) as api:
377 res = api.UnshadowUser(admin_pb2.UnshadowUserReq(user=surfer.username, admin_note="rehabilitated"))
378 assert not res.shadowed
379 assert len(res.admin_actions) == 1
380 assert res.admin_actions[0].action_type == "unshadow"
381 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
383 with session_scope() as session:
384 assert (
385 session.execute(
386 select(ModerationState)
387 .where(ModerationState.object_type == ModerationObjectType.host_request)
388 .where(ModerationState.object_id == shadow_cascade_request_id)
389 )
390 .scalar_one()
391 .visibility
392 == ModerationVisibility.visible
393 )
394 assert (
395 session.execute(
396 select(ModerationState)
397 .where(ModerationState.object_type == ModerationObjectType.host_request)
398 .where(ModerationState.object_id == admin_hidden_request_id)
399 )
400 .scalar_one()
401 .visibility
402 == ModerationVisibility.hidden
403 )
406def test_ShadowUser_blank_note(db):
407 super_user, super_token = generate_user(is_superuser=True)
408 normal_user, _ = generate_user()
410 with real_admin_session(super_token) as api:
411 with pytest.raises(grpc.RpcError) as e:
412 api.ShadowUser(admin_pb2.ShadowUserReq(user=normal_user.username, admin_note=" \t "))
413 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
416def test_AddAdminNote(db):
417 super_user, super_token = generate_user(is_superuser=True)
418 normal_user, _ = generate_user()
419 admin_note1 = "User reported strange behavior"
420 admin_note2 = "Insert private information here"
422 with real_admin_session(super_token) as api:
423 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note1))
424 assert res.user_id == normal_user.id
425 assert res.username == normal_user.username
426 assert res.email == normal_user.email
427 assert res.gender == normal_user.gender
428 assert parse_date(res.birthdate) == normal_user.birthdate
429 assert not res.banned
430 assert not res.deleted
431 assert len(res.admin_actions) == 1
432 assert res.admin_actions[0].action_type == "note"
433 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
434 assert res.admin_actions[0].note == admin_note1
436 with real_admin_session(super_token) as api:
437 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note2))
438 assert len(res.admin_actions) == 2
439 assert res.admin_actions[0].note == admin_note1
440 assert res.admin_actions[1].note == admin_note2
443def test_AddAdminNote_blank(db):
444 super_user, super_token = generate_user(is_superuser=True)
445 normal_user, _ = generate_user()
446 empty_admin_note = " \t \n "
448 with real_admin_session(super_token) as api:
449 with pytest.raises(grpc.RpcError) as e:
450 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=empty_admin_note))
451 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
452 assert e.value.details() == "Provide exactly one of admin_note or data."
455def test_AddAdminNote_data(db):
456 super_user, super_token = generate_user(is_superuser=True)
457 normal_user, _ = generate_user()
458 payload = '{"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}'
460 with real_admin_session(super_token) as api:
461 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data=payload))
462 assert len(res.admin_actions) == 1
463 assert res.admin_actions[0].action_type == "note"
464 assert res.admin_actions[0].note == ""
465 assert json.loads(res.admin_actions[0].data) == {"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}
468def test_AddAdminNote_both_note_and_data(db):
469 super_user, super_token = generate_user(is_superuser=True)
470 normal_user, _ = generate_user()
472 with real_admin_session(super_token) as api:
473 with pytest.raises(grpc.RpcError) as e:
474 api.AddAdminNote(
475 admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="note text", data='{"x": 1}')
476 )
477 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
478 assert e.value.details() == "Provide exactly one of admin_note or data."
481def test_AddAdminNote_neither(db):
482 super_user, super_token = generate_user(is_superuser=True)
483 normal_user, _ = generate_user()
485 with real_admin_session(super_token) as api:
486 with pytest.raises(grpc.RpcError) as e:
487 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username))
488 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
489 assert e.value.details() == "Provide exactly one of admin_note or data."
492def test_AddAdminNote_invalid_json(db):
493 super_user, super_token = generate_user(is_superuser=True)
494 normal_user, _ = generate_user()
496 with real_admin_session(super_token) as api:
497 with pytest.raises(grpc.RpcError) as e:
498 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data="{not valid json"))
499 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
500 assert e.value.details() == "The admin note data must be valid JSON."
503def test_admin_content_reports(db):
504 super_user, super_token = generate_user(is_superuser=True)
505 normal_user, token = generate_user()
506 bad_user1, _ = generate_user()
507 bad_user2, _ = generate_user()
509 with reporting_session(token) as api:
510 api.Report(
511 reporting_pb2.ReportReq(
512 reason="spam",
513 description="r1",
514 content_ref="comment/123",
515 author_user=bad_user1.username,
516 user_agent="n/a",
517 page="https://couchers.org/comment/123",
518 )
519 )
520 api.Report(
521 reporting_pb2.ReportReq(
522 reason="spam",
523 description="r2",
524 content_ref="comment/124",
525 author_user=bad_user2.username,
526 user_agent="n/a",
527 page="https://couchers.org/comment/124",
528 )
529 )
530 api.Report(
531 reporting_pb2.ReportReq(
532 reason="something else",
533 description="r3",
534 content_ref="page/321",
535 author_user=bad_user1.username,
536 user_agent="n/a",
537 page="https://couchers.org/page/321",
538 )
539 )
541 with session_scope() as session:
542 id_by_description: dict[str, int] = dict(
543 session.execute(select(ContentReport.description, ContentReport.id)).all() # type: ignore[arg-type]
544 )
546 with real_admin_session(super_token) as api:
547 with pytest.raises(grpc.RpcError) as e:
548 api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=-1))
549 assert e.value.code() == grpc.StatusCode.NOT_FOUND
550 assert e.value.details() == "Content report not found."
552 res = api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=id_by_description["r2"]))
553 rep = res.content_report
554 assert rep.content_report_id == id_by_description["r2"]
555 assert rep.reporting_user_id == normal_user.id
556 assert rep.author_user_id == bad_user2.id
557 assert rep.reason == "spam"
558 assert rep.description == "r2"
559 assert rep.content_ref == "comment/124"
560 assert rep.user_agent == "n/a"
561 assert rep.page == "https://couchers.org/comment/124"
563 res = api.GetContentReportsForAuthor(admin_pb2.GetContentReportsForAuthorReq(user=bad_user1.username))
564 assert res.content_reports[0].content_report_id == id_by_description["r3"]
565 assert res.content_reports[1].content_report_id == id_by_description["r1"]
568def test_DeleteUser(db):
569 super_user, super_token = generate_user(is_superuser=True)
570 normal_user, normal_token = generate_user()
572 with real_admin_session(super_token) as api:
573 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
574 assert res.user_id == normal_user.id
575 assert res.username == normal_user.username
576 assert res.email == normal_user.email
577 assert res.gender == normal_user.gender
578 assert parse_date(res.birthdate) == normal_user.birthdate
579 assert not res.banned
580 assert res.deleted
582 with real_admin_session(super_token) as api:
583 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
584 assert res.user_id == normal_user.id
585 assert res.username == normal_user.username
586 assert res.email == normal_user.email
587 assert res.gender == normal_user.gender
588 assert parse_date(res.birthdate) == normal_user.birthdate
589 assert not res.banned
590 assert not res.deleted
593def test_RecoverDeletedUser_after_user_initiated_deletion(db, push_collector: PushCollector):
594 """
595 When a user deletes their account through the normal flow (ConfirmDeleteAccount),
596 undelete_token and undelete_until are set. The admin RecoverDeletedUser must clear
597 these fields to satisfy the undelete_nullity database constraint.
598 """
599 super_user, super_token = generate_user(is_superuser=True)
600 normal_user, normal_token = generate_user()
601 user_id = normal_user.id
603 # User initiates account deletion
604 with account_session(normal_token) as account:
605 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
607 # Get the deletion confirmation token
608 with session_scope() as session:
609 deletion_token = session.execute(select(AccountDeletionToken)).scalar_one().token
611 # User confirms account deletion (this sets undelete_token and undelete_until)
612 with auth_api_session() as (auth_api, metadata_interceptor):
613 auth_api.ConfirmDeleteAccount(auth_pb2.ConfirmDeleteAccountReq(token=deletion_token))
615 # Verify the user is deleted and has undelete fields set
616 with session_scope() as session:
617 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
618 assert user.deleted_at is not None
619 assert user.undelete_token is not None
620 assert user.undelete_until is not None
622 # Admin recovers the user
623 with real_admin_session(super_token) as api:
624 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
625 assert res.user_id == user_id
626 assert not res.deleted
628 # Verify undelete fields are cleared
629 with session_scope() as session:
630 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
631 assert user.deleted_at is None
632 assert user.undelete_token is None
633 assert user.undelete_until is None
636def test_CreateApiKey(db, email_collector: EmailCollector, push_collector: PushCollector):
637 with session_scope() as session:
638 super_user, super_token = generate_user(is_superuser=True)
639 normal_user, normal_token = generate_user()
641 assert (
642 session.execute(
643 select(func.count())
644 .select_from(UserSession)
645 .where(UserSession.is_api_key == True)
646 .where(UserSession.user_id == normal_user.id)
647 ).scalar_one()
648 == 0
649 )
651 with real_admin_session(super_token) as api:
652 res = api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=normal_user.username))
654 email = email_collector.pop_for_recipient(normal_user.email, last=True)
655 assert email.subject == "[TEST] Your API key for Couchers.org"
657 with session_scope() as session:
658 token = session.execute(
659 select(UserSession.token)
660 .where(UserSession.is_valid)
661 .where(UserSession.is_api_key == True)
662 .where(UserSession.user_id == normal_user.id)
663 ).scalar_one()
665 assert token in email.plain
666 assert token in email.html
668 assert email.recipient == normal_user.email
669 assert "api key" in email.subject.lower()
670 unique_string = "We've issued you with the following API key:"
671 assert unique_string in email.plain
672 assert unique_string in email.html
673 assert "support@couchers.org" in email.plain
674 assert "support@couchers.org" in email.html
676 push = push_collector.pop_for_user(normal_user.id, last=True)
677 assert push.content.title == "API key created"
678 assert push.content.body == "Details were sent to you via email."
681def test_GetChats(db):
682 super_user, super_token = generate_user(is_superuser=True)
683 normal_user, normal_token = generate_user()
685 with real_admin_session(super_token) as api:
686 res = api.GetChats(admin_pb2.GetChatsReq(user=normal_user.username))
687 assert res.user.user_id == normal_user.id
688 assert res.user.username == normal_user.username
689 assert res.user.name == normal_user.name
690 # New user should have no chats
691 assert len(res.host_requests) == 0
692 assert len(res.group_chats) == 0
695def test_badges(db, email_collector: EmailCollector, push_collector: PushCollector):
696 super_user, super_token = generate_user(is_superuser=True)
697 normal_user, normal_token = generate_user()
699 with real_admin_session(super_token) as api:
700 # can add a badge
701 assert "swagster" not in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
702 res = api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
703 assert "swagster" in res.badges
705 # badge emails are disabled by default
706 assert email_collector.count_for_recipient(normal_user.email) == 0
708 push = push_collector.pop_for_user(normal_user.id, last=True)
709 assert push.content.title == "New profile badge: Swagster"
710 assert push.content.body == "The Swagster badge was added to your profile."
712 # can't add/edit special tags
713 with pytest.raises(grpc.RpcError) as e:
714 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="founder"))
715 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
716 assert e.value.details() == "Admins cannot edit that badge."
718 # double add badge
719 with pytest.raises(grpc.RpcError) as e:
720 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
721 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
722 assert e.value.details() == "The user already has that badge."
724 # can remove badge
725 assert "swagster" in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
726 res = api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
727 assert "swagster" not in res.badges
729 # badge emails are disabled by default
730 assert email_collector.count_for_recipient(normal_user.email) == 0
732 push = push_collector.pop_for_user(normal_user.id, last=True)
733 assert push.content.title == "Profile badge removed"
734 assert push.content.body == "The Swagster badge was removed from your profile."
736 # not found on user
737 with pytest.raises(grpc.RpcError) as e:
738 api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
739 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
740 assert e.value.details() == "The user does not have that badge."
742 # not found in general
743 with pytest.raises(grpc.RpcError) as e:
744 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="nonexistentbadge"))
745 assert e.value.code() == grpc.StatusCode.NOT_FOUND
746 assert e.value.details() == "Badge not found."
749def test_DeleteEvent(db):
750 super_user, super_token = generate_user(is_superuser=True)
751 normal_user, normal_token = generate_user()
753 with session_scope() as session:
754 create_community(session, 0, 2, "Community", [normal_user], [], None)
756 start_time = now() + timedelta(hours=2)
757 end_time = start_time + timedelta(hours=3)
758 with events_session(normal_token) as api:
759 res = api.CreateEvent(
760 events_pb2.CreateEventReq(
761 title="Dummy Title",
762 content="Dummy content.",
763 photo_key=None,
764 location=events_pb2.EventLocation(
765 address="Near Null Island",
766 lat=0.1,
767 lng=0.2,
768 ),
769 start_datetime_iso8601_local=datetime_to_iso8601_local(start_time),
770 end_datetime_iso8601_local=datetime_to_iso8601_local(end_time),
771 )
772 )
773 event_id = res.event_id
774 assert not res.is_deleted
776 with session_scope() as session:
777 with real_admin_session(super_token) as api:
778 api.DeleteEvent(
779 admin_pb2.DeleteEventReq(
780 event_id=event_id,
781 )
782 )
783 occurrence = session.get_one(EventOccurrence, ident=event_id)
784 assert occurrence.is_deleted
787def test_ListUserIds(db):
788 super_user, super_token = generate_user(is_superuser=True)
789 normal_user, normal_token = generate_user()
791 with real_admin_session(super_token) as api:
792 res = api.ListUserIds(
793 admin_pb2.ListUserIdsReq(
794 start_time=Timestamp_from_datetime(datetime(2000, 1, 1, tzinfo=UTC)),
795 end_time=Timestamp_from_datetime(now()),
796 )
797 )
798 assert len(res.user_ids) == 2
799 assert sorted(res.user_ids) == sorted([super_user.id, normal_user.id])
801 with real_admin_session(super_token) as api:
802 res = api.ListUserIds(
803 admin_pb2.ListUserIdsReq(start_time=Timestamp_from_datetime(now()), end_time=Timestamp_from_datetime(now()))
804 )
805 assert res.user_ids == []
808def test_EditReferenceText(db):
809 super_user, super_token = generate_user(is_superuser=True)
810 test_new_text = "New Text"
812 user1, user1_token = generate_user()
813 user2, user2_token = generate_user()
814 make_friends(user1, user2)
816 with session_scope() as session:
817 with references_session(user1_token) as api:
818 reference = api.WriteFriendReference(
819 references_pb2.WriteFriendReferenceReq(
820 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
821 )
822 )
824 with real_admin_session(super_token) as admin_api:
825 admin_api.EditReferenceText(
826 admin_pb2.EditReferenceTextReq(reference_id=reference.reference_id, new_text=test_new_text)
827 )
829 session.expire_all()
831 modified_reference = session.execute(
832 select(Reference).where(Reference.id == reference.reference_id)
833 ).scalar_one()
834 assert modified_reference.text == test_new_text
837def test_DeleteReference_deprecated(db):
838 """DeleteReference is deprecated; admins should hide via UMS instead."""
839 super_user, super_token = generate_user(is_superuser=True)
841 user1, user1_token = generate_user()
842 user2, user2_token = generate_user()
843 make_friends(user1, user2)
845 with references_session(user1_token) as api:
846 reference = api.WriteFriendReference(
847 references_pb2.WriteFriendReferenceReq(
848 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
849 )
850 )
852 with real_admin_session(super_token) as admin_api:
853 with pytest.raises(grpc.RpcError) as e:
854 admin_api.DeleteReference(admin_pb2.DeleteReferenceReq(reference_id=reference.reference_id))
855 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
858def test_GetUserReferences(db):
859 super_user, super_token = generate_user(is_superuser=True)
861 user1, user1_token = generate_user()
862 user2, user2_token = generate_user()
863 user3, user3_token = generate_user()
864 make_friends(user1, user2)
865 make_friends(user1, user3)
866 make_friends(user2, user3)
868 # user1 writes reference about user2
869 with references_session(user1_token) as api:
870 ref1 = api.WriteFriendReference(
871 references_pb2.WriteFriendReferenceReq(
872 to_user_id=user2.id,
873 text="Reference from user1 to user2",
874 private_text="",
875 was_appropriate=True,
876 rating=1,
877 )
878 )
880 # user2 writes reference about user1
881 with references_session(user2_token) as api:
882 ref2 = api.WriteFriendReference(
883 references_pb2.WriteFriendReferenceReq(
884 to_user_id=user1.id,
885 text="Reference from user2 to user1",
886 private_text="Private note",
887 was_appropriate=True,
888 rating=0.8,
889 )
890 )
892 # user3 writes reference about user1
893 with references_session(user3_token) as api:
894 ref3 = api.WriteFriendReference(
895 references_pb2.WriteFriendReferenceReq(
896 to_user_id=user1.id,
897 text="Reference from user3 to user1",
898 private_text="",
899 was_appropriate=False,
900 rating=0.5,
901 )
902 )
904 # Test GetUserReferences for user1 (admin view shows everything regardless of UMS state).
905 with real_admin_session(super_token) as admin_api:
906 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
908 # user1 wrote 1 reference
909 assert len(res.references_from) == 1
910 assert res.references_from[0].reference_id == ref1.reference_id
911 assert res.references_from[0].from_user_id == user1.id
912 assert res.references_from[0].to_user_id == user2.id
913 assert res.references_from[0].text == "Reference from user1 to user2"
915 # user1 received 2 references
916 assert len(res.references_to) == 2
917 # Ordered by id descending, so ref3 comes first
918 assert res.references_to[0].reference_id == ref3.reference_id
919 assert res.references_to[0].was_appropriate is False
921 assert res.references_to[1].reference_id == ref2.reference_id
922 assert res.references_to[1].private_text == "Private note"
923 assert res.references_to[1].rating == 0.8
926def test_GetUserReferences_not_found(db):
927 super_user, super_token = generate_user(is_superuser=True)
929 with real_admin_session(super_token) as admin_api:
930 with pytest.raises(grpc.RpcError) as e:
931 admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user="nonexistent"))
932 assert e.value.code() == grpc.StatusCode.NOT_FOUND
935def test_GetFriendRequests(db):
936 super_user, super_token = generate_user(is_superuser=True)
938 user1, _ = generate_user()
939 user2, _ = generate_user()
940 user3, _ = generate_user()
941 user4, _ = generate_user()
943 # Create a mix of friend requests directly so we control the state
944 def _add_friend_request(from_user_id, to_user_id, status, visibility, time_responded=None):
945 with session_scope() as session:
946 mod_state = ModerationState(
947 object_type=ModerationObjectType.friend_request,
948 object_id=0,
949 visibility=visibility,
950 )
951 session.add(mod_state)
952 session.flush()
953 rel = FriendRelationship(
954 from_user_id=from_user_id,
955 to_user_id=to_user_id,
956 status=status,
957 moderation_state_id=mod_state.id,
958 time_responded=time_responded,
959 )
960 session.add(rel)
961 session.flush()
962 mod_state.object_id = rel.id
964 # user1 -> user2: pending, shadowed
965 _add_friend_request(user1.id, user2.id, FriendStatus.pending, ModerationVisibility.shadowed)
966 # user1 -> user3: accepted, visible
967 _add_friend_request(user1.id, user3.id, FriendStatus.accepted, ModerationVisibility.visible, time_responded=now())
968 # user4 -> user1: rejected, visible
969 _add_friend_request(user4.id, user1.id, FriendStatus.rejected, ModerationVisibility.visible, time_responded=now())
971 with real_admin_session(super_token) as admin_api:
972 res = admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user=user1.username))
974 # user1 sent two: to user2 (pending) and to user3 (accepted), ordered by id desc
975 assert len(res.sent) == 2
976 assert res.sent[0].from_user.user_id == user1.id
977 assert res.sent[0].to_user.user_id == user3.id
978 assert res.sent[0].status == "accepted"
979 assert res.sent[0].HasField("time_responded")
980 assert res.sent[0].moderation_visibility == "visible"
982 assert res.sent[1].from_user.user_id == user1.id
983 assert res.sent[1].to_user.user_id == user2.id
984 assert res.sent[1].status == "pending"
985 assert not res.sent[1].HasField("time_responded")
986 assert res.sent[1].moderation_visibility == "shadowed"
988 # user1 received one: from user4 (rejected)
989 assert len(res.received) == 1
990 assert res.received[0].from_user.user_id == user4.id
991 assert res.received[0].to_user.user_id == user1.id
992 assert res.received[0].status == "rejected"
995def test_GetFriendRequests_not_found(db):
996 super_user, super_token = generate_user(is_superuser=True)
998 with real_admin_session(super_token) as admin_api:
999 with pytest.raises(grpc.RpcError) as e:
1000 admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user="nonexistent"))
1001 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1004def test_AddUsersToModerationUserList(db):
1005 super_user, super_token = generate_user(is_superuser=True)
1006 user1, _ = generate_user()
1007 user2, _ = generate_user()
1008 user3, _ = generate_user()
1009 user4, _ = generate_user()
1010 user5, _ = generate_user()
1011 moderation_list_id = add_users_to_new_moderation_list([user1])
1013 with session_scope() as session:
1014 with real_admin_session(super_token) as api:
1015 # Test adding users to a non-existent moderation list (should raise an error)
1016 with pytest.raises(grpc.RpcError) as e:
1017 api.AddUsersToModerationUserList(
1018 admin_pb2.AddUsersToModerationUserListReq(users=[user2.username], moderation_list_id=999),
1019 )
1020 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1021 assert "Moderation user list not found." == e.value.details()
1023 # Test with non-existent user (should raise an error)
1024 with pytest.raises(grpc.RpcError) as e:
1025 api.AddUsersToModerationUserList(
1026 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, "nonexistent"]),
1027 )
1028 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1029 assert "Couldn't find that user." == e.value.details()
1031 # Test successful creation of new moderation list (no moderation_list_id provided)
1032 res = api.AddUsersToModerationUserList(
1033 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, user2.username, user3.username]),
1034 )
1035 assert res.moderation_list_id > 0
1036 with session_scope() as session:
1037 moderation_user_list = session.get(ModerationUserList, res.moderation_list_id)
1038 assert moderation_user_list is not None
1039 assert len(moderation_user_list.users) == 3
1040 assert {user1.id, user2.id, user3.id}.issubset({user.id for user in moderation_user_list.users})
1042 # Test list endpoint returns same moderation list with same members not repeated
1043 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1044 assert len(listRes.moderation_lists) == 1
1045 assert listRes.moderation_lists[0].moderation_list_id == res.moderation_list_id
1046 assert len(listRes.moderation_lists[0].members) == 3
1047 assert {user1.id, user2.id, user3.id}.issubset({m.user_id for m in listRes.moderation_lists[0].members})
1049 # Test user can be in multiple moderation lists
1050 listRes3 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1051 assert len(listRes3.moderation_lists) == 2
1053 # Test adding users to an existing moderation list
1054 res2 = api.AddUsersToModerationUserList(
1055 admin_pb2.AddUsersToModerationUserListReq(
1056 users=[user4.username, user5.username], moderation_list_id=moderation_list_id
1057 ),
1058 )
1059 assert res2.moderation_list_id == moderation_list_id
1060 with session_scope() as session:
1061 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1062 assert len(moderation_user_list.users) == 3
1063 assert {user1.id, user4.id, user5.id}.issubset({user.id for user in moderation_user_list.users})
1065 # Test list user moderation lists endpoint returns the right moderation list
1066 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user5.username))
1067 assert len(listRes2.moderation_lists) == 1
1068 assert listRes2.moderation_lists[0].moderation_list_id == moderation_list_id
1069 assert len(listRes2.moderation_lists[0].members) == 3
1070 assert {user1.id, user4.id, user5.id}.issubset({m.user_id for m in listRes2.moderation_lists[0].members})
1073def test_RemoveUserFromModerationUserList(db):
1074 super_user, super_token = generate_user(is_superuser=True)
1075 user1, _ = generate_user()
1076 user2, _ = generate_user()
1077 user3, _ = generate_user()
1078 moderation_list_id = add_users_to_new_moderation_list([user1, user2])
1080 with real_admin_session(super_token) as api:
1081 # Test with non-existent user (should raise error)
1082 with pytest.raises(grpc.RpcError) as e:
1083 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user="nonexistent"))
1084 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1085 assert "Couldn't find that user." == e.value.details()
1087 # Test without providing moderation list id (should raise error)
1088 with pytest.raises(grpc.RpcError) as e:
1089 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username))
1090 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1091 assert "Missing moderation user list id." == e.value.details()
1093 # Test removing user that's not in the provided moderation list (should raise error)
1094 with pytest.raises(grpc.RpcError) as e:
1095 api.RemoveUserFromModerationUserList(
1096 admin_pb2.RemoveUserFromModerationUserListReq(
1097 user=user3.username, moderation_list_id=moderation_list_id
1098 )
1099 )
1100 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1101 assert "User is not in the moderation user list." == e.value.details()
1103 # Test successful removal
1104 api.RemoveUserFromModerationUserList(
1105 admin_pb2.RemoveUserFromModerationUserListReq(user=user1.username, moderation_list_id=moderation_list_id)
1106 )
1107 with session_scope() as session:
1108 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1109 assert user1.id not in {user.id for user in moderation_user_list.users}
1110 assert user2.id in {user.id for user in moderation_user_list.users}
1112 # Test list user moderation lists endpoint returns right number of moderation lists
1113 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1114 assert len(listRes.moderation_lists) == 0
1115 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1116 assert len(listRes2.moderation_lists) == 1
1118 # Test removing all users from moderation list should also delete the moderation list
1119 api.RemoveUserFromModerationUserList(
1120 admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username, moderation_list_id=moderation_list_id)
1121 )
1122 with session_scope() as session:
1123 assert session.get(ModerationUserList, moderation_list_id) is None
1126def test_admin_delete_account_url(db, email_collector: EmailCollector, push_collector: PushCollector):
1127 super_user, super_token = generate_user(is_superuser=True)
1129 user, token = generate_user()
1130 user_id = user.id
1132 with real_admin_session(super_token) as admin_api:
1133 url = admin_api.CreateAccountDeletionLink(
1134 admin_pb2.CreateAccountDeletionLinkReq(user=user.username)
1135 ).account_deletion_confirm_url
1137 assert push_collector.count_for_user(user_id) == 0
1139 with session_scope() as session:
1140 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
1141 token = token_o.token
1142 assert token_o.user.id == user_id
1143 assert url == f"http://localhost:3000/delete-account?token={token}"
1145 with auth_api_session() as (auth_api, metadata_interceptor):
1146 auth_api.ConfirmDeleteAccount(
1147 auth_pb2.ConfirmDeleteAccountReq(
1148 token=token,
1149 )
1150 )
1152 push = push_collector.pop_for_user(user_id, last=True)
1153 assert push.content.title == "Account deleted"
1154 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
1155 email_collector.pop_for_recipient(user.email, last=True)
1158def test_AccessStats(db):
1159 super_user, super_token = generate_user(is_superuser=True)
1160 normal_user, normal_token = generate_user()
1162 # Insert UserActivity rows: a couple inside the default 90-day window, one well
1163 # outside it, and one with NULL ip_address / user_agent. The INET column is
1164 # returned by psycopg3 as an IPv4Address/IPv6Address object, which used to
1165 # crash the proto string assignment.
1166 in_window_1 = now() - timedelta(days=1)
1167 in_window_2 = now() - timedelta(days=10)
1168 out_of_window = now() - timedelta(days=200)
1169 with session_scope() as session:
1170 session.add(
1171 UserActivity(
1172 user_id=normal_user.id, period=in_window_1, ip_address="1.2.3.4", user_agent="ua-a", api_calls=5
1173 )
1174 )
1175 session.add(
1176 UserActivity(
1177 user_id=normal_user.id, period=in_window_2, ip_address="2001:db8::1", user_agent="ua-b", api_calls=3
1178 )
1179 )
1180 session.add(
1181 UserActivity(
1182 user_id=normal_user.id, period=out_of_window, ip_address="9.9.9.9", user_agent="ua-old", api_calls=99
1183 )
1184 )
1185 session.add(UserActivity(user_id=normal_user.id, period=in_window_1, api_calls=1))
1187 with real_admin_session(super_token) as api:
1188 res = api.AccessStats(admin_pb2.AccessStatsReq(user=normal_user.username))
1190 by_ip = {s.ip_address: s for s in res.stats}
1191 assert "1.2.3.4" in by_ip
1192 assert by_ip["1.2.3.4"].api_call_count == 5
1193 assert by_ip["1.2.3.4"].user_agent == "ua-a"
1194 assert "2001:db8::1" in by_ip
1195 assert by_ip["2001:db8::1"].api_call_count == 3
1196 # NULL ip_address row produces an empty-string ip_address in the proto
1197 assert "" in by_ip
1198 assert by_ip[""].api_call_count == 1
1199 # out-of-window row is excluded by the 90-day default
1200 assert "9.9.9.9" not in by_ip
1202 # explicit end_time should bound the upper end of the window (regression: was >=)
1203 with real_admin_session(super_token) as api:
1204 res = api.AccessStats(
1205 admin_pb2.AccessStatsReq(
1206 user=normal_user.username,
1207 start_time=Timestamp_from_datetime(now() - timedelta(days=5)),
1208 end_time=Timestamp_from_datetime(now()),
1209 )
1210 )
1211 ips = {s.ip_address for s in res.stats}
1212 assert ips == {"1.2.3.4", ""}
1215def test_SetLastDonated(db):
1216 super_user, super_token = generate_user(is_superuser=True)
1217 normal_user, normal_token = generate_user(last_donated=None)
1219 with real_admin_session(super_token) as api:
1220 # user starts with no last_donated
1221 with session_scope() as session:
1222 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1223 assert user.last_donated is None
1225 # can set last_donated
1226 donation_time = now() - timedelta(days=30)
1227 res = api.SetLastDonated(
1228 admin_pb2.SetLastDonatedReq(
1229 user=normal_user.username,
1230 last_donated=Timestamp_from_datetime(donation_time),
1231 )
1232 )
1234 with session_scope() as session:
1235 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1236 assert user.last_donated is not None
1237 # check timestamp is close (within a second)
1238 assert abs((user.last_donated - donation_time).total_seconds()) < 1
1240 # can clear last_donated by not setting the field
1241 res = api.SetLastDonated(admin_pb2.SetLastDonatedReq(user=normal_user.username))
1243 with session_scope() as session:
1244 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1245 assert user.last_donated is None
1247 # user not found
1248 with pytest.raises(grpc.RpcError) as e:
1249 api.SetLastDonated(admin_pb2.SetLastDonatedReq(user="nonexistent"))
1250 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1251 assert e.value.details() == "Couldn't find that user."
1254def test_admin_actions_level(db):
1255 super_user, super_token = generate_user(is_superuser=True)
1256 normal_user, _ = generate_user()
1258 with real_admin_session(super_token) as api:
1259 # Default level is NORMAL
1260 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="normal note"))
1261 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
1263 # Explicitly set to DEBUG
1264 res = api.AddAdminNote(
1265 admin_pb2.AddAdminNoteReq(
1266 user=normal_user.username,
1267 admin_note="debug note",
1268 level=admin_pb2.ADMIN_ACTION_LEVEL_DEBUG,
1269 )
1270 )
1271 assert len(res.admin_actions) == 2
1272 assert res.admin_actions[1].level == admin_pb2.ADMIN_ACTION_LEVEL_DEBUG
1274 # Explicitly set to HIGH
1275 res = api.AddAdminNote(
1276 admin_pb2.AddAdminNoteReq(
1277 user=normal_user.username,
1278 admin_note="high note",
1279 level=admin_pb2.ADMIN_ACTION_LEVEL_HIGH,
1280 )
1281 )
1282 assert len(res.admin_actions) == 3
1283 assert res.admin_actions[2].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
1285 # Explicitly set to TRACE
1286 res = api.AddAdminNote(
1287 admin_pb2.AddAdminNoteReq(
1288 user=normal_user.username,
1289 admin_note="trace note",
1290 level=admin_pb2.ADMIN_ACTION_LEVEL_TRACE,
1291 )
1292 )
1293 assert len(res.admin_actions) == 4
1294 assert res.admin_actions[3].level == admin_pb2.ADMIN_ACTION_LEVEL_TRACE
1297def test_admin_actions_on_mutations(db, push_collector: PushCollector):
1298 super_user, super_token = generate_user(is_superuser=True)
1299 normal_user, _ = generate_user()
1301 original_gender = normal_user.gender
1302 original_birthdate = normal_user.birthdate
1304 with real_admin_session(super_token) as api:
1305 # ChangeUserGender
1306 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
1307 assert any(
1308 a.action_type == "change_gender" and a.note == f"Changed from '{original_gender}' to 'Machine'"
1309 for a in res.admin_actions
1310 )
1312 # ChangeUserBirthdate
1313 res = api.ChangeUserBirthdate(
1314 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-01-01")
1315 )
1316 assert any(
1317 a.action_type == "change_birthdate" and a.note == f"Changed from {original_birthdate} to 1990-01-01"
1318 for a in res.admin_actions
1319 )
1321 # SetPassportSexGenderException
1322 res = api.SetPassportSexGenderException(
1323 admin_pb2.SetPassportSexGenderExceptionReq(user=normal_user.username, passport_sex_gender_exception=True)
1324 )
1325 assert any(
1326 a.action_type == "set_passport_sex_gender_exception" and a.note == "Changed from False to True"
1327 for a in res.admin_actions
1328 )
1330 # SendModNote with notify
1331 res = api.SendModNote(
1332 admin_pb2.SendModNoteReq(
1333 user=normal_user.username, content="Please update your profile", internal_id="test1"
1334 )
1335 )
1336 assert any(
1337 a.action_type == "send_mod_note" and a.note == "Notify user: Yes\n\nPlease update your profile"
1338 for a in res.admin_actions
1339 )
1341 # SendModNote with do_not_notify
1342 res = api.SendModNote(
1343 admin_pb2.SendModNoteReq(
1344 user=normal_user.username,
1345 content="Silent note",
1346 internal_id="test2",
1347 do_not_notify=True,
1348 )
1349 )
1350 assert any(
1351 a.action_type == "send_mod_note" and a.note == "Notify user: No\n\nSilent note" for a in res.admin_actions
1352 )
1354 # DeleteUser
1355 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
1356 assert any(a.action_type == "delete_user" for a in res.admin_actions)
1357 assert any(
1358 a.action_type == "delete_user" and a.level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH for a in res.admin_actions
1359 )
1361 # RecoverDeletedUser
1362 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
1363 assert any(a.action_type == "recover_user" for a in res.admin_actions)
1365 # MarkUserNeedsLocationUpdate
1366 res = api.MarkUserNeedsLocationUpdate(admin_pb2.MarkUserNeedsLocationUpdateReq(user=normal_user.username))
1367 assert any(
1368 a.action_type == "mark_needs_location_update" and a.note == "Marked user as needing location update"
1369 for a in res.admin_actions
1370 )
1372 # SetLastDonated
1373 res = api.SetLastDonated(
1374 admin_pb2.SetLastDonatedReq(
1375 user=normal_user.username,
1376 last_donated=Timestamp_from_datetime(now()),
1377 )
1378 )
1379 assert any(a.action_type == "set_last_donated" for a in res.admin_actions)
1382def test_create_admin_tag(db):
1383 super_user, super_token = generate_user(is_superuser=True)
1385 with real_admin_session(super_token) as api:
1386 res = api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1387 assert res.tag == "test-tag"
1388 assert res.admin_tag_id > 0
1391def test_create_admin_tag_duplicate(db):
1392 super_user, super_token = generate_user(is_superuser=True)
1394 with real_admin_session(super_token) as api:
1395 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1396 with pytest.raises(grpc.RpcError) as e:
1397 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1398 assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
1399 assert e.value.details() == "That admin tag already exists."
1402def test_create_admin_tag_empty(db):
1403 super_user, super_token = generate_user(is_superuser=True)
1405 with real_admin_session(super_token) as api:
1406 with pytest.raises(grpc.RpcError) as e:
1407 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=""))
1408 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1409 assert e.value.details() == "The admin tag cannot be empty."
1411 with pytest.raises(grpc.RpcError) as e:
1412 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=" "))
1413 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1414 assert e.value.details() == "The admin tag cannot be empty."
1417def test_list_admin_tags(db):
1418 super_user, super_token = generate_user(is_superuser=True)
1420 with real_admin_session(super_token) as api:
1421 # Empty initially
1422 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1423 assert len(res.tags) == 0
1425 # Add some tags
1426 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="bravo"))
1427 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="alpha"))
1429 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1430 assert len(res.tags) == 2
1431 # Ordered alphabetically
1432 assert res.tags[0].tag == "alpha"
1433 assert res.tags[1].tag == "bravo"
1436def test_add_admin_tag_to_user(db):
1437 super_user, super_token = generate_user(is_superuser=True)
1438 normal_user, _ = generate_user()
1440 with real_admin_session(super_token) as api:
1441 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1443 res = api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1444 assert "vip" in res.admin_tags
1445 assert any(a.action_type == "add_tag" and a.tag == "vip" for a in res.admin_actions)
1448def test_add_admin_tag_to_user_duplicate(db):
1449 super_user, super_token = generate_user(is_superuser=True)
1450 normal_user, _ = generate_user()
1452 with real_admin_session(super_token) as api:
1453 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1454 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1456 with pytest.raises(grpc.RpcError) as e:
1457 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1458 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1459 assert e.value.details() == "The user already has that admin tag."
1462def test_add_admin_tag_to_user_tag_not_found(db):
1463 super_user, super_token = generate_user(is_superuser=True)
1464 normal_user, _ = generate_user()
1466 with real_admin_session(super_token) as api:
1467 with pytest.raises(grpc.RpcError) as e:
1468 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="nonexistent"))
1469 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1470 assert e.value.details() == "Admin tag not found."
1473def test_remove_admin_tag_from_user(db):
1474 super_user, super_token = generate_user(is_superuser=True)
1475 normal_user, _ = generate_user()
1477 with real_admin_session(super_token) as api:
1478 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1479 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1481 res = api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1482 assert "vip" not in res.admin_tags
1483 assert any(a.action_type == "remove_tag" and a.tag == "vip" for a in res.admin_actions)
1486def test_remove_admin_tag_from_user_not_assigned(db):
1487 super_user, super_token = generate_user(is_superuser=True)
1488 normal_user, _ = generate_user()
1490 with real_admin_session(super_token) as api:
1491 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1493 with pytest.raises(grpc.RpcError) as e:
1494 api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1495 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1496 assert e.value.details() == "The user does not have that admin tag."
1499def test_search_users_by_admin_tag(db):
1500 super_user, super_token = generate_user(is_superuser=True)
1501 user1, _ = generate_user()
1502 user2, _ = generate_user()
1503 user3, _ = generate_user()
1505 with real_admin_session(super_token) as api:
1506 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1507 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="flagged"))
1509 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user1.username, tag="vip"))
1510 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="vip"))
1511 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="flagged"))
1513 # Search for users with "vip" tag
1514 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip"]))
1515 user_ids = {u.user_id for u in res.users}
1516 assert user1.id in user_ids
1517 assert user2.id in user_ids
1518 assert user3.id not in user_ids
1520 # Search for users with both "vip" AND "flagged" tags (AND logic)
1521 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip", "flagged"]))
1522 user_ids = {u.user_id for u in res.users}
1523 assert user2.id in user_ids
1524 assert user1.id not in user_ids
1526 # Search for non-existent tag returns no results
1527 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["nonexistent"]))
1528 assert len(res.users) == 0
1531def test_search_users_by_admin_note(db):
1532 super_user, super_token = generate_user(is_superuser=True)
1533 user1, _ = generate_user()
1534 user2, _ = generate_user()
1536 with real_admin_session(super_token) as api:
1537 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="suspicious activity"))
1538 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="normal user"))
1540 # Search by admin action log content (ilike)
1541 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_action_log="%suspicious%"))
1542 user_ids = {u.user_id for u in res.users}
1543 assert user1.id in user_ids
1544 assert user2.id not in user_ids
1547def test_ListAdminActions_empty(db):
1548 super_user, super_token = generate_user(is_superuser=True)
1550 with real_admin_session(super_token) as api:
1551 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1552 assert len(res.admin_actions) == 0
1553 assert res.next_page_token == ""
1556def test_ListAdminActions_returns_newest_first_with_target_info(db):
1557 super_user, super_token = generate_user(is_superuser=True)
1558 user1, _ = generate_user()
1559 user2, _ = generate_user()
1561 with real_admin_session(super_token) as api:
1562 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="first note"))
1563 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="second note"))
1564 api.BanUser(admin_pb2.BanUserReq(user=user1.username, admin_note="ban reason"))
1566 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1568 assert len(res.admin_actions) == 3
1569 # Newest first
1570 assert res.admin_actions[0].action_type == "ban"
1571 assert res.admin_actions[0].target_user_id == user1.id
1572 assert res.admin_actions[0].target_username == user1.username
1573 assert res.admin_actions[0].admin_user_id == super_user.id
1574 assert res.admin_actions[0].admin_username == super_user.username
1575 assert res.admin_actions[1].action_type == "note"
1576 assert res.admin_actions[1].target_user_id == user2.id
1577 assert res.admin_actions[2].action_type == "note"
1578 assert res.admin_actions[2].target_user_id == user1.id
1581def test_ListAdminActions_filter_by_admin_and_target(db):
1582 super1, super1_token = generate_user(is_superuser=True)
1583 super2, super2_token = generate_user(is_superuser=True)
1584 user1, _ = generate_user()
1585 user2, _ = generate_user()
1587 with real_admin_session(super1_token) as api:
1588 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super1 to user1"))
1589 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="from super1 to user2"))
1590 with real_admin_session(super2_token) as api:
1591 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super2 to user1"))
1593 with real_admin_session(super1_token) as api:
1594 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id))
1595 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super1 to user2"}
1597 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(target_user_id=user1.id))
1598 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super2 to user1"}
1600 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id, target_user_id=user1.id))
1601 assert [a.note for a in res.admin_actions] == ["from super1 to user1"]
1604def test_ListAdminActions_pagination(db):
1605 super_user, super_token = generate_user(is_superuser=True)
1606 user, _ = generate_user()
1608 with real_admin_session(super_token) as api:
1609 for i in range(3):
1610 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user.username, admin_note=f"note {i}"))
1612 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2))
1613 assert len(res.admin_actions) == 2
1614 assert res.next_page_token != ""
1615 first_page_notes = [a.note for a in res.admin_actions]
1617 res2 = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2, page_token=res.next_page_token))
1618 assert len(res2.admin_actions) == 1
1619 assert res2.next_page_token == ""
1621 all_notes = first_page_notes + [a.note for a in res2.admin_actions]
1622 assert set(all_notes) == {"note 0", "note 1", "note 2"}
1625def test_ListUserUploads(db):
1626 super_user, super_token = generate_user(is_superuser=True)
1627 user, _ = generate_user(complete_profile=False)
1628 other_user, _ = generate_user()
1630 with session_scope() as session:
1631 for i in range(3):
1632 session.add(
1633 Upload(
1634 key=f"key{i}",
1635 filename=f"photo{i}.jpg",
1636 creator_user_id=user.id,
1637 credit=f"credit {i}" if i == 0 else None,
1638 )
1639 )
1640 session.add(Upload(key="other_key", filename="other.jpg", creator_user_id=other_user.id))
1642 with real_admin_session(super_token) as api:
1643 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1645 assert len(res.uploads) == 3
1646 assert res.next_page_token == ""
1647 assert {u.filename for u in res.uploads} == {"photo0.jpg", "photo1.jpg", "photo2.jpg"}
1649 upload0 = next(u for u in res.uploads if u.key == "key0")
1650 assert upload0.credit == "credit 0"
1651 assert upload0.full_url.endswith("/img/full/photo0.jpg")
1652 assert upload0.thumbnail_url.endswith("/img/thumbnail/photo0.jpg")
1653 assert upload0.HasField("created")
1656def test_ListUserUploads_pagination(db):
1657 super_user, super_token = generate_user(is_superuser=True)
1658 user, _ = generate_user(complete_profile=False)
1660 with session_scope() as session:
1661 for i in range(3):
1662 session.add(Upload(key=f"key{i}", filename=f"photo{i}.jpg", creator_user_id=user.id))
1664 with real_admin_session(super_token) as api:
1665 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username, page_size=2))
1666 assert len(res.uploads) == 2
1667 assert res.next_page_token != ""
1668 first_page_keys = [u.key for u in res.uploads]
1670 res2 = api.ListUserUploads(
1671 admin_pb2.ListUserUploadsReq(user=user.username, page_size=2, page_token=res.next_page_token)
1672 )
1673 assert len(res2.uploads) == 1
1674 assert res2.next_page_token == ""
1676 all_keys = first_page_keys + [u.key for u in res2.uploads]
1677 assert set(all_keys) == {"key0", "key1", "key2"}
1680def test_ListUserUploads_uses(db):
1681 super_user, super_token = generate_user(is_superuser=True)
1682 user, _ = generate_user(complete_profile=False)
1684 with session_scope() as session:
1685 session.add(Upload(key="used_key", filename="used.jpg", creator_user_id=user.id))
1686 session.add(Upload(key="unused_key", filename="unused.jpg", creator_user_id=user.id))
1687 gallery = PhotoGallery(owner_user_id=user.id)
1688 session.add(gallery)
1689 session.flush()
1690 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="used_key", position=1.0))
1692 with real_admin_session(super_token) as api:
1693 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1695 uploads = {u.key: u for u in res.uploads}
1696 assert list(uploads["unused_key"].uses) == []
1698 used_uses = uploads["used_key"].uses
1699 assert len(used_uses) == 1
1700 assert used_uses[0].type == admin_pb2.UPLOAD_USE_TYPE_PROFILE_GALLERY_PHOTO_AVATAR
1701 assert used_uses[0].is_current
1702 assert used_uses[0].user_id == user.id
1705def test_ListUserUploads_not_found(db):
1706 super_user, super_token = generate_user(is_superuser=True)
1708 with real_admin_session(super_token) as api:
1709 with pytest.raises(grpc.RpcError) as e:
1710 api.ListUserUploads(admin_pb2.ListUserUploadsReq(user="nonexistent"))
1711 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1714# community invite feature tested in test_events.py
1715# SendBlogPostNotification tested in test_notifications.py
1716# MarkUserNeedsLocationUpdate tested in test_jail.py
1719def _ota_manifest(*, version, fingerprint, created_at="2026-05-31T00:00:00.000Z"):
1720 return {
1721 "id": f"id-{version}",
1722 "createdAt": created_at,
1723 "runtimeVersion": fingerprint,
1724 "launchAsset": {"key": "bundle", "url": f"https://cdn.testing.invalid/{version}/bundle.hbc"},
1725 "assets": [],
1726 "metadata": {},
1727 "extra": {},
1728 }
1731def _ota_signed_multipart(manifest):
1732 # Mimics the signed multipart body the CDN holds (signature header omitted; we only read the JSON).
1733 boundary = "COUCHERS_OTA_BOUNDARY"
1735 def part(name, body, content_type):
1736 return f'--{boundary}\r\ncontent-disposition: form-data; name="{name}"\r\ncontent-type: {content_type}\r\n\r\n{body}\r\n'
1738 body = (
1739 part("manifest", json.dumps(manifest), "application/json; charset=utf-8")
1740 + part("extensions", "{}", "application/json")
1741 + f"--{boundary}--\r\n"
1742 )
1743 return f"multipart/mixed; boundary={boundary}", body.encode()
1746def _patch_ota_cdn(manifests):
1747 # manifests: {version: manifest_dict}. URL is {cdn_root}/{version}/{platform}/manifest.
1748 def fake(url):
1749 version = url.split("/")[-3]
1750 if version not in manifests:
1751 return "multipart/mixed; boundary=COUCHERS_OTA_BOUNDARY", b""
1752 return _ota_signed_multipart(manifests[version])
1754 return patch("couchers.servicers.admin._fetch_signed_manifest", side_effect=fake)
1757def test_CreateOTAPackage(db):
1758 super_user, super_token = generate_user(is_superuser=True)
1760 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1761 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1762 res = api.CreateOTAPackage(
1763 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1764 )
1766 assert res.platform == admin_pb2.OTA_PLATFORM_IOS
1767 assert res.fingerprint == "ios-fp"
1768 assert res.version == "v1.3.1.aaaa"
1769 assert res.manifest_id == "id-v1.3.1.aaaa"
1770 assert res.banned is False
1771 assert res.live is True
1772 assert res.creator_user_id == super_user.id
1775def test_CreateOTAPackage_invalid(db):
1776 _, super_token = generate_user(is_superuser=True)
1778 manifests = {"v-incomplete": {"id": "x"}} # on the CDN but missing runtimeVersion / createdAt
1779 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1780 # missing version
1781 with pytest.raises(grpc.RpcError) as e:
1782 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1783 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1785 # nothing published at this version
1786 with pytest.raises(grpc.RpcError) as e:
1787 api.CreateOTAPackage(
1788 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-missing")
1789 )
1790 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1792 # manifest present but missing required fields
1793 with pytest.raises(grpc.RpcError) as e:
1794 api.CreateOTAPackage(
1795 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-incomplete")
1796 )
1797 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1800def test_CreateOTAPackage_rejects_duplicate_version(db):
1801 _, super_token = generate_user(is_superuser=True)
1803 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1804 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1805 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa"))
1806 with pytest.raises(grpc.RpcError) as e:
1807 api.CreateOTAPackage(
1808 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1809 )
1810 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1813def test_ListOTAPackages(db):
1814 _, super_token = generate_user(is_superuser=True)
1816 manifests = {
1817 "v1.3.1.ios": _ota_manifest(version="v1.3.1.ios", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1818 "v1.3.2.ios": _ota_manifest(version="v1.3.2.ios", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1819 "v1.3.2.android": _ota_manifest(
1820 version="v1.3.2.android", fingerprint="android-fp", created_at="2026-06-01T00:00:00.000Z"
1821 ),
1822 }
1823 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1824 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.ios"))
1825 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.ios"))
1826 api.CreateOTAPackage(
1827 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_ANDROID, version="v1.3.2.android")
1828 )
1830 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1831 # newest (by manifest createdAt) first
1832 assert [p.version for p in res.packages] == ["v1.3.2.android", "v1.3.2.ios", "v1.3.1.ios"]
1833 # only the newest per (platform, fingerprint) is live
1834 live = {p.version: p.live for p in res.packages}
1835 assert live == {"v1.3.2.android": True, "v1.3.2.ios": True, "v1.3.1.ios": False}
1837 ios = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1838 assert [p.version for p in ios.packages] == ["v1.3.2.ios", "v1.3.1.ios"]
1841def test_BanOTAPackage(db):
1842 super_user, super_token = generate_user(is_superuser=True)
1844 manifests = {
1845 "v1.3.1.good": _ota_manifest(
1846 version="v1.3.1.good", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"
1847 ),
1848 "v1.3.2.bad": _ota_manifest(version="v1.3.2.bad", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1849 }
1850 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1851 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.good"))
1852 second = api.CreateOTAPackage(
1853 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.bad")
1854 )
1855 assert second.live is True
1857 banned = api.BanOTAPackage(
1858 admin_pb2.BanOTAPackageReq(ota_package_id=second.ota_package_id, reason="bad bundle")
1859 )
1860 assert banned.banned is True
1861 assert banned.banned_reason == "bad bundle"
1862 assert banned.banned_by_user_id == super_user.id
1863 assert banned.live is False
1865 # banning the newest stops new check-ins getting it; the previous one becomes live again
1866 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(include_banned=True))
1867 live = {p.version: p.live for p in res.packages}
1868 assert live == {"v1.3.2.bad": False, "v1.3.1.good": True}
1870 # banned packages are excluded by default
1871 non_banned = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1872 assert [p.version for p in non_banned.packages] == ["v1.3.1.good"]
1875def test_BanOTAPackage_requires_reason(db):
1876 _, super_token = generate_user(is_superuser=True)
1878 manifests = {
1879 "v1.3.1": _ota_manifest(version="v1.3.1", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1880 }
1881 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1882 pkg = api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1"))
1883 with pytest.raises(grpc.RpcError) as e:
1884 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id))
1885 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1886 with pytest.raises(grpc.RpcError) as e:
1887 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id, reason=" "))
1888 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1891def test_BanOTAPackage_not_found(db):
1892 _, super_token = generate_user(is_superuser=True)
1894 with real_admin_session(super_token) as api:
1895 with pytest.raises(grpc.RpcError) as e:
1896 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=123456, reason="never mind"))
1897 assert e.value.code() == grpc.StatusCode.NOT_FOUND