Coverage for app/backend/src/tests/test_admin.py: 100%
1062 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-26 00:33 +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 add_users_to_new_moderation_list, generate_user, make_friends
43from tests.fixtures.misc import EmailCollector, PushCollector
44from tests.fixtures.sessions import (
45 account_session,
46 auth_api_session,
47 events_session,
48 real_admin_session,
49 references_session,
50 reporting_session,
51 requests_session,
52)
53from tests.test_communities import create_community
54from tests.test_requests import valid_request_text
57@pytest.fixture(autouse=True)
58def _(testconfig):
59 pass
62def test_access_by_normal_user(db):
63 normal_user, normal_token = generate_user()
65 with real_admin_session(normal_token) as api:
66 # all requests to the admin servicer should break when done by a non-super_user
67 with pytest.raises(grpc.RpcError) as e:
68 api.GetUserDetails(
69 admin_pb2.GetUserDetailsReq(
70 user=str(normal_user.id),
71 )
72 )
73 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
76def test_GetNonvisibleUserAccessLog(db):
77 super_user, super_token = generate_user(is_superuser=True)
78 target, _ = generate_user(username="target")
79 viewer, _ = generate_user(username="viewer")
81 with session_scope() as session:
82 session.add(
83 NonvisibleUserAccess(
84 access_type=NonvisibleUserAccessType.login_attempt,
85 target_user_id=target.id,
86 target_state=NonvisibleUserState.banned,
87 actor_user_id=target.id,
88 ip_address="1.2.3.4",
89 sofa="device-cookie",
90 )
91 )
92 session.add(
93 NonvisibleUserAccess(
94 access_type=NonvisibleUserAccessType.ghost_served,
95 target_user_id=target.id,
96 target_state=NonvisibleUserState.banned,
97 actor_user_id=viewer.id,
98 )
99 )
100 session.add(
101 NonvisibleUserAccess(
102 access_type=NonvisibleUserAccessType.ghost_served,
103 target_user_id=target.id,
104 target_state=NonvisibleUserState.banned,
105 actor_user_id=None,
106 )
107 )
109 with real_admin_session(super_token) as api:
110 res = api.GetNonvisibleUserAccessLog(admin_pb2.GetNonvisibleUserAccessLogReq(user="target"))
112 assert len(res.entries) == 3
113 for entry in res.entries:
114 assert entry.target_user_id == target.id
115 assert entry.target_state == admin_pb2.NONVISIBLE_USER_STATE_BANNED
117 login = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_LOGIN_ATTEMPT]
118 views = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_GHOST_SERVED]
119 assert len(login) == 1
120 assert len(views) == 2
122 assert login[0].actor_user_id.value == target.id
123 assert login[0].actor_username == "target"
124 assert login[0].ip_address == "1.2.3.4"
125 assert login[0].sofa == "device-cookie"
127 logged_in_view = [e for e in views if e.actor_username == "viewer"]
128 logged_out_view = [e for e in views if not e.actor_username]
129 assert len(logged_in_view) == 1
130 assert logged_in_view[0].actor_user_id.value == viewer.id
131 assert len(logged_out_view) == 1
132 assert not logged_out_view[0].HasField("actor_user_id")
135def test_GetUser(db):
136 super_user, super_token = generate_user(is_superuser=True)
137 normal_user, normal_token = generate_user()
139 with real_admin_session(super_token) as api:
140 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
141 assert res.user_id == normal_user.id
142 assert res.username == normal_user.username
144 with real_admin_session(super_token) as api:
145 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note="Testing banning"))
147 with real_admin_session(super_token) as api:
148 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
149 assert res.user_id == normal_user.id
150 assert res.username == normal_user.username
153def test_GetUserDetails(db):
154 super_user, super_token = generate_user(is_superuser=True)
155 normal_user, normal_token = generate_user()
157 with real_admin_session(super_token) as api:
158 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=str(normal_user.id)))
159 assert res.user_id == normal_user.id
160 assert res.username == normal_user.username
161 assert res.email == normal_user.email
162 assert res.gender == normal_user.gender
163 assert parse_date(res.birthdate) == normal_user.birthdate
164 assert not res.banned
165 assert not res.deleted
167 with real_admin_session(super_token) as api:
168 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
169 assert res.user_id == normal_user.id
170 assert res.username == normal_user.username
171 assert res.email == normal_user.email
172 assert res.gender == normal_user.gender
173 assert parse_date(res.birthdate) == normal_user.birthdate
174 assert not res.banned
175 assert not res.deleted
177 with real_admin_session(super_token) as api:
178 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.email))
179 assert res.user_id == normal_user.id
180 assert res.username == normal_user.username
181 assert res.email == normal_user.email
182 assert res.gender == normal_user.gender
183 assert parse_date(res.birthdate) == normal_user.birthdate
184 assert not res.banned
185 assert not res.deleted
188def test_ChangeUserGender(db, email_collector: EmailCollector, push_collector: PushCollector):
189 super_user, super_token = generate_user(is_superuser=True)
190 normal_user, normal_token = generate_user()
192 with real_admin_session(super_token) as api:
193 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
194 assert res.user_id == normal_user.id
195 assert res.username == normal_user.username
196 assert res.email == normal_user.email
197 assert res.gender == "Machine"
198 assert parse_date(res.birthdate) == normal_user.birthdate
199 assert not res.banned
200 assert not res.deleted
202 email = email_collector.pop_for_recipient(normal_user.email, last=True)
203 assert email.subject == "[TEST] Your gender was changed"
204 assert email.recipient == normal_user.email
205 assert "Machine" in email.plain
206 assert "Machine" in email.html
208 push = push_collector.pop_for_user(normal_user.id, last=True)
209 assert push.content.title == "Gender changed"
210 assert push.content.body == "An admin changed your gender to Machine."
213def test_ChangeUserBirthdate(db, email_collector: EmailCollector, push_collector: PushCollector):
214 super_user, super_token = generate_user(is_superuser=True)
215 normal_user, normal_token = generate_user(birthdate=date(year=2000, month=1, day=1))
217 with real_admin_session(super_token) as api:
218 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
219 assert parse_date(res.birthdate) == date(year=2000, month=1, day=1)
221 res = api.ChangeUserBirthdate(
222 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-05-25")
223 )
225 assert res.user_id == normal_user.id
226 assert res.username == normal_user.username
227 assert res.email == normal_user.email
228 assert res.birthdate == "1990-05-25"
229 assert res.gender == normal_user.gender
230 assert not res.banned
231 assert not res.deleted
233 email = email_collector.pop_for_recipient(normal_user.email, last=True)
234 assert email.subject == "[TEST] Your date of birth was changed"
235 assert email.recipient == normal_user.email
236 assert "1990" in email.plain
237 assert "1990" in email.html
239 push = push_collector.pop_for_user(normal_user.id, last=True)
240 assert push.content.title == "Birthdate changed"
241 assert push.content.body == "An admin changed your date of birth to May 25, 1990."
244def test_BanUser(db):
245 super_user, super_token = generate_user(is_superuser=True)
246 normal_user, _ = generate_user()
247 admin_note = "A good reason"
249 with real_admin_session(super_token) as api:
250 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note=admin_note))
251 assert res.user_id == normal_user.id
252 assert res.username == normal_user.username
253 assert res.email == normal_user.email
254 assert res.gender == normal_user.gender
255 assert parse_date(res.birthdate) == normal_user.birthdate
256 assert res.banned
257 assert not res.deleted
258 assert len(res.admin_actions) == 1
259 assert res.admin_actions[0].action_type == "ban"
260 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
261 assert res.admin_actions[0].note == admin_note
262 assert res.admin_actions[0].admin_user_id == super_user.id
263 assert res.admin_actions[0].admin_username == super_user.username
266def test_UnbanUser(db):
267 super_user, super_token = generate_user(is_superuser=True)
268 normal_user, _ = generate_user()
269 admin_note = "A good reason"
271 with real_admin_session(super_token) as api:
272 res = api.UnbanUser(admin_pb2.UnbanUserReq(user=normal_user.username, admin_note=admin_note))
273 assert res.user_id == normal_user.id
274 assert res.username == normal_user.username
275 assert res.email == normal_user.email
276 assert res.gender == normal_user.gender
277 assert parse_date(res.birthdate) == normal_user.birthdate
278 assert not res.banned
279 assert not res.deleted
280 assert len(res.admin_actions) == 1
281 assert res.admin_actions[0].action_type == "unban"
282 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
285def test_ShadowUser(db):
286 super_user, super_token = generate_user(is_superuser=True)
287 surfer, surfer_token = generate_user()
288 host, _ = generate_user()
289 admin_note = "Spammer"
291 # Create a host request from `surfer` and approve its moderation state to VISIBLE so we can verify the cascade
292 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
293 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
294 with requests_session(surfer_token) as api:
295 host_request_id = api.CreateHostRequest(
296 requests_pb2.CreateHostRequestReq(
297 host_user_id=host.id,
298 from_date=today_plus_2,
299 to_date=today_plus_3,
300 text=valid_request_text(),
301 )
302 ).host_request_id
303 with session_scope() as session:
304 state = session.execute(
305 select(ModerationState)
306 .where(ModerationState.object_type == ModerationObjectType.host_request)
307 .where(ModerationState.object_id == host_request_id)
308 ).scalar_one()
309 state.visibility = ModerationVisibility.visible
311 with real_admin_session(super_token) as api:
312 res = api.ShadowUser(admin_pb2.ShadowUserReq(user=surfer.username, admin_note=admin_note))
313 assert res.user_id == surfer.id
314 assert res.shadowed
315 assert not res.banned
316 assert not res.deleted
317 assert len(res.admin_actions) == 1
318 assert res.admin_actions[0].action_type == "shadow"
319 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
320 assert res.admin_actions[0].note == admin_note
322 # The previously-visible host request is now shadowed
323 with session_scope() as session:
324 state = session.execute(
325 select(ModerationState)
326 .where(ModerationState.object_type == ModerationObjectType.host_request)
327 .where(ModerationState.object_id == host_request_id)
328 ).scalar_one()
329 assert state.visibility == ModerationVisibility.shadowed
332def test_UnshadowUser(db):
333 super_user, super_token = generate_user(is_superuser=True)
334 surfer, surfer_token = generate_user()
335 host, _ = generate_user()
337 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
338 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
339 with requests_session(surfer_token) as api:
340 shadow_cascade_request_id = api.CreateHostRequest(
341 requests_pb2.CreateHostRequestReq(
342 host_user_id=host.id,
343 from_date=today_plus_2,
344 to_date=today_plus_3,
345 text=valid_request_text(),
346 )
347 ).host_request_id
348 admin_hidden_request_id = api.CreateHostRequest(
349 requests_pb2.CreateHostRequestReq(
350 host_user_id=host.id,
351 from_date=today_plus_2,
352 to_date=today_plus_3,
353 text=valid_request_text(),
354 )
355 ).host_request_id
357 with session_scope() as session:
358 session.execute(select(User).where(User.id == surfer.id)).scalar_one().shadowed_at = now()
359 session.execute(
360 select(ModerationState)
361 .where(ModerationState.object_type == ModerationObjectType.host_request)
362 .where(ModerationState.object_id == shadow_cascade_request_id)
363 ).scalar_one().visibility = ModerationVisibility.shadowed
364 session.execute(
365 select(ModerationState)
366 .where(ModerationState.object_type == ModerationObjectType.host_request)
367 .where(ModerationState.object_id == admin_hidden_request_id)
368 ).scalar_one().visibility = ModerationVisibility.hidden
370 with real_admin_session(super_token) as api:
371 res = api.UnshadowUser(admin_pb2.UnshadowUserReq(user=surfer.username, admin_note="rehabilitated"))
372 assert not res.shadowed
373 assert len(res.admin_actions) == 1
374 assert res.admin_actions[0].action_type == "unshadow"
375 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
377 with session_scope() as session:
378 assert (
379 session.execute(
380 select(ModerationState)
381 .where(ModerationState.object_type == ModerationObjectType.host_request)
382 .where(ModerationState.object_id == shadow_cascade_request_id)
383 )
384 .scalar_one()
385 .visibility
386 == ModerationVisibility.visible
387 )
388 assert (
389 session.execute(
390 select(ModerationState)
391 .where(ModerationState.object_type == ModerationObjectType.host_request)
392 .where(ModerationState.object_id == admin_hidden_request_id)
393 )
394 .scalar_one()
395 .visibility
396 == ModerationVisibility.hidden
397 )
400def test_ShadowUser_blank_note(db):
401 super_user, super_token = generate_user(is_superuser=True)
402 normal_user, _ = generate_user()
404 with real_admin_session(super_token) as api:
405 with pytest.raises(grpc.RpcError) as e:
406 api.ShadowUser(admin_pb2.ShadowUserReq(user=normal_user.username, admin_note=" \t "))
407 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
410def test_AddAdminNote(db):
411 super_user, super_token = generate_user(is_superuser=True)
412 normal_user, _ = generate_user()
413 admin_note1 = "User reported strange behavior"
414 admin_note2 = "Insert private information here"
416 with real_admin_session(super_token) as api:
417 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note1))
418 assert res.user_id == normal_user.id
419 assert res.username == normal_user.username
420 assert res.email == normal_user.email
421 assert res.gender == normal_user.gender
422 assert parse_date(res.birthdate) == normal_user.birthdate
423 assert not res.banned
424 assert not res.deleted
425 assert len(res.admin_actions) == 1
426 assert res.admin_actions[0].action_type == "note"
427 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
428 assert res.admin_actions[0].note == admin_note1
430 with real_admin_session(super_token) as api:
431 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note2))
432 assert len(res.admin_actions) == 2
433 assert res.admin_actions[0].note == admin_note1
434 assert res.admin_actions[1].note == admin_note2
437def test_AddAdminNote_blank(db):
438 super_user, super_token = generate_user(is_superuser=True)
439 normal_user, _ = generate_user()
440 empty_admin_note = " \t \n "
442 with real_admin_session(super_token) as api:
443 with pytest.raises(grpc.RpcError) as e:
444 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=empty_admin_note))
445 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
446 assert e.value.details() == "Provide exactly one of admin_note or data."
449def test_AddAdminNote_data(db):
450 super_user, super_token = generate_user(is_superuser=True)
451 normal_user, _ = generate_user()
452 payload = '{"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}'
454 with real_admin_session(super_token) as api:
455 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data=payload))
456 assert len(res.admin_actions) == 1
457 assert res.admin_actions[0].action_type == "note"
458 assert res.admin_actions[0].note == ""
459 assert json.loads(res.admin_actions[0].data) == {"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}
462def test_AddAdminNote_both_note_and_data(db):
463 super_user, super_token = generate_user(is_superuser=True)
464 normal_user, _ = generate_user()
466 with real_admin_session(super_token) as api:
467 with pytest.raises(grpc.RpcError) as e:
468 api.AddAdminNote(
469 admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="note text", data='{"x": 1}')
470 )
471 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
472 assert e.value.details() == "Provide exactly one of admin_note or data."
475def test_AddAdminNote_neither(db):
476 super_user, super_token = generate_user(is_superuser=True)
477 normal_user, _ = generate_user()
479 with real_admin_session(super_token) as api:
480 with pytest.raises(grpc.RpcError) as e:
481 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username))
482 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
483 assert e.value.details() == "Provide exactly one of admin_note or data."
486def test_AddAdminNote_invalid_json(db):
487 super_user, super_token = generate_user(is_superuser=True)
488 normal_user, _ = generate_user()
490 with real_admin_session(super_token) as api:
491 with pytest.raises(grpc.RpcError) as e:
492 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data="{not valid json"))
493 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
494 assert e.value.details() == "The admin note data must be valid JSON."
497def test_admin_content_reports(db):
498 super_user, super_token = generate_user(is_superuser=True)
499 normal_user, token = generate_user()
500 bad_user1, _ = generate_user()
501 bad_user2, _ = generate_user()
503 with reporting_session(token) as api:
504 api.Report(
505 reporting_pb2.ReportReq(
506 reason="spam",
507 description="r1",
508 content_ref="comment/123",
509 author_user=bad_user1.username,
510 user_agent="n/a",
511 page="https://couchers.org/comment/123",
512 )
513 )
514 api.Report(
515 reporting_pb2.ReportReq(
516 reason="spam",
517 description="r2",
518 content_ref="comment/124",
519 author_user=bad_user2.username,
520 user_agent="n/a",
521 page="https://couchers.org/comment/124",
522 )
523 )
524 api.Report(
525 reporting_pb2.ReportReq(
526 reason="something else",
527 description="r3",
528 content_ref="page/321",
529 author_user=bad_user1.username,
530 user_agent="n/a",
531 page="https://couchers.org/page/321",
532 )
533 )
535 with session_scope() as session:
536 id_by_description: dict[str, int] = dict(
537 session.execute(select(ContentReport.description, ContentReport.id)).all() # type: ignore[arg-type]
538 )
540 with real_admin_session(super_token) as api:
541 with pytest.raises(grpc.RpcError) as e:
542 api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=-1))
543 assert e.value.code() == grpc.StatusCode.NOT_FOUND
544 assert e.value.details() == "Content report not found."
546 res = api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=id_by_description["r2"]))
547 rep = res.content_report
548 assert rep.content_report_id == id_by_description["r2"]
549 assert rep.reporting_user_id == normal_user.id
550 assert rep.author_user_id == bad_user2.id
551 assert rep.reason == "spam"
552 assert rep.description == "r2"
553 assert rep.content_ref == "comment/124"
554 assert rep.user_agent == "n/a"
555 assert rep.page == "https://couchers.org/comment/124"
557 res = api.GetContentReportsForAuthor(admin_pb2.GetContentReportsForAuthorReq(user=bad_user1.username))
558 assert res.content_reports[0].content_report_id == id_by_description["r3"]
559 assert res.content_reports[1].content_report_id == id_by_description["r1"]
562def test_DeleteUser(db):
563 super_user, super_token = generate_user(is_superuser=True)
564 normal_user, normal_token = generate_user()
566 with real_admin_session(super_token) as api:
567 res = api.DeleteUser(admin_pb2.DeleteUserReq(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 res.deleted
576 with real_admin_session(super_token) as api:
577 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
578 assert res.user_id == normal_user.id
579 assert res.username == normal_user.username
580 assert res.email == normal_user.email
581 assert res.gender == normal_user.gender
582 assert parse_date(res.birthdate) == normal_user.birthdate
583 assert not res.banned
584 assert not res.deleted
587def test_RecoverDeletedUser_after_user_initiated_deletion(db, push_collector: PushCollector):
588 """
589 When a user deletes their account through the normal flow (ConfirmDeleteAccount),
590 undelete_token and undelete_until are set. The admin RecoverDeletedUser must clear
591 these fields to satisfy the undelete_nullity database constraint.
592 """
593 super_user, super_token = generate_user(is_superuser=True)
594 normal_user, normal_token = generate_user()
595 user_id = normal_user.id
597 # User initiates account deletion
598 with account_session(normal_token) as account:
599 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
601 # Get the deletion confirmation token
602 with session_scope() as session:
603 deletion_token = session.execute(select(AccountDeletionToken)).scalar_one().token
605 # User confirms account deletion (this sets undelete_token and undelete_until)
606 with auth_api_session() as (auth_api, metadata_interceptor):
607 auth_api.ConfirmDeleteAccount(auth_pb2.ConfirmDeleteAccountReq(token=deletion_token))
609 # Verify the user is deleted and has undelete fields set
610 with session_scope() as session:
611 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
612 assert user.deleted_at is not None
613 assert user.undelete_token is not None
614 assert user.undelete_until is not None
616 # Admin recovers the user
617 with real_admin_session(super_token) as api:
618 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
619 assert res.user_id == user_id
620 assert not res.deleted
622 # Verify undelete fields are cleared
623 with session_scope() as session:
624 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
625 assert user.deleted_at is None
626 assert user.undelete_token is None
627 assert user.undelete_until is None
630def test_CreateApiKey(db, email_collector: EmailCollector, push_collector: PushCollector):
631 with session_scope() as session:
632 super_user, super_token = generate_user(is_superuser=True)
633 normal_user, normal_token = generate_user()
635 assert (
636 session.execute(
637 select(func.count())
638 .select_from(UserSession)
639 .where(UserSession.is_api_key == True)
640 .where(UserSession.user_id == normal_user.id)
641 ).scalar_one()
642 == 0
643 )
645 with real_admin_session(super_token) as api:
646 res = api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=normal_user.username))
648 email = email_collector.pop_for_recipient(normal_user.email, last=True)
649 assert email.subject == "[TEST] Your API key for Couchers.org"
651 with session_scope() as session:
652 token = session.execute(
653 select(UserSession.token)
654 .where(UserSession.is_valid)
655 .where(UserSession.is_api_key == True)
656 .where(UserSession.user_id == normal_user.id)
657 ).scalar_one()
659 assert token in email.plain
660 assert token in email.html
662 assert email.recipient == normal_user.email
663 assert "api key" in email.subject.lower()
664 unique_string = "We've issued you with the following API key:"
665 assert unique_string in email.plain
666 assert unique_string in email.html
667 assert "support@couchers.org" in email.plain
668 assert "support@couchers.org" in email.html
670 push = push_collector.pop_for_user(normal_user.id, last=True)
671 assert push.content.title == "API key created"
672 assert push.content.body == "Details were sent to you via email."
675def test_GetChats(db):
676 super_user, super_token = generate_user(is_superuser=True)
677 normal_user, normal_token = generate_user()
679 with real_admin_session(super_token) as api:
680 res = api.GetChats(admin_pb2.GetChatsReq(user=normal_user.username))
681 assert res.user.user_id == normal_user.id
682 assert res.user.username == normal_user.username
683 assert res.user.name == normal_user.name
684 # New user should have no chats
685 assert len(res.host_requests) == 0
686 assert len(res.group_chats) == 0
689def test_badges(db, email_collector: EmailCollector, push_collector: PushCollector):
690 super_user, super_token = generate_user(is_superuser=True)
691 normal_user, normal_token = generate_user()
693 with real_admin_session(super_token) as api:
694 # can add a badge
695 assert "swagster" not in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
696 res = api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
697 assert "swagster" in res.badges
699 # badge emails are disabled by default
700 assert email_collector.count_for_recipient(normal_user.email) == 0
702 push = push_collector.pop_for_user(normal_user.id, last=True)
703 assert push.content.title == "New profile badge: Swagster"
704 assert push.content.body == "The Swagster badge was added to your profile."
706 # can't add/edit special tags
707 with pytest.raises(grpc.RpcError) as e:
708 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="founder"))
709 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
710 assert e.value.details() == "Admins cannot edit that badge."
712 # double add badge
713 with pytest.raises(grpc.RpcError) as e:
714 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
715 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
716 assert e.value.details() == "The user already has that badge."
718 # can remove badge
719 assert "swagster" in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
720 res = api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
721 assert "swagster" not in res.badges
723 # badge emails are disabled by default
724 assert email_collector.count_for_recipient(normal_user.email) == 0
726 push = push_collector.pop_for_user(normal_user.id, last=True)
727 assert push.content.title == "Profile badge removed"
728 assert push.content.body == "The Swagster badge was removed from your profile."
730 # not found on user
731 with pytest.raises(grpc.RpcError) as e:
732 api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
733 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
734 assert e.value.details() == "The user does not have that badge."
736 # not found in general
737 with pytest.raises(grpc.RpcError) as e:
738 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="nonexistentbadge"))
739 assert e.value.code() == grpc.StatusCode.NOT_FOUND
740 assert e.value.details() == "Badge not found."
743def test_DeleteEvent(db):
744 super_user, super_token = generate_user(is_superuser=True)
745 normal_user, normal_token = generate_user()
747 with session_scope() as session:
748 create_community(session, 0, 2, "Community", [normal_user], [], None)
750 start_time = now() + timedelta(hours=2)
751 end_time = start_time + timedelta(hours=3)
752 with events_session(normal_token) as api:
753 res = api.CreateEvent(
754 events_pb2.CreateEventReq(
755 title="Dummy Title",
756 content="Dummy content.",
757 photo_key=None,
758 location=events_pb2.EventLocation(
759 address="Near Null Island",
760 lat=0.1,
761 lng=0.2,
762 ),
763 start_datetime_iso8601_local=datetime_to_iso8601_local(start_time),
764 end_datetime_iso8601_local=datetime_to_iso8601_local(end_time),
765 )
766 )
767 event_id = res.event_id
768 assert not res.is_deleted
770 with session_scope() as session:
771 with real_admin_session(super_token) as api:
772 api.DeleteEvent(
773 admin_pb2.DeleteEventReq(
774 event_id=event_id,
775 )
776 )
777 occurrence = session.get_one(EventOccurrence, ident=event_id)
778 assert occurrence.is_deleted
781def test_ListUserIds(db):
782 super_user, super_token = generate_user(is_superuser=True)
783 normal_user, normal_token = generate_user()
785 with real_admin_session(super_token) as api:
786 res = api.ListUserIds(
787 admin_pb2.ListUserIdsReq(
788 start_time=Timestamp_from_datetime(datetime(2000, 1, 1, tzinfo=UTC)),
789 end_time=Timestamp_from_datetime(now()),
790 )
791 )
792 assert len(res.user_ids) == 2
793 assert sorted(res.user_ids) == sorted([super_user.id, normal_user.id])
795 with real_admin_session(super_token) as api:
796 res = api.ListUserIds(
797 admin_pb2.ListUserIdsReq(start_time=Timestamp_from_datetime(now()), end_time=Timestamp_from_datetime(now()))
798 )
799 assert res.user_ids == []
802def test_EditReferenceText(db):
803 super_user, super_token = generate_user(is_superuser=True)
804 test_new_text = "New Text"
806 user1, user1_token = generate_user()
807 user2, user2_token = generate_user()
808 make_friends(user1, user2)
810 with session_scope() as session:
811 with references_session(user1_token) as api:
812 reference = api.WriteFriendReference(
813 references_pb2.WriteFriendReferenceReq(
814 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
815 )
816 )
818 with real_admin_session(super_token) as admin_api:
819 admin_api.EditReferenceText(
820 admin_pb2.EditReferenceTextReq(reference_id=reference.reference_id, new_text=test_new_text)
821 )
823 session.expire_all()
825 modified_reference = session.execute(
826 select(Reference).where(Reference.id == reference.reference_id)
827 ).scalar_one()
828 assert modified_reference.text == test_new_text
831def test_DeleteReference_deprecated(db):
832 """DeleteReference is deprecated; admins should hide via UMS instead."""
833 super_user, super_token = generate_user(is_superuser=True)
835 user1, user1_token = generate_user()
836 user2, user2_token = generate_user()
837 make_friends(user1, user2)
839 with references_session(user1_token) as api:
840 reference = api.WriteFriendReference(
841 references_pb2.WriteFriendReferenceReq(
842 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
843 )
844 )
846 with real_admin_session(super_token) as admin_api:
847 with pytest.raises(grpc.RpcError) as e:
848 admin_api.DeleteReference(admin_pb2.DeleteReferenceReq(reference_id=reference.reference_id))
849 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
852def test_GetUserReferences(db):
853 super_user, super_token = generate_user(is_superuser=True)
855 user1, user1_token = generate_user()
856 user2, user2_token = generate_user()
857 user3, user3_token = generate_user()
858 make_friends(user1, user2)
859 make_friends(user1, user3)
860 make_friends(user2, user3)
862 # user1 writes reference about user2
863 with references_session(user1_token) as api:
864 ref1 = api.WriteFriendReference(
865 references_pb2.WriteFriendReferenceReq(
866 to_user_id=user2.id,
867 text="Reference from user1 to user2",
868 private_text="",
869 was_appropriate=True,
870 rating=1,
871 )
872 )
874 # user2 writes reference about user1
875 with references_session(user2_token) as api:
876 ref2 = api.WriteFriendReference(
877 references_pb2.WriteFriendReferenceReq(
878 to_user_id=user1.id,
879 text="Reference from user2 to user1",
880 private_text="Private note",
881 was_appropriate=True,
882 rating=0.8,
883 )
884 )
886 # user3 writes reference about user1
887 with references_session(user3_token) as api:
888 ref3 = api.WriteFriendReference(
889 references_pb2.WriteFriendReferenceReq(
890 to_user_id=user1.id,
891 text="Reference from user3 to user1",
892 private_text="",
893 was_appropriate=False,
894 rating=0.5,
895 )
896 )
898 # Test GetUserReferences for user1 (admin view shows everything regardless of UMS state).
899 with real_admin_session(super_token) as admin_api:
900 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
902 # user1 wrote 1 reference
903 assert len(res.references_from) == 1
904 assert res.references_from[0].reference_id == ref1.reference_id
905 assert res.references_from[0].from_user_id == user1.id
906 assert res.references_from[0].to_user_id == user2.id
907 assert res.references_from[0].text == "Reference from user1 to user2"
909 # user1 received 2 references
910 assert len(res.references_to) == 2
911 # Ordered by id descending, so ref3 comes first
912 assert res.references_to[0].reference_id == ref3.reference_id
913 assert res.references_to[0].was_appropriate is False
915 assert res.references_to[1].reference_id == ref2.reference_id
916 assert res.references_to[1].private_text == "Private note"
917 assert res.references_to[1].rating == 0.8
920def test_GetUserReferences_not_found(db):
921 super_user, super_token = generate_user(is_superuser=True)
923 with real_admin_session(super_token) as admin_api:
924 with pytest.raises(grpc.RpcError) as e:
925 admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user="nonexistent"))
926 assert e.value.code() == grpc.StatusCode.NOT_FOUND
929def test_GetFriendRequests(db):
930 super_user, super_token = generate_user(is_superuser=True)
932 user1, _ = generate_user()
933 user2, _ = generate_user()
934 user3, _ = generate_user()
935 user4, _ = generate_user()
937 # Create a mix of friend requests directly so we control the state
938 def _add_friend_request(from_user_id, to_user_id, status, visibility, time_responded=None):
939 with session_scope() as session:
940 mod_state = ModerationState(
941 object_type=ModerationObjectType.friend_request,
942 object_id=0,
943 visibility=visibility,
944 )
945 session.add(mod_state)
946 session.flush()
947 rel = FriendRelationship(
948 from_user_id=from_user_id,
949 to_user_id=to_user_id,
950 status=status,
951 moderation_state_id=mod_state.id,
952 time_responded=time_responded,
953 )
954 session.add(rel)
955 session.flush()
956 mod_state.object_id = rel.id
958 # user1 -> user2: pending, shadowed
959 _add_friend_request(user1.id, user2.id, FriendStatus.pending, ModerationVisibility.shadowed)
960 # user1 -> user3: accepted, visible
961 _add_friend_request(user1.id, user3.id, FriendStatus.accepted, ModerationVisibility.visible, time_responded=now())
962 # user4 -> user1: rejected, visible
963 _add_friend_request(user4.id, user1.id, FriendStatus.rejected, ModerationVisibility.visible, time_responded=now())
965 with real_admin_session(super_token) as admin_api:
966 res = admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user=user1.username))
968 # user1 sent two: to user2 (pending) and to user3 (accepted), ordered by id desc
969 assert len(res.sent) == 2
970 assert res.sent[0].from_user.user_id == user1.id
971 assert res.sent[0].to_user.user_id == user3.id
972 assert res.sent[0].status == "accepted"
973 assert res.sent[0].HasField("time_responded")
974 assert res.sent[0].moderation_visibility == "visible"
976 assert res.sent[1].from_user.user_id == user1.id
977 assert res.sent[1].to_user.user_id == user2.id
978 assert res.sent[1].status == "pending"
979 assert not res.sent[1].HasField("time_responded")
980 assert res.sent[1].moderation_visibility == "shadowed"
982 # user1 received one: from user4 (rejected)
983 assert len(res.received) == 1
984 assert res.received[0].from_user.user_id == user4.id
985 assert res.received[0].to_user.user_id == user1.id
986 assert res.received[0].status == "rejected"
989def test_GetFriendRequests_not_found(db):
990 super_user, super_token = generate_user(is_superuser=True)
992 with real_admin_session(super_token) as admin_api:
993 with pytest.raises(grpc.RpcError) as e:
994 admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user="nonexistent"))
995 assert e.value.code() == grpc.StatusCode.NOT_FOUND
998def test_AddUsersToModerationUserList(db):
999 super_user, super_token = generate_user(is_superuser=True)
1000 user1, _ = generate_user()
1001 user2, _ = generate_user()
1002 user3, _ = generate_user()
1003 user4, _ = generate_user()
1004 user5, _ = generate_user()
1005 moderation_list_id = add_users_to_new_moderation_list([user1])
1007 with session_scope() as session:
1008 with real_admin_session(super_token) as api:
1009 # Test adding users to a non-existent moderation list (should raise an error)
1010 with pytest.raises(grpc.RpcError) as e:
1011 api.AddUsersToModerationUserList(
1012 admin_pb2.AddUsersToModerationUserListReq(users=[user2.username], moderation_list_id=999),
1013 )
1014 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1015 assert "Moderation user list not found." == e.value.details()
1017 # Test with non-existent user (should raise an error)
1018 with pytest.raises(grpc.RpcError) as e:
1019 api.AddUsersToModerationUserList(
1020 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, "nonexistent"]),
1021 )
1022 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1023 assert "Couldn't find that user." == e.value.details()
1025 # Test successful creation of new moderation list (no moderation_list_id provided)
1026 res = api.AddUsersToModerationUserList(
1027 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, user2.username, user3.username]),
1028 )
1029 assert res.moderation_list_id > 0
1030 with session_scope() as session:
1031 moderation_user_list = session.get(ModerationUserList, res.moderation_list_id)
1032 assert moderation_user_list is not None
1033 assert len(moderation_user_list.users) == 3
1034 assert {user1.id, user2.id, user3.id}.issubset({user.id for user in moderation_user_list.users})
1036 # Test list endpoint returns same moderation list with same members not repeated
1037 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1038 assert len(listRes.moderation_lists) == 1
1039 assert listRes.moderation_lists[0].moderation_list_id == res.moderation_list_id
1040 assert len(listRes.moderation_lists[0].members) == 3
1041 assert {user1.id, user2.id, user3.id}.issubset({m.user_id for m in listRes.moderation_lists[0].members})
1043 # Test user can be in multiple moderation lists
1044 listRes3 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1045 assert len(listRes3.moderation_lists) == 2
1047 # Test adding users to an existing moderation list
1048 res2 = api.AddUsersToModerationUserList(
1049 admin_pb2.AddUsersToModerationUserListReq(
1050 users=[user4.username, user5.username], moderation_list_id=moderation_list_id
1051 ),
1052 )
1053 assert res2.moderation_list_id == moderation_list_id
1054 with session_scope() as session:
1055 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1056 assert len(moderation_user_list.users) == 3
1057 assert {user1.id, user4.id, user5.id}.issubset({user.id for user in moderation_user_list.users})
1059 # Test list user moderation lists endpoint returns the right moderation list
1060 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user5.username))
1061 assert len(listRes2.moderation_lists) == 1
1062 assert listRes2.moderation_lists[0].moderation_list_id == moderation_list_id
1063 assert len(listRes2.moderation_lists[0].members) == 3
1064 assert {user1.id, user4.id, user5.id}.issubset({m.user_id for m in listRes2.moderation_lists[0].members})
1067def test_RemoveUserFromModerationUserList(db):
1068 super_user, super_token = generate_user(is_superuser=True)
1069 user1, _ = generate_user()
1070 user2, _ = generate_user()
1071 user3, _ = generate_user()
1072 moderation_list_id = add_users_to_new_moderation_list([user1, user2])
1074 with real_admin_session(super_token) as api:
1075 # Test with non-existent user (should raise error)
1076 with pytest.raises(grpc.RpcError) as e:
1077 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user="nonexistent"))
1078 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1079 assert "Couldn't find that user." == e.value.details()
1081 # Test without providing moderation list id (should raise error)
1082 with pytest.raises(grpc.RpcError) as e:
1083 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username))
1084 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1085 assert "Missing moderation user list id." == e.value.details()
1087 # Test removing user that's not in the provided moderation list (should raise error)
1088 with pytest.raises(grpc.RpcError) as e:
1089 api.RemoveUserFromModerationUserList(
1090 admin_pb2.RemoveUserFromModerationUserListReq(
1091 user=user3.username, moderation_list_id=moderation_list_id
1092 )
1093 )
1094 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1095 assert "User is not in the moderation user list." == e.value.details()
1097 # Test successful removal
1098 api.RemoveUserFromModerationUserList(
1099 admin_pb2.RemoveUserFromModerationUserListReq(user=user1.username, moderation_list_id=moderation_list_id)
1100 )
1101 with session_scope() as session:
1102 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1103 assert user1.id not in {user.id for user in moderation_user_list.users}
1104 assert user2.id in {user.id for user in moderation_user_list.users}
1106 # Test list user moderation lists endpoint returns right number of moderation lists
1107 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1108 assert len(listRes.moderation_lists) == 0
1109 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1110 assert len(listRes2.moderation_lists) == 1
1112 # Test removing all users from moderation list should also delete the moderation list
1113 api.RemoveUserFromModerationUserList(
1114 admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username, moderation_list_id=moderation_list_id)
1115 )
1116 with session_scope() as session:
1117 assert session.get(ModerationUserList, moderation_list_id) is None
1120def test_admin_delete_account_url(db, email_collector: EmailCollector, push_collector: PushCollector):
1121 super_user, super_token = generate_user(is_superuser=True)
1123 user, token = generate_user()
1124 user_id = user.id
1126 with real_admin_session(super_token) as admin_api:
1127 url = admin_api.CreateAccountDeletionLink(
1128 admin_pb2.CreateAccountDeletionLinkReq(user=user.username)
1129 ).account_deletion_confirm_url
1131 assert push_collector.count_for_user(user_id) == 0
1133 with session_scope() as session:
1134 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
1135 token = token_o.token
1136 assert token_o.user.id == user_id
1137 assert url == f"http://localhost:3000/delete-account?token={token}"
1139 with auth_api_session() as (auth_api, metadata_interceptor):
1140 auth_api.ConfirmDeleteAccount(
1141 auth_pb2.ConfirmDeleteAccountReq(
1142 token=token,
1143 )
1144 )
1146 push = push_collector.pop_for_user(user_id, last=True)
1147 assert push.content.title == "Account deleted"
1148 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
1149 email_collector.pop_for_recipient(user.email, last=True)
1152def test_AccessStats(db):
1153 super_user, super_token = generate_user(is_superuser=True)
1154 normal_user, normal_token = generate_user()
1156 # Insert UserActivity rows: a couple inside the default 90-day window, one well
1157 # outside it, and one with NULL ip_address / user_agent. The INET column is
1158 # returned by psycopg3 as an IPv4Address/IPv6Address object, which used to
1159 # crash the proto string assignment.
1160 in_window_1 = now() - timedelta(days=1)
1161 in_window_2 = now() - timedelta(days=10)
1162 out_of_window = now() - timedelta(days=200)
1163 with session_scope() as session:
1164 session.add(
1165 UserActivity(
1166 user_id=normal_user.id, period=in_window_1, ip_address="1.2.3.4", user_agent="ua-a", api_calls=5
1167 )
1168 )
1169 session.add(
1170 UserActivity(
1171 user_id=normal_user.id, period=in_window_2, ip_address="2001:db8::1", user_agent="ua-b", api_calls=3
1172 )
1173 )
1174 session.add(
1175 UserActivity(
1176 user_id=normal_user.id, period=out_of_window, ip_address="9.9.9.9", user_agent="ua-old", api_calls=99
1177 )
1178 )
1179 session.add(UserActivity(user_id=normal_user.id, period=in_window_1, api_calls=1))
1181 with real_admin_session(super_token) as api:
1182 res = api.AccessStats(admin_pb2.AccessStatsReq(user=normal_user.username))
1184 by_ip = {s.ip_address: s for s in res.stats}
1185 assert "1.2.3.4" in by_ip
1186 assert by_ip["1.2.3.4"].api_call_count == 5
1187 assert by_ip["1.2.3.4"].user_agent == "ua-a"
1188 assert "2001:db8::1" in by_ip
1189 assert by_ip["2001:db8::1"].api_call_count == 3
1190 # NULL ip_address row produces an empty-string ip_address in the proto
1191 assert "" in by_ip
1192 assert by_ip[""].api_call_count == 1
1193 # out-of-window row is excluded by the 90-day default
1194 assert "9.9.9.9" not in by_ip
1196 # explicit end_time should bound the upper end of the window (regression: was >=)
1197 with real_admin_session(super_token) as api:
1198 res = api.AccessStats(
1199 admin_pb2.AccessStatsReq(
1200 user=normal_user.username,
1201 start_time=Timestamp_from_datetime(now() - timedelta(days=5)),
1202 end_time=Timestamp_from_datetime(now()),
1203 )
1204 )
1205 ips = {s.ip_address for s in res.stats}
1206 assert ips == {"1.2.3.4", ""}
1209def test_SetLastDonated(db):
1210 super_user, super_token = generate_user(is_superuser=True)
1211 normal_user, normal_token = generate_user(last_donated=None)
1213 with real_admin_session(super_token) as api:
1214 # user starts with no last_donated
1215 with session_scope() as session:
1216 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1217 assert user.last_donated is None
1219 # can set last_donated
1220 donation_time = now() - timedelta(days=30)
1221 res = api.SetLastDonated(
1222 admin_pb2.SetLastDonatedReq(
1223 user=normal_user.username,
1224 last_donated=Timestamp_from_datetime(donation_time),
1225 )
1226 )
1228 with session_scope() as session:
1229 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1230 assert user.last_donated is not None
1231 # check timestamp is close (within a second)
1232 assert abs((user.last_donated - donation_time).total_seconds()) < 1
1234 # can clear last_donated by not setting the field
1235 res = api.SetLastDonated(admin_pb2.SetLastDonatedReq(user=normal_user.username))
1237 with session_scope() as session:
1238 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1239 assert user.last_donated is None
1241 # user not found
1242 with pytest.raises(grpc.RpcError) as e:
1243 api.SetLastDonated(admin_pb2.SetLastDonatedReq(user="nonexistent"))
1244 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1245 assert e.value.details() == "Couldn't find that user."
1248def test_admin_actions_level(db):
1249 super_user, super_token = generate_user(is_superuser=True)
1250 normal_user, _ = generate_user()
1252 with real_admin_session(super_token) as api:
1253 # Default level is NORMAL
1254 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="normal note"))
1255 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
1257 # Explicitly set to DEBUG
1258 res = api.AddAdminNote(
1259 admin_pb2.AddAdminNoteReq(
1260 user=normal_user.username,
1261 admin_note="debug note",
1262 level=admin_pb2.ADMIN_ACTION_LEVEL_DEBUG,
1263 )
1264 )
1265 assert len(res.admin_actions) == 2
1266 assert res.admin_actions[1].level == admin_pb2.ADMIN_ACTION_LEVEL_DEBUG
1268 # Explicitly set to HIGH
1269 res = api.AddAdminNote(
1270 admin_pb2.AddAdminNoteReq(
1271 user=normal_user.username,
1272 admin_note="high note",
1273 level=admin_pb2.ADMIN_ACTION_LEVEL_HIGH,
1274 )
1275 )
1276 assert len(res.admin_actions) == 3
1277 assert res.admin_actions[2].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
1279 # Explicitly set to TRACE
1280 res = api.AddAdminNote(
1281 admin_pb2.AddAdminNoteReq(
1282 user=normal_user.username,
1283 admin_note="trace note",
1284 level=admin_pb2.ADMIN_ACTION_LEVEL_TRACE,
1285 )
1286 )
1287 assert len(res.admin_actions) == 4
1288 assert res.admin_actions[3].level == admin_pb2.ADMIN_ACTION_LEVEL_TRACE
1291def test_admin_actions_on_mutations(db, push_collector: PushCollector):
1292 super_user, super_token = generate_user(is_superuser=True)
1293 normal_user, _ = generate_user()
1295 original_gender = normal_user.gender
1296 original_birthdate = normal_user.birthdate
1298 with real_admin_session(super_token) as api:
1299 # ChangeUserGender
1300 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
1301 assert any(
1302 a.action_type == "change_gender" and a.note == f"Changed from '{original_gender}' to 'Machine'"
1303 for a in res.admin_actions
1304 )
1306 # ChangeUserBirthdate
1307 res = api.ChangeUserBirthdate(
1308 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-01-01")
1309 )
1310 assert any(
1311 a.action_type == "change_birthdate" and a.note == f"Changed from {original_birthdate} to 1990-01-01"
1312 for a in res.admin_actions
1313 )
1315 # SetPassportSexGenderException
1316 res = api.SetPassportSexGenderException(
1317 admin_pb2.SetPassportSexGenderExceptionReq(user=normal_user.username, passport_sex_gender_exception=True)
1318 )
1319 assert any(
1320 a.action_type == "set_passport_sex_gender_exception" and a.note == "Changed from False to True"
1321 for a in res.admin_actions
1322 )
1324 # SendModNote with notify
1325 res = api.SendModNote(
1326 admin_pb2.SendModNoteReq(
1327 user=normal_user.username, content="Please update your profile", internal_id="test1"
1328 )
1329 )
1330 assert any(
1331 a.action_type == "send_mod_note" and a.note == "Notify user: Yes\n\nPlease update your profile"
1332 for a in res.admin_actions
1333 )
1335 # SendModNote with do_not_notify
1336 res = api.SendModNote(
1337 admin_pb2.SendModNoteReq(
1338 user=normal_user.username,
1339 content="Silent note",
1340 internal_id="test2",
1341 do_not_notify=True,
1342 )
1343 )
1344 assert any(
1345 a.action_type == "send_mod_note" and a.note == "Notify user: No\n\nSilent note" for a in res.admin_actions
1346 )
1348 # DeleteUser
1349 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
1350 assert any(a.action_type == "delete_user" for a in res.admin_actions)
1351 assert any(
1352 a.action_type == "delete_user" and a.level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH for a in res.admin_actions
1353 )
1355 # RecoverDeletedUser
1356 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
1357 assert any(a.action_type == "recover_user" for a in res.admin_actions)
1359 # MarkUserNeedsLocationUpdate
1360 res = api.MarkUserNeedsLocationUpdate(admin_pb2.MarkUserNeedsLocationUpdateReq(user=normal_user.username))
1361 assert any(
1362 a.action_type == "mark_needs_location_update" and a.note == "Marked user as needing location update"
1363 for a in res.admin_actions
1364 )
1366 # SetLastDonated
1367 res = api.SetLastDonated(
1368 admin_pb2.SetLastDonatedReq(
1369 user=normal_user.username,
1370 last_donated=Timestamp_from_datetime(now()),
1371 )
1372 )
1373 assert any(a.action_type == "set_last_donated" for a in res.admin_actions)
1376def test_create_admin_tag(db):
1377 super_user, super_token = generate_user(is_superuser=True)
1379 with real_admin_session(super_token) as api:
1380 res = api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1381 assert res.tag == "test-tag"
1382 assert res.admin_tag_id > 0
1385def test_create_admin_tag_duplicate(db):
1386 super_user, super_token = generate_user(is_superuser=True)
1388 with real_admin_session(super_token) as api:
1389 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1390 with pytest.raises(grpc.RpcError) as e:
1391 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1392 assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
1393 assert e.value.details() == "That admin tag already exists."
1396def test_create_admin_tag_empty(db):
1397 super_user, super_token = generate_user(is_superuser=True)
1399 with real_admin_session(super_token) as api:
1400 with pytest.raises(grpc.RpcError) as e:
1401 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=""))
1402 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1403 assert e.value.details() == "The admin tag cannot be empty."
1405 with pytest.raises(grpc.RpcError) as e:
1406 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=" "))
1407 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1408 assert e.value.details() == "The admin tag cannot be empty."
1411def test_list_admin_tags(db):
1412 super_user, super_token = generate_user(is_superuser=True)
1414 with real_admin_session(super_token) as api:
1415 # Empty initially
1416 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1417 assert len(res.tags) == 0
1419 # Add some tags
1420 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="bravo"))
1421 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="alpha"))
1423 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1424 assert len(res.tags) == 2
1425 # Ordered alphabetically
1426 assert res.tags[0].tag == "alpha"
1427 assert res.tags[1].tag == "bravo"
1430def test_add_admin_tag_to_user(db):
1431 super_user, super_token = generate_user(is_superuser=True)
1432 normal_user, _ = generate_user()
1434 with real_admin_session(super_token) as api:
1435 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1437 res = api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1438 assert "vip" in res.admin_tags
1439 assert any(a.action_type == "add_tag" and a.tag == "vip" for a in res.admin_actions)
1442def test_add_admin_tag_to_user_duplicate(db):
1443 super_user, super_token = generate_user(is_superuser=True)
1444 normal_user, _ = generate_user()
1446 with real_admin_session(super_token) as api:
1447 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1448 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1450 with pytest.raises(grpc.RpcError) as e:
1451 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1452 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1453 assert e.value.details() == "The user already has that admin tag."
1456def test_add_admin_tag_to_user_tag_not_found(db):
1457 super_user, super_token = generate_user(is_superuser=True)
1458 normal_user, _ = generate_user()
1460 with real_admin_session(super_token) as api:
1461 with pytest.raises(grpc.RpcError) as e:
1462 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="nonexistent"))
1463 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1464 assert e.value.details() == "Admin tag not found."
1467def test_remove_admin_tag_from_user(db):
1468 super_user, super_token = generate_user(is_superuser=True)
1469 normal_user, _ = generate_user()
1471 with real_admin_session(super_token) as api:
1472 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1473 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1475 res = api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1476 assert "vip" not in res.admin_tags
1477 assert any(a.action_type == "remove_tag" and a.tag == "vip" for a in res.admin_actions)
1480def test_remove_admin_tag_from_user_not_assigned(db):
1481 super_user, super_token = generate_user(is_superuser=True)
1482 normal_user, _ = generate_user()
1484 with real_admin_session(super_token) as api:
1485 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1487 with pytest.raises(grpc.RpcError) as e:
1488 api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1489 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1490 assert e.value.details() == "The user does not have that admin tag."
1493def test_search_users_by_admin_tag(db):
1494 super_user, super_token = generate_user(is_superuser=True)
1495 user1, _ = generate_user()
1496 user2, _ = generate_user()
1497 user3, _ = generate_user()
1499 with real_admin_session(super_token) as api:
1500 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1501 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="flagged"))
1503 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user1.username, tag="vip"))
1504 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="vip"))
1505 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="flagged"))
1507 # Search for users with "vip" tag
1508 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip"]))
1509 user_ids = {u.user_id for u in res.users}
1510 assert user1.id in user_ids
1511 assert user2.id in user_ids
1512 assert user3.id not in user_ids
1514 # Search for users with both "vip" AND "flagged" tags (AND logic)
1515 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip", "flagged"]))
1516 user_ids = {u.user_id for u in res.users}
1517 assert user2.id in user_ids
1518 assert user1.id not in user_ids
1520 # Search for non-existent tag returns no results
1521 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["nonexistent"]))
1522 assert len(res.users) == 0
1525def test_search_users_by_admin_note(db):
1526 super_user, super_token = generate_user(is_superuser=True)
1527 user1, _ = generate_user()
1528 user2, _ = generate_user()
1530 with real_admin_session(super_token) as api:
1531 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="suspicious activity"))
1532 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="normal user"))
1534 # Search by admin action log content (ilike)
1535 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_action_log="%suspicious%"))
1536 user_ids = {u.user_id for u in res.users}
1537 assert user1.id in user_ids
1538 assert user2.id not in user_ids
1541def test_ListAdminActions_empty(db):
1542 super_user, super_token = generate_user(is_superuser=True)
1544 with real_admin_session(super_token) as api:
1545 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1546 assert len(res.admin_actions) == 0
1547 assert res.next_page_token == ""
1550def test_ListAdminActions_returns_newest_first_with_target_info(db):
1551 super_user, super_token = generate_user(is_superuser=True)
1552 user1, _ = generate_user()
1553 user2, _ = generate_user()
1555 with real_admin_session(super_token) as api:
1556 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="first note"))
1557 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="second note"))
1558 api.BanUser(admin_pb2.BanUserReq(user=user1.username, admin_note="ban reason"))
1560 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1562 assert len(res.admin_actions) == 3
1563 # Newest first
1564 assert res.admin_actions[0].action_type == "ban"
1565 assert res.admin_actions[0].target_user_id == user1.id
1566 assert res.admin_actions[0].target_username == user1.username
1567 assert res.admin_actions[0].admin_user_id == super_user.id
1568 assert res.admin_actions[0].admin_username == super_user.username
1569 assert res.admin_actions[1].action_type == "note"
1570 assert res.admin_actions[1].target_user_id == user2.id
1571 assert res.admin_actions[2].action_type == "note"
1572 assert res.admin_actions[2].target_user_id == user1.id
1575def test_ListAdminActions_filter_by_admin_and_target(db):
1576 super1, super1_token = generate_user(is_superuser=True)
1577 super2, super2_token = generate_user(is_superuser=True)
1578 user1, _ = generate_user()
1579 user2, _ = generate_user()
1581 with real_admin_session(super1_token) as api:
1582 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super1 to user1"))
1583 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="from super1 to user2"))
1584 with real_admin_session(super2_token) as api:
1585 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super2 to user1"))
1587 with real_admin_session(super1_token) as api:
1588 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id))
1589 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super1 to user2"}
1591 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(target_user_id=user1.id))
1592 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super2 to user1"}
1594 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id, target_user_id=user1.id))
1595 assert [a.note for a in res.admin_actions] == ["from super1 to user1"]
1598def test_ListAdminActions_pagination(db):
1599 super_user, super_token = generate_user(is_superuser=True)
1600 user, _ = generate_user()
1602 with real_admin_session(super_token) as api:
1603 for i in range(3):
1604 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user.username, admin_note=f"note {i}"))
1606 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2))
1607 assert len(res.admin_actions) == 2
1608 assert res.next_page_token != ""
1609 first_page_notes = [a.note for a in res.admin_actions]
1611 res2 = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2, page_token=res.next_page_token))
1612 assert len(res2.admin_actions) == 1
1613 assert res2.next_page_token == ""
1615 all_notes = first_page_notes + [a.note for a in res2.admin_actions]
1616 assert set(all_notes) == {"note 0", "note 1", "note 2"}
1619def test_ListUserUploads(db):
1620 super_user, super_token = generate_user(is_superuser=True)
1621 user, _ = generate_user(complete_profile=False)
1622 other_user, _ = generate_user()
1624 with session_scope() as session:
1625 for i in range(3):
1626 session.add(
1627 Upload(
1628 key=f"key{i}",
1629 filename=f"photo{i}.jpg",
1630 creator_user_id=user.id,
1631 credit=f"credit {i}" if i == 0 else None,
1632 )
1633 )
1634 session.add(Upload(key="other_key", filename="other.jpg", creator_user_id=other_user.id))
1636 with real_admin_session(super_token) as api:
1637 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1639 assert len(res.uploads) == 3
1640 assert res.next_page_token == ""
1641 assert {u.filename for u in res.uploads} == {"photo0.jpg", "photo1.jpg", "photo2.jpg"}
1643 upload0 = next(u for u in res.uploads if u.key == "key0")
1644 assert upload0.credit == "credit 0"
1645 assert upload0.full_url.endswith("/img/full/photo0.jpg")
1646 assert upload0.thumbnail_url.endswith("/img/thumbnail/photo0.jpg")
1647 assert upload0.HasField("created")
1650def test_ListUserUploads_pagination(db):
1651 super_user, super_token = generate_user(is_superuser=True)
1652 user, _ = generate_user(complete_profile=False)
1654 with session_scope() as session:
1655 for i in range(3):
1656 session.add(Upload(key=f"key{i}", filename=f"photo{i}.jpg", creator_user_id=user.id))
1658 with real_admin_session(super_token) as api:
1659 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username, page_size=2))
1660 assert len(res.uploads) == 2
1661 assert res.next_page_token != ""
1662 first_page_keys = [u.key for u in res.uploads]
1664 res2 = api.ListUserUploads(
1665 admin_pb2.ListUserUploadsReq(user=user.username, page_size=2, page_token=res.next_page_token)
1666 )
1667 assert len(res2.uploads) == 1
1668 assert res2.next_page_token == ""
1670 all_keys = first_page_keys + [u.key for u in res2.uploads]
1671 assert set(all_keys) == {"key0", "key1", "key2"}
1674def test_ListUserUploads_uses(db):
1675 super_user, super_token = generate_user(is_superuser=True)
1676 user, _ = generate_user(complete_profile=False)
1678 with session_scope() as session:
1679 session.add(Upload(key="used_key", filename="used.jpg", creator_user_id=user.id))
1680 session.add(Upload(key="unused_key", filename="unused.jpg", creator_user_id=user.id))
1681 gallery = PhotoGallery(owner_user_id=user.id)
1682 session.add(gallery)
1683 session.flush()
1684 session.add(PhotoGalleryItem(gallery_id=gallery.id, upload_key="used_key", position=1.0))
1686 with real_admin_session(super_token) as api:
1687 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1689 uploads = {u.key: u for u in res.uploads}
1690 assert list(uploads["unused_key"].uses) == []
1692 used_uses = uploads["used_key"].uses
1693 assert len(used_uses) == 1
1694 assert used_uses[0].type == admin_pb2.UPLOAD_USE_TYPE_PROFILE_GALLERY_PHOTO_AVATAR
1695 assert used_uses[0].is_current
1696 assert used_uses[0].user_id == user.id
1699def test_ListUserUploads_not_found(db):
1700 super_user, super_token = generate_user(is_superuser=True)
1702 with real_admin_session(super_token) as api:
1703 with pytest.raises(grpc.RpcError) as e:
1704 api.ListUserUploads(admin_pb2.ListUserUploadsReq(user="nonexistent"))
1705 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1708# community invite feature tested in test_events.py
1709# SendBlogPostNotification tested in test_notifications.py
1710# MarkUserNeedsLocationUpdate tested in test_jail.py
1713def _ota_manifest(*, version, fingerprint, created_at="2026-05-31T00:00:00.000Z"):
1714 return {
1715 "id": f"id-{version}",
1716 "createdAt": created_at,
1717 "runtimeVersion": fingerprint,
1718 "launchAsset": {"key": "bundle", "url": f"https://cdn.testing.invalid/{version}/bundle.hbc"},
1719 "assets": [],
1720 "metadata": {},
1721 "extra": {},
1722 }
1725def _ota_signed_multipart(manifest):
1726 # Mimics the signed multipart body the CDN holds (signature header omitted; we only read the JSON).
1727 boundary = "COUCHERS_OTA_BOUNDARY"
1729 def part(name, body, content_type):
1730 return f'--{boundary}\r\ncontent-disposition: form-data; name="{name}"\r\ncontent-type: {content_type}\r\n\r\n{body}\r\n'
1732 body = (
1733 part("manifest", json.dumps(manifest), "application/json; charset=utf-8")
1734 + part("extensions", "{}", "application/json")
1735 + f"--{boundary}--\r\n"
1736 )
1737 return f"multipart/mixed; boundary={boundary}", body.encode()
1740def _patch_ota_cdn(manifests):
1741 # manifests: {version: manifest_dict}. URL is {cdn_root}/{version}/{platform}/manifest.
1742 def fake(url):
1743 version = url.split("/")[-3]
1744 if version not in manifests:
1745 return "multipart/mixed; boundary=COUCHERS_OTA_BOUNDARY", b""
1746 return _ota_signed_multipart(manifests[version])
1748 return patch("couchers.servicers.admin._fetch_signed_manifest", side_effect=fake)
1751def test_CreateOTAPackage(db):
1752 super_user, super_token = generate_user(is_superuser=True)
1754 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1755 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1756 res = api.CreateOTAPackage(
1757 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1758 )
1760 assert res.platform == admin_pb2.OTA_PLATFORM_IOS
1761 assert res.fingerprint == "ios-fp"
1762 assert res.version == "v1.3.1.aaaa"
1763 assert res.manifest_id == "id-v1.3.1.aaaa"
1764 assert res.banned is False
1765 assert res.live is True
1766 assert res.creator_user_id == super_user.id
1769def test_CreateOTAPackage_invalid(db):
1770 _, super_token = generate_user(is_superuser=True)
1772 manifests = {"v-incomplete": {"id": "x"}} # on the CDN but missing runtimeVersion / createdAt
1773 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1774 # missing version
1775 with pytest.raises(grpc.RpcError) as e:
1776 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1777 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1779 # nothing published at this version
1780 with pytest.raises(grpc.RpcError) as e:
1781 api.CreateOTAPackage(
1782 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-missing")
1783 )
1784 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1786 # manifest present but missing required fields
1787 with pytest.raises(grpc.RpcError) as e:
1788 api.CreateOTAPackage(
1789 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-incomplete")
1790 )
1791 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1794def test_CreateOTAPackage_rejects_duplicate_version(db):
1795 _, super_token = generate_user(is_superuser=True)
1797 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1798 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1799 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa"))
1800 with pytest.raises(grpc.RpcError) as e:
1801 api.CreateOTAPackage(
1802 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1803 )
1804 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1807def test_ListOTAPackages(db):
1808 _, super_token = generate_user(is_superuser=True)
1810 manifests = {
1811 "v1.3.1.ios": _ota_manifest(version="v1.3.1.ios", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1812 "v1.3.2.ios": _ota_manifest(version="v1.3.2.ios", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1813 "v1.3.2.android": _ota_manifest(
1814 version="v1.3.2.android", fingerprint="android-fp", created_at="2026-06-01T00:00:00.000Z"
1815 ),
1816 }
1817 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1818 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.ios"))
1819 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.ios"))
1820 api.CreateOTAPackage(
1821 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_ANDROID, version="v1.3.2.android")
1822 )
1824 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1825 # newest (by manifest createdAt) first
1826 assert [p.version for p in res.packages] == ["v1.3.2.android", "v1.3.2.ios", "v1.3.1.ios"]
1827 # only the newest per (platform, fingerprint) is live
1828 live = {p.version: p.live for p in res.packages}
1829 assert live == {"v1.3.2.android": True, "v1.3.2.ios": True, "v1.3.1.ios": False}
1831 ios = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1832 assert [p.version for p in ios.packages] == ["v1.3.2.ios", "v1.3.1.ios"]
1835def test_BanOTAPackage(db):
1836 super_user, super_token = generate_user(is_superuser=True)
1838 manifests = {
1839 "v1.3.1.good": _ota_manifest(
1840 version="v1.3.1.good", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"
1841 ),
1842 "v1.3.2.bad": _ota_manifest(version="v1.3.2.bad", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1843 }
1844 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1845 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.good"))
1846 second = api.CreateOTAPackage(
1847 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.bad")
1848 )
1849 assert second.live is True
1851 banned = api.BanOTAPackage(
1852 admin_pb2.BanOTAPackageReq(ota_package_id=second.ota_package_id, reason="bad bundle")
1853 )
1854 assert banned.banned is True
1855 assert banned.banned_reason == "bad bundle"
1856 assert banned.banned_by_user_id == super_user.id
1857 assert banned.live is False
1859 # banning the newest stops new check-ins getting it; the previous one becomes live again
1860 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(include_banned=True))
1861 live = {p.version: p.live for p in res.packages}
1862 assert live == {"v1.3.2.bad": False, "v1.3.1.good": True}
1864 # banned packages are excluded by default
1865 non_banned = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1866 assert [p.version for p in non_banned.packages] == ["v1.3.1.good"]
1869def test_BanOTAPackage_requires_reason(db):
1870 _, super_token = generate_user(is_superuser=True)
1872 manifests = {
1873 "v1.3.1": _ota_manifest(version="v1.3.1", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1874 }
1875 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1876 pkg = api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1"))
1877 with pytest.raises(grpc.RpcError) as e:
1878 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id))
1879 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1880 with pytest.raises(grpc.RpcError) as e:
1881 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id, reason=" "))
1882 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1885def test_BanOTAPackage_not_found(db):
1886 _, super_token = generate_user(is_superuser=True)
1888 with real_admin_session(super_token) as api:
1889 with pytest.raises(grpc.RpcError) as e:
1890 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=123456, reason="never mind"))
1891 assert e.value.code() == grpc.StatusCode.NOT_FOUND