Coverage for app/backend/src/tests/test_admin.py: 100%
1040 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-06-14 11:44 +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 Reference,
25 Upload,
26 User,
27 UserActivity,
28 UserSession,
29)
30from couchers.proto import (
31 account_pb2,
32 admin_pb2,
33 auth_pb2,
34 events_pb2,
35 references_pb2,
36 reporting_pb2,
37 requests_pb2,
38)
39from couchers.utils import Timestamp_from_datetime, now, parse_date
40from tests.fixtures.db import add_users_to_new_moderation_list, generate_user, make_friends
41from tests.fixtures.misc import EmailCollector, PushCollector
42from tests.fixtures.sessions import (
43 account_session,
44 auth_api_session,
45 events_session,
46 real_admin_session,
47 references_session,
48 reporting_session,
49 requests_session,
50)
51from tests.test_communities import create_community
52from tests.test_requests import valid_request_text
55@pytest.fixture(autouse=True)
56def _(testconfig):
57 pass
60def test_access_by_normal_user(db):
61 normal_user, normal_token = generate_user()
63 with real_admin_session(normal_token) as api:
64 # all requests to the admin servicer should break when done by a non-super_user
65 with pytest.raises(grpc.RpcError) as e:
66 api.GetUserDetails(
67 admin_pb2.GetUserDetailsReq(
68 user=str(normal_user.id),
69 )
70 )
71 assert e.value.code() == grpc.StatusCode.PERMISSION_DENIED
74def test_GetNonvisibleUserAccessLog(db):
75 super_user, super_token = generate_user(is_superuser=True)
76 target, _ = generate_user(username="target")
77 viewer, _ = generate_user(username="viewer")
79 with session_scope() as session:
80 session.add(
81 NonvisibleUserAccess(
82 access_type=NonvisibleUserAccessType.login_attempt,
83 target_user_id=target.id,
84 target_state=NonvisibleUserState.banned,
85 actor_user_id=target.id,
86 ip_address="1.2.3.4",
87 sofa="device-cookie",
88 )
89 )
90 session.add(
91 NonvisibleUserAccess(
92 access_type=NonvisibleUserAccessType.ghost_served,
93 target_user_id=target.id,
94 target_state=NonvisibleUserState.banned,
95 actor_user_id=viewer.id,
96 )
97 )
98 session.add(
99 NonvisibleUserAccess(
100 access_type=NonvisibleUserAccessType.ghost_served,
101 target_user_id=target.id,
102 target_state=NonvisibleUserState.banned,
103 actor_user_id=None,
104 )
105 )
107 with real_admin_session(super_token) as api:
108 res = api.GetNonvisibleUserAccessLog(admin_pb2.GetNonvisibleUserAccessLogReq(user="target"))
110 assert len(res.entries) == 3
111 for entry in res.entries:
112 assert entry.target_user_id == target.id
113 assert entry.target_state == admin_pb2.NONVISIBLE_USER_STATE_BANNED
115 login = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_LOGIN_ATTEMPT]
116 views = [e for e in res.entries if e.access_type == admin_pb2.NONVISIBLE_USER_ACCESS_TYPE_GHOST_SERVED]
117 assert len(login) == 1
118 assert len(views) == 2
120 assert login[0].actor_user_id.value == target.id
121 assert login[0].actor_username == "target"
122 assert login[0].ip_address == "1.2.3.4"
123 assert login[0].sofa == "device-cookie"
125 logged_in_view = [e for e in views if e.actor_username == "viewer"]
126 logged_out_view = [e for e in views if not e.actor_username]
127 assert len(logged_in_view) == 1
128 assert logged_in_view[0].actor_user_id.value == viewer.id
129 assert len(logged_out_view) == 1
130 assert not logged_out_view[0].HasField("actor_user_id")
133def test_GetUser(db):
134 super_user, super_token = generate_user(is_superuser=True)
135 normal_user, normal_token = generate_user()
137 with real_admin_session(super_token) as api:
138 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
139 assert res.user_id == normal_user.id
140 assert res.username == normal_user.username
142 with real_admin_session(super_token) as api:
143 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note="Testing banning"))
145 with real_admin_session(super_token) as api:
146 res = api.GetUser(admin_pb2.GetUserReq(user=str(normal_user.id)))
147 assert res.user_id == normal_user.id
148 assert res.username == normal_user.username
151def test_GetUserDetails(db):
152 super_user, super_token = generate_user(is_superuser=True)
153 normal_user, normal_token = generate_user()
155 with real_admin_session(super_token) as api:
156 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=str(normal_user.id)))
157 assert res.user_id == normal_user.id
158 assert res.username == normal_user.username
159 assert res.email == normal_user.email
160 assert res.gender == normal_user.gender
161 assert parse_date(res.birthdate) == normal_user.birthdate
162 assert not res.banned
163 assert not res.deleted
165 with real_admin_session(super_token) as api:
166 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
167 assert res.user_id == normal_user.id
168 assert res.username == normal_user.username
169 assert res.email == normal_user.email
170 assert res.gender == normal_user.gender
171 assert parse_date(res.birthdate) == normal_user.birthdate
172 assert not res.banned
173 assert not res.deleted
175 with real_admin_session(super_token) as api:
176 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.email))
177 assert res.user_id == normal_user.id
178 assert res.username == normal_user.username
179 assert res.email == normal_user.email
180 assert res.gender == normal_user.gender
181 assert parse_date(res.birthdate) == normal_user.birthdate
182 assert not res.banned
183 assert not res.deleted
186def test_ChangeUserGender(db, email_collector: EmailCollector, push_collector: PushCollector):
187 super_user, super_token = generate_user(is_superuser=True)
188 normal_user, normal_token = generate_user()
190 with real_admin_session(super_token) as api:
191 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
192 assert res.user_id == normal_user.id
193 assert res.username == normal_user.username
194 assert res.email == normal_user.email
195 assert res.gender == "Machine"
196 assert parse_date(res.birthdate) == normal_user.birthdate
197 assert not res.banned
198 assert not res.deleted
200 email = email_collector.pop_for_recipient(normal_user.email, last=True)
201 assert email.subject == "[TEST] Your gender was changed"
202 assert email.recipient == normal_user.email
203 assert "Machine" in email.plain
204 assert "Machine" in email.html
206 push = push_collector.pop_for_user(normal_user.id, last=True)
207 assert push.content.title == "Gender changed"
208 assert push.content.body == "An admin changed your gender to Machine."
211def test_ChangeUserBirthdate(db, email_collector: EmailCollector, push_collector: PushCollector):
212 super_user, super_token = generate_user(is_superuser=True)
213 normal_user, normal_token = generate_user(birthdate=date(year=2000, month=1, day=1))
215 with real_admin_session(super_token) as api:
216 res = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username))
217 assert parse_date(res.birthdate) == date(year=2000, month=1, day=1)
219 res = api.ChangeUserBirthdate(
220 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-05-25")
221 )
223 assert res.user_id == normal_user.id
224 assert res.username == normal_user.username
225 assert res.email == normal_user.email
226 assert res.birthdate == "1990-05-25"
227 assert res.gender == normal_user.gender
228 assert not res.banned
229 assert not res.deleted
231 email = email_collector.pop_for_recipient(normal_user.email, last=True)
232 assert email.subject == "[TEST] Your date of birth was changed"
233 assert email.recipient == normal_user.email
234 assert "1990" in email.plain
235 assert "1990" in email.html
237 push = push_collector.pop_for_user(normal_user.id, last=True)
238 assert push.content.title == "Birthdate changed"
239 assert push.content.body == "An admin changed your date of birth to May 25, 1990."
242def test_BanUser(db):
243 super_user, super_token = generate_user(is_superuser=True)
244 normal_user, _ = generate_user()
245 admin_note = "A good reason"
247 with real_admin_session(super_token) as api:
248 res = api.BanUser(admin_pb2.BanUserReq(user=normal_user.username, admin_note=admin_note))
249 assert res.user_id == normal_user.id
250 assert res.username == normal_user.username
251 assert res.email == normal_user.email
252 assert res.gender == normal_user.gender
253 assert parse_date(res.birthdate) == normal_user.birthdate
254 assert res.banned
255 assert not res.deleted
256 assert len(res.admin_actions) == 1
257 assert res.admin_actions[0].action_type == "ban"
258 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
259 assert res.admin_actions[0].note == admin_note
260 assert res.admin_actions[0].admin_user_id == super_user.id
261 assert res.admin_actions[0].admin_username == super_user.username
264def test_UnbanUser(db):
265 super_user, super_token = generate_user(is_superuser=True)
266 normal_user, _ = generate_user()
267 admin_note = "A good reason"
269 with real_admin_session(super_token) as api:
270 res = api.UnbanUser(admin_pb2.UnbanUserReq(user=normal_user.username, admin_note=admin_note))
271 assert res.user_id == normal_user.id
272 assert res.username == normal_user.username
273 assert res.email == normal_user.email
274 assert res.gender == normal_user.gender
275 assert parse_date(res.birthdate) == normal_user.birthdate
276 assert not res.banned
277 assert not res.deleted
278 assert len(res.admin_actions) == 1
279 assert res.admin_actions[0].action_type == "unban"
280 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
283def test_ShadowUser(db):
284 super_user, super_token = generate_user(is_superuser=True)
285 surfer, surfer_token = generate_user()
286 host, _ = generate_user()
287 admin_note = "Spammer"
289 # Create a host request from `surfer` and approve its moderation state to VISIBLE so we can verify the cascade
290 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
291 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
292 with requests_session(surfer_token) as api:
293 host_request_id = api.CreateHostRequest(
294 requests_pb2.CreateHostRequestReq(
295 host_user_id=host.id,
296 from_date=today_plus_2,
297 to_date=today_plus_3,
298 text=valid_request_text(),
299 )
300 ).host_request_id
301 with session_scope() as session:
302 state = session.execute(
303 select(ModerationState)
304 .where(ModerationState.object_type == ModerationObjectType.host_request)
305 .where(ModerationState.object_id == host_request_id)
306 ).scalar_one()
307 state.visibility = ModerationVisibility.visible
309 with real_admin_session(super_token) as api:
310 res = api.ShadowUser(admin_pb2.ShadowUserReq(user=surfer.username, admin_note=admin_note))
311 assert res.user_id == surfer.id
312 assert res.shadowed
313 assert not res.banned
314 assert not res.deleted
315 assert len(res.admin_actions) == 1
316 assert res.admin_actions[0].action_type == "shadow"
317 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
318 assert res.admin_actions[0].note == admin_note
320 # The previously-visible host request is now shadowed
321 with session_scope() as session:
322 state = session.execute(
323 select(ModerationState)
324 .where(ModerationState.object_type == ModerationObjectType.host_request)
325 .where(ModerationState.object_id == host_request_id)
326 ).scalar_one()
327 assert state.visibility == ModerationVisibility.shadowed
330def test_UnshadowUser(db):
331 super_user, super_token = generate_user(is_superuser=True)
332 surfer, surfer_token = generate_user()
333 host, _ = generate_user()
335 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
336 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
337 with requests_session(surfer_token) as api:
338 shadow_cascade_request_id = api.CreateHostRequest(
339 requests_pb2.CreateHostRequestReq(
340 host_user_id=host.id,
341 from_date=today_plus_2,
342 to_date=today_plus_3,
343 text=valid_request_text(),
344 )
345 ).host_request_id
346 admin_hidden_request_id = api.CreateHostRequest(
347 requests_pb2.CreateHostRequestReq(
348 host_user_id=host.id,
349 from_date=today_plus_2,
350 to_date=today_plus_3,
351 text=valid_request_text(),
352 )
353 ).host_request_id
355 with session_scope() as session:
356 session.execute(select(User).where(User.id == surfer.id)).scalar_one().shadowed_at = now()
357 session.execute(
358 select(ModerationState)
359 .where(ModerationState.object_type == ModerationObjectType.host_request)
360 .where(ModerationState.object_id == shadow_cascade_request_id)
361 ).scalar_one().visibility = ModerationVisibility.shadowed
362 session.execute(
363 select(ModerationState)
364 .where(ModerationState.object_type == ModerationObjectType.host_request)
365 .where(ModerationState.object_id == admin_hidden_request_id)
366 ).scalar_one().visibility = ModerationVisibility.hidden
368 with real_admin_session(super_token) as api:
369 res = api.UnshadowUser(admin_pb2.UnshadowUserReq(user=surfer.username, admin_note="rehabilitated"))
370 assert not res.shadowed
371 assert len(res.admin_actions) == 1
372 assert res.admin_actions[0].action_type == "unshadow"
373 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
375 with session_scope() as session:
376 assert (
377 session.execute(
378 select(ModerationState)
379 .where(ModerationState.object_type == ModerationObjectType.host_request)
380 .where(ModerationState.object_id == shadow_cascade_request_id)
381 )
382 .scalar_one()
383 .visibility
384 == ModerationVisibility.visible
385 )
386 assert (
387 session.execute(
388 select(ModerationState)
389 .where(ModerationState.object_type == ModerationObjectType.host_request)
390 .where(ModerationState.object_id == admin_hidden_request_id)
391 )
392 .scalar_one()
393 .visibility
394 == ModerationVisibility.hidden
395 )
398def test_ShadowUser_blank_note(db):
399 super_user, super_token = generate_user(is_superuser=True)
400 normal_user, _ = generate_user()
402 with real_admin_session(super_token) as api:
403 with pytest.raises(grpc.RpcError) as e:
404 api.ShadowUser(admin_pb2.ShadowUserReq(user=normal_user.username, admin_note=" \t "))
405 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
408def test_AddAdminNote(db):
409 super_user, super_token = generate_user(is_superuser=True)
410 normal_user, _ = generate_user()
411 admin_note1 = "User reported strange behavior"
412 admin_note2 = "Insert private information here"
414 with real_admin_session(super_token) as api:
415 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note1))
416 assert res.user_id == normal_user.id
417 assert res.username == normal_user.username
418 assert res.email == normal_user.email
419 assert res.gender == normal_user.gender
420 assert parse_date(res.birthdate) == normal_user.birthdate
421 assert not res.banned
422 assert not res.deleted
423 assert len(res.admin_actions) == 1
424 assert res.admin_actions[0].action_type == "note"
425 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
426 assert res.admin_actions[0].note == admin_note1
428 with real_admin_session(super_token) as api:
429 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=admin_note2))
430 assert len(res.admin_actions) == 2
431 assert res.admin_actions[0].note == admin_note1
432 assert res.admin_actions[1].note == admin_note2
435def test_AddAdminNote_blank(db):
436 super_user, super_token = generate_user(is_superuser=True)
437 normal_user, _ = generate_user()
438 empty_admin_note = " \t \n "
440 with real_admin_session(super_token) as api:
441 with pytest.raises(grpc.RpcError) as e:
442 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note=empty_admin_note))
443 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
444 assert e.value.details() == "Provide exactly one of admin_note or data."
447def test_AddAdminNote_data(db):
448 super_user, super_token = generate_user(is_superuser=True)
449 normal_user, _ = generate_user()
450 payload = '{"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}'
452 with real_admin_session(super_token) as api:
453 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data=payload))
454 assert len(res.admin_actions) == 1
455 assert res.admin_actions[0].action_type == "note"
456 assert res.admin_actions[0].note == ""
457 assert json.loads(res.admin_actions[0].data) == {"kind": "flag", "score": 0.87, "reasons": ["spam", "burst"]}
460def test_AddAdminNote_both_note_and_data(db):
461 super_user, super_token = generate_user(is_superuser=True)
462 normal_user, _ = generate_user()
464 with real_admin_session(super_token) as api:
465 with pytest.raises(grpc.RpcError) as e:
466 api.AddAdminNote(
467 admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="note text", data='{"x": 1}')
468 )
469 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
470 assert e.value.details() == "Provide exactly one of admin_note or data."
473def test_AddAdminNote_neither(db):
474 super_user, super_token = generate_user(is_superuser=True)
475 normal_user, _ = generate_user()
477 with real_admin_session(super_token) as api:
478 with pytest.raises(grpc.RpcError) as e:
479 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username))
480 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
481 assert e.value.details() == "Provide exactly one of admin_note or data."
484def test_AddAdminNote_invalid_json(db):
485 super_user, super_token = generate_user(is_superuser=True)
486 normal_user, _ = generate_user()
488 with real_admin_session(super_token) as api:
489 with pytest.raises(grpc.RpcError) as e:
490 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, data="{not valid json"))
491 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
492 assert e.value.details() == "The admin note data must be valid JSON."
495def test_admin_content_reports(db):
496 super_user, super_token = generate_user(is_superuser=True)
497 normal_user, token = generate_user()
498 bad_user1, _ = generate_user()
499 bad_user2, _ = generate_user()
501 with reporting_session(token) as api:
502 api.Report(
503 reporting_pb2.ReportReq(
504 reason="spam",
505 description="r1",
506 content_ref="comment/123",
507 author_user=bad_user1.username,
508 user_agent="n/a",
509 page="https://couchers.org/comment/123",
510 )
511 )
512 api.Report(
513 reporting_pb2.ReportReq(
514 reason="spam",
515 description="r2",
516 content_ref="comment/124",
517 author_user=bad_user2.username,
518 user_agent="n/a",
519 page="https://couchers.org/comment/124",
520 )
521 )
522 api.Report(
523 reporting_pb2.ReportReq(
524 reason="something else",
525 description="r3",
526 content_ref="page/321",
527 author_user=bad_user1.username,
528 user_agent="n/a",
529 page="https://couchers.org/page/321",
530 )
531 )
533 with session_scope() as session:
534 id_by_description: dict[str, int] = dict(
535 session.execute(select(ContentReport.description, ContentReport.id)).all() # type: ignore[arg-type]
536 )
538 with real_admin_session(super_token) as api:
539 with pytest.raises(grpc.RpcError) as e:
540 api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=-1))
541 assert e.value.code() == grpc.StatusCode.NOT_FOUND
542 assert e.value.details() == "Content report not found."
544 res = api.GetContentReport(admin_pb2.GetContentReportReq(content_report_id=id_by_description["r2"]))
545 rep = res.content_report
546 assert rep.content_report_id == id_by_description["r2"]
547 assert rep.reporting_user_id == normal_user.id
548 assert rep.author_user_id == bad_user2.id
549 assert rep.reason == "spam"
550 assert rep.description == "r2"
551 assert rep.content_ref == "comment/124"
552 assert rep.user_agent == "n/a"
553 assert rep.page == "https://couchers.org/comment/124"
555 res = api.GetContentReportsForAuthor(admin_pb2.GetContentReportsForAuthorReq(user=bad_user1.username))
556 assert res.content_reports[0].content_report_id == id_by_description["r3"]
557 assert res.content_reports[1].content_report_id == id_by_description["r1"]
560def test_DeleteUser(db):
561 super_user, super_token = generate_user(is_superuser=True)
562 normal_user, normal_token = generate_user()
564 with real_admin_session(super_token) as api:
565 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
566 assert res.user_id == normal_user.id
567 assert res.username == normal_user.username
568 assert res.email == normal_user.email
569 assert res.gender == normal_user.gender
570 assert parse_date(res.birthdate) == normal_user.birthdate
571 assert not res.banned
572 assert res.deleted
574 with real_admin_session(super_token) as api:
575 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
576 assert res.user_id == normal_user.id
577 assert res.username == normal_user.username
578 assert res.email == normal_user.email
579 assert res.gender == normal_user.gender
580 assert parse_date(res.birthdate) == normal_user.birthdate
581 assert not res.banned
582 assert not res.deleted
585def test_RecoverDeletedUser_after_user_initiated_deletion(db, push_collector: PushCollector):
586 """
587 When a user deletes their account through the normal flow (ConfirmDeleteAccount),
588 undelete_token and undelete_until are set. The admin RecoverDeletedUser must clear
589 these fields to satisfy the undelete_nullity database constraint.
590 """
591 super_user, super_token = generate_user(is_superuser=True)
592 normal_user, normal_token = generate_user()
593 user_id = normal_user.id
595 # User initiates account deletion
596 with account_session(normal_token) as account:
597 account.DeleteAccount(account_pb2.DeleteAccountReq(confirm=True))
599 # Get the deletion confirmation token
600 with session_scope() as session:
601 deletion_token = session.execute(select(AccountDeletionToken)).scalar_one().token
603 # User confirms account deletion (this sets undelete_token and undelete_until)
604 with auth_api_session() as (auth_api, metadata_interceptor):
605 auth_api.ConfirmDeleteAccount(auth_pb2.ConfirmDeleteAccountReq(token=deletion_token))
607 # Verify the user is deleted and has undelete fields set
608 with session_scope() as session:
609 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
610 assert user.deleted_at is not None
611 assert user.undelete_token is not None
612 assert user.undelete_until is not None
614 # Admin recovers the user
615 with real_admin_session(super_token) as api:
616 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
617 assert res.user_id == user_id
618 assert not res.deleted
620 # Verify undelete fields are cleared
621 with session_scope() as session:
622 user = session.execute(select(User).where(User.id == user_id)).scalar_one()
623 assert user.deleted_at is None
624 assert user.undelete_token is None
625 assert user.undelete_until is None
628def test_CreateApiKey(db, email_collector: EmailCollector, push_collector: PushCollector):
629 with session_scope() as session:
630 super_user, super_token = generate_user(is_superuser=True)
631 normal_user, normal_token = generate_user()
633 assert (
634 session.execute(
635 select(func.count())
636 .select_from(UserSession)
637 .where(UserSession.is_api_key == True)
638 .where(UserSession.user_id == normal_user.id)
639 ).scalar_one()
640 == 0
641 )
643 with real_admin_session(super_token) as api:
644 res = api.CreateApiKey(admin_pb2.CreateApiKeyReq(user=normal_user.username))
646 email = email_collector.pop_for_recipient(normal_user.email, last=True)
647 assert email.subject == "[TEST] Your API key for Couchers.org"
649 with session_scope() as session:
650 token = session.execute(
651 select(UserSession.token)
652 .where(UserSession.is_valid)
653 .where(UserSession.is_api_key == True)
654 .where(UserSession.user_id == normal_user.id)
655 ).scalar_one()
657 assert token in email.plain
658 assert token in email.html
660 assert email.recipient == normal_user.email
661 assert "api key" in email.subject.lower()
662 unique_string = "We've issued you with the following API key:"
663 assert unique_string in email.plain
664 assert unique_string in email.html
665 assert "support@couchers.org" in email.plain
666 assert "support@couchers.org" in email.html
668 push = push_collector.pop_for_user(normal_user.id, last=True)
669 assert push.content.title == "API key created"
670 assert push.content.body == "Details were sent to you via email."
673def test_GetChats(db):
674 super_user, super_token = generate_user(is_superuser=True)
675 normal_user, normal_token = generate_user()
677 with real_admin_session(super_token) as api:
678 res = api.GetChats(admin_pb2.GetChatsReq(user=normal_user.username))
679 assert res.user.user_id == normal_user.id
680 assert res.user.username == normal_user.username
681 assert res.user.name == normal_user.name
682 # New user should have no chats
683 assert len(res.host_requests) == 0
684 assert len(res.group_chats) == 0
687def test_badges(db, email_collector: EmailCollector, push_collector: PushCollector):
688 super_user, super_token = generate_user(is_superuser=True)
689 normal_user, normal_token = generate_user()
691 with real_admin_session(super_token) as api:
692 # can add a badge
693 assert "swagster" not in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
694 res = api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
695 assert "swagster" in res.badges
697 # badge emails are disabled by default
698 assert email_collector.count_for_recipient(normal_user.email) == 0
700 push = push_collector.pop_for_user(normal_user.id, last=True)
701 assert push.content.title == "New profile badge: Swagster"
702 assert push.content.body == "The Swagster badge was added to your profile."
704 # can't add/edit special tags
705 with pytest.raises(grpc.RpcError) as e:
706 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="founder"))
707 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
708 assert e.value.details() == "Admins cannot edit that badge."
710 # double add badge
711 with pytest.raises(grpc.RpcError) as e:
712 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="swagster"))
713 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
714 assert e.value.details() == "The user already has that badge."
716 # can remove badge
717 assert "swagster" in api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=normal_user.username)).badges
718 res = api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
719 assert "swagster" not in res.badges
721 # badge emails are disabled by default
722 assert email_collector.count_for_recipient(normal_user.email) == 0
724 push = push_collector.pop_for_user(normal_user.id, last=True)
725 assert push.content.title == "Profile badge removed"
726 assert push.content.body == "The Swagster badge was removed from your profile."
728 # not found on user
729 with pytest.raises(grpc.RpcError) as e:
730 api.RemoveBadge(admin_pb2.RemoveBadgeReq(user=normal_user.username, badge_id="swagster"))
731 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
732 assert e.value.details() == "The user does not have that badge."
734 # not found in general
735 with pytest.raises(grpc.RpcError) as e:
736 api.AddBadge(admin_pb2.AddBadgeReq(user=normal_user.username, badge_id="nonexistentbadge"))
737 assert e.value.code() == grpc.StatusCode.NOT_FOUND
738 assert e.value.details() == "Badge not found."
741def test_DeleteEvent(db):
742 super_user, super_token = generate_user(is_superuser=True)
743 normal_user, normal_token = generate_user()
745 with session_scope() as session:
746 create_community(session, 0, 2, "Community", [normal_user], [], None)
748 start_time = now() + timedelta(hours=2)
749 end_time = start_time + timedelta(hours=3)
750 with events_session(normal_token) as api:
751 res = api.CreateEvent(
752 events_pb2.CreateEventReq(
753 title="Dummy Title",
754 content="Dummy content.",
755 photo_key=None,
756 offline_information=events_pb2.OfflineEventInformation(
757 address="Near Null Island",
758 lat=0.1,
759 lng=0.2,
760 ),
761 start_time=Timestamp_from_datetime(start_time),
762 end_time=Timestamp_from_datetime(end_time),
763 timezone="UTC",
764 )
765 )
766 event_id = res.event_id
767 assert not res.is_deleted
769 with session_scope() as session:
770 with real_admin_session(super_token) as api:
771 api.DeleteEvent(
772 admin_pb2.DeleteEventReq(
773 event_id=event_id,
774 )
775 )
776 occurrence = session.get_one(EventOccurrence, ident=event_id)
777 assert occurrence.is_deleted
780def test_ListUserIds(db):
781 super_user, super_token = generate_user(is_superuser=True)
782 normal_user, normal_token = generate_user()
784 with real_admin_session(super_token) as api:
785 res = api.ListUserIds(
786 admin_pb2.ListUserIdsReq(
787 start_time=Timestamp_from_datetime(datetime(2000, 1, 1, tzinfo=UTC)),
788 end_time=Timestamp_from_datetime(now()),
789 )
790 )
791 assert len(res.user_ids) == 2
792 assert sorted(res.user_ids) == sorted([super_user.id, normal_user.id])
794 with real_admin_session(super_token) as api:
795 res = api.ListUserIds(
796 admin_pb2.ListUserIdsReq(start_time=Timestamp_from_datetime(now()), end_time=Timestamp_from_datetime(now()))
797 )
798 assert res.user_ids == []
801def test_EditReferenceText(db):
802 super_user, super_token = generate_user(is_superuser=True)
803 test_new_text = "New Text"
805 user1, user1_token = generate_user()
806 user2, user2_token = generate_user()
807 make_friends(user1, user2)
809 with session_scope() as session:
810 with references_session(user1_token) as api:
811 reference = api.WriteFriendReference(
812 references_pb2.WriteFriendReferenceReq(
813 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
814 )
815 )
817 with real_admin_session(super_token) as admin_api:
818 admin_api.EditReferenceText(
819 admin_pb2.EditReferenceTextReq(reference_id=reference.reference_id, new_text=test_new_text)
820 )
822 session.expire_all()
824 modified_reference = session.execute(
825 select(Reference).where(Reference.id == reference.reference_id)
826 ).scalar_one()
827 assert modified_reference.text == test_new_text
830def test_DeleteReference_deprecated(db):
831 """DeleteReference is deprecated; admins should hide via UMS instead."""
832 super_user, super_token = generate_user(is_superuser=True)
834 user1, user1_token = generate_user()
835 user2, user2_token = generate_user()
836 make_friends(user1, user2)
838 with references_session(user1_token) as api:
839 reference = api.WriteFriendReference(
840 references_pb2.WriteFriendReferenceReq(
841 to_user_id=user2.id, text="Old Text", private_text="", was_appropriate=True, rating=1
842 )
843 )
845 with real_admin_session(super_token) as admin_api:
846 with pytest.raises(grpc.RpcError) as e:
847 admin_api.DeleteReference(admin_pb2.DeleteReferenceReq(reference_id=reference.reference_id))
848 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
851def test_GetUserReferences(db):
852 super_user, super_token = generate_user(is_superuser=True)
854 user1, user1_token = generate_user()
855 user2, user2_token = generate_user()
856 user3, user3_token = generate_user()
857 make_friends(user1, user2)
858 make_friends(user1, user3)
859 make_friends(user2, user3)
861 # user1 writes reference about user2
862 with references_session(user1_token) as api:
863 ref1 = api.WriteFriendReference(
864 references_pb2.WriteFriendReferenceReq(
865 to_user_id=user2.id,
866 text="Reference from user1 to user2",
867 private_text="",
868 was_appropriate=True,
869 rating=1,
870 )
871 )
873 # user2 writes reference about user1
874 with references_session(user2_token) as api:
875 ref2 = api.WriteFriendReference(
876 references_pb2.WriteFriendReferenceReq(
877 to_user_id=user1.id,
878 text="Reference from user2 to user1",
879 private_text="Private note",
880 was_appropriate=True,
881 rating=0.8,
882 )
883 )
885 # user3 writes reference about user1
886 with references_session(user3_token) as api:
887 ref3 = api.WriteFriendReference(
888 references_pb2.WriteFriendReferenceReq(
889 to_user_id=user1.id,
890 text="Reference from user3 to user1",
891 private_text="",
892 was_appropriate=False,
893 rating=0.5,
894 )
895 )
897 # Test GetUserReferences for user1 (admin view shows everything regardless of UMS state).
898 with real_admin_session(super_token) as admin_api:
899 res = admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user=user1.username))
901 # user1 wrote 1 reference
902 assert len(res.references_from) == 1
903 assert res.references_from[0].reference_id == ref1.reference_id
904 assert res.references_from[0].from_user_id == user1.id
905 assert res.references_from[0].to_user_id == user2.id
906 assert res.references_from[0].text == "Reference from user1 to user2"
908 # user1 received 2 references
909 assert len(res.references_to) == 2
910 # Ordered by id descending, so ref3 comes first
911 assert res.references_to[0].reference_id == ref3.reference_id
912 assert res.references_to[0].was_appropriate is False
914 assert res.references_to[1].reference_id == ref2.reference_id
915 assert res.references_to[1].private_text == "Private note"
916 assert res.references_to[1].rating == 0.8
919def test_GetUserReferences_not_found(db):
920 super_user, super_token = generate_user(is_superuser=True)
922 with real_admin_session(super_token) as admin_api:
923 with pytest.raises(grpc.RpcError) as e:
924 admin_api.GetUserReferences(admin_pb2.GetUserReferencesReq(user="nonexistent"))
925 assert e.value.code() == grpc.StatusCode.NOT_FOUND
928def test_GetFriendRequests(db):
929 super_user, super_token = generate_user(is_superuser=True)
931 user1, _ = generate_user()
932 user2, _ = generate_user()
933 user3, _ = generate_user()
934 user4, _ = generate_user()
936 # Create a mix of friend requests directly so we control the state
937 def _add_friend_request(from_user_id, to_user_id, status, visibility, time_responded=None):
938 with session_scope() as session:
939 mod_state = ModerationState(
940 object_type=ModerationObjectType.friend_request,
941 object_id=0,
942 visibility=visibility,
943 )
944 session.add(mod_state)
945 session.flush()
946 rel = FriendRelationship(
947 from_user_id=from_user_id,
948 to_user_id=to_user_id,
949 status=status,
950 moderation_state_id=mod_state.id,
951 time_responded=time_responded,
952 )
953 session.add(rel)
954 session.flush()
955 mod_state.object_id = rel.id
957 # user1 -> user2: pending, shadowed
958 _add_friend_request(user1.id, user2.id, FriendStatus.pending, ModerationVisibility.shadowed)
959 # user1 -> user3: accepted, visible
960 _add_friend_request(user1.id, user3.id, FriendStatus.accepted, ModerationVisibility.visible, time_responded=now())
961 # user4 -> user1: rejected, visible
962 _add_friend_request(user4.id, user1.id, FriendStatus.rejected, ModerationVisibility.visible, time_responded=now())
964 with real_admin_session(super_token) as admin_api:
965 res = admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user=user1.username))
967 # user1 sent two: to user2 (pending) and to user3 (accepted), ordered by id desc
968 assert len(res.sent) == 2
969 assert res.sent[0].from_user.user_id == user1.id
970 assert res.sent[0].to_user.user_id == user3.id
971 assert res.sent[0].status == "accepted"
972 assert res.sent[0].HasField("time_responded")
973 assert res.sent[0].moderation_visibility == "visible"
975 assert res.sent[1].from_user.user_id == user1.id
976 assert res.sent[1].to_user.user_id == user2.id
977 assert res.sent[1].status == "pending"
978 assert not res.sent[1].HasField("time_responded")
979 assert res.sent[1].moderation_visibility == "shadowed"
981 # user1 received one: from user4 (rejected)
982 assert len(res.received) == 1
983 assert res.received[0].from_user.user_id == user4.id
984 assert res.received[0].to_user.user_id == user1.id
985 assert res.received[0].status == "rejected"
988def test_GetFriendRequests_not_found(db):
989 super_user, super_token = generate_user(is_superuser=True)
991 with real_admin_session(super_token) as admin_api:
992 with pytest.raises(grpc.RpcError) as e:
993 admin_api.GetFriendRequests(admin_pb2.GetFriendRequestsReq(user="nonexistent"))
994 assert e.value.code() == grpc.StatusCode.NOT_FOUND
997def test_AddUsersToModerationUserList(db):
998 super_user, super_token = generate_user(is_superuser=True)
999 user1, _ = generate_user()
1000 user2, _ = generate_user()
1001 user3, _ = generate_user()
1002 user4, _ = generate_user()
1003 user5, _ = generate_user()
1004 moderation_list_id = add_users_to_new_moderation_list([user1])
1006 with session_scope() as session:
1007 with real_admin_session(super_token) as api:
1008 # Test adding users to a non-existent moderation list (should raise an error)
1009 with pytest.raises(grpc.RpcError) as e:
1010 api.AddUsersToModerationUserList(
1011 admin_pb2.AddUsersToModerationUserListReq(users=[user2.username], moderation_list_id=999),
1012 )
1013 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1014 assert "Moderation user list not found." == e.value.details()
1016 # Test with non-existent user (should raise an error)
1017 with pytest.raises(grpc.RpcError) as e:
1018 api.AddUsersToModerationUserList(
1019 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, "nonexistent"]),
1020 )
1021 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1022 assert "Couldn't find that user." == e.value.details()
1024 # Test successful creation of new moderation list (no moderation_list_id provided)
1025 res = api.AddUsersToModerationUserList(
1026 admin_pb2.AddUsersToModerationUserListReq(users=[user1.username, user2.username, user3.username]),
1027 )
1028 assert res.moderation_list_id > 0
1029 with session_scope() as session:
1030 moderation_user_list = session.get(ModerationUserList, res.moderation_list_id)
1031 assert moderation_user_list is not None
1032 assert len(moderation_user_list.users) == 3
1033 assert {user1.id, user2.id, user3.id}.issubset({user.id for user in moderation_user_list.users})
1035 # Test list endpoint returns same moderation list with same members not repeated
1036 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1037 assert len(listRes.moderation_lists) == 1
1038 assert listRes.moderation_lists[0].moderation_list_id == res.moderation_list_id
1039 assert len(listRes.moderation_lists[0].members) == 3
1040 assert {user1.id, user2.id, user3.id}.issubset({m.user_id for m in listRes.moderation_lists[0].members})
1042 # Test user can be in multiple moderation lists
1043 listRes3 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1044 assert len(listRes3.moderation_lists) == 2
1046 # Test adding users to an existing moderation list
1047 res2 = api.AddUsersToModerationUserList(
1048 admin_pb2.AddUsersToModerationUserListReq(
1049 users=[user4.username, user5.username], moderation_list_id=moderation_list_id
1050 ),
1051 )
1052 assert res2.moderation_list_id == moderation_list_id
1053 with session_scope() as session:
1054 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1055 assert len(moderation_user_list.users) == 3
1056 assert {user1.id, user4.id, user5.id}.issubset({user.id for user in moderation_user_list.users})
1058 # Test list user moderation lists endpoint returns the right moderation list
1059 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user5.username))
1060 assert len(listRes2.moderation_lists) == 1
1061 assert listRes2.moderation_lists[0].moderation_list_id == moderation_list_id
1062 assert len(listRes2.moderation_lists[0].members) == 3
1063 assert {user1.id, user4.id, user5.id}.issubset({m.user_id for m in listRes2.moderation_lists[0].members})
1066def test_RemoveUserFromModerationUserList(db):
1067 super_user, super_token = generate_user(is_superuser=True)
1068 user1, _ = generate_user()
1069 user2, _ = generate_user()
1070 user3, _ = generate_user()
1071 moderation_list_id = add_users_to_new_moderation_list([user1, user2])
1073 with real_admin_session(super_token) as api:
1074 # Test with non-existent user (should raise error)
1075 with pytest.raises(grpc.RpcError) as e:
1076 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user="nonexistent"))
1077 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1078 assert "Couldn't find that user." == e.value.details()
1080 # Test without providing moderation list id (should raise error)
1081 with pytest.raises(grpc.RpcError) as e:
1082 api.RemoveUserFromModerationUserList(admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username))
1083 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1084 assert "Missing moderation user list id." == e.value.details()
1086 # Test removing user that's not in the provided moderation list (should raise error)
1087 with pytest.raises(grpc.RpcError) as e:
1088 api.RemoveUserFromModerationUserList(
1089 admin_pb2.RemoveUserFromModerationUserListReq(
1090 user=user3.username, moderation_list_id=moderation_list_id
1091 )
1092 )
1093 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1094 assert "User is not in the moderation user list." == e.value.details()
1096 # Test successful removal
1097 api.RemoveUserFromModerationUserList(
1098 admin_pb2.RemoveUserFromModerationUserListReq(user=user1.username, moderation_list_id=moderation_list_id)
1099 )
1100 with session_scope() as session:
1101 moderation_user_list = session.get_one(ModerationUserList, moderation_list_id)
1102 assert user1.id not in {user.id for user in moderation_user_list.users}
1103 assert user2.id in {user.id for user in moderation_user_list.users}
1105 # Test list user moderation lists endpoint returns right number of moderation lists
1106 listRes = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user1.username))
1107 assert len(listRes.moderation_lists) == 0
1108 listRes2 = api.ListModerationUserLists(admin_pb2.ListModerationUserListsReq(user=user2.username))
1109 assert len(listRes2.moderation_lists) == 1
1111 # Test removing all users from moderation list should also delete the moderation list
1112 api.RemoveUserFromModerationUserList(
1113 admin_pb2.RemoveUserFromModerationUserListReq(user=user2.username, moderation_list_id=moderation_list_id)
1114 )
1115 with session_scope() as session:
1116 assert session.get(ModerationUserList, moderation_list_id) is None
1119def test_admin_delete_account_url(db, email_collector: EmailCollector, push_collector: PushCollector):
1120 super_user, super_token = generate_user(is_superuser=True)
1122 user, token = generate_user()
1123 user_id = user.id
1125 with real_admin_session(super_token) as admin_api:
1126 url = admin_api.CreateAccountDeletionLink(
1127 admin_pb2.CreateAccountDeletionLinkReq(user=user.username)
1128 ).account_deletion_confirm_url
1130 assert push_collector.count_for_user(user_id) == 0
1132 with session_scope() as session:
1133 token_o = session.execute(select(AccountDeletionToken)).scalar_one()
1134 token = token_o.token
1135 assert token_o.user.id == user_id
1136 assert url == f"http://localhost:3000/delete-account?token={token}"
1138 with auth_api_session() as (auth_api, metadata_interceptor):
1139 auth_api.ConfirmDeleteAccount(
1140 auth_pb2.ConfirmDeleteAccountReq(
1141 token=token,
1142 )
1143 )
1145 push = push_collector.pop_for_user(user_id, last=True)
1146 assert push.content.title == "Account deleted"
1147 assert push.content.body == "You can restore it within 7 days using the link we emailed you."
1148 email_collector.pop_for_recipient(user.email, last=True)
1151def test_AccessStats(db):
1152 super_user, super_token = generate_user(is_superuser=True)
1153 normal_user, normal_token = generate_user()
1155 # Insert UserActivity rows: a couple inside the default 90-day window, one well
1156 # outside it, and one with NULL ip_address / user_agent. The INET column is
1157 # returned by psycopg3 as an IPv4Address/IPv6Address object, which used to
1158 # crash the proto string assignment.
1159 in_window_1 = now() - timedelta(days=1)
1160 in_window_2 = now() - timedelta(days=10)
1161 out_of_window = now() - timedelta(days=200)
1162 with session_scope() as session:
1163 session.add(
1164 UserActivity(
1165 user_id=normal_user.id, period=in_window_1, ip_address="1.2.3.4", user_agent="ua-a", api_calls=5
1166 )
1167 )
1168 session.add(
1169 UserActivity(
1170 user_id=normal_user.id, period=in_window_2, ip_address="2001:db8::1", user_agent="ua-b", api_calls=3
1171 )
1172 )
1173 session.add(
1174 UserActivity(
1175 user_id=normal_user.id, period=out_of_window, ip_address="9.9.9.9", user_agent="ua-old", api_calls=99
1176 )
1177 )
1178 session.add(UserActivity(user_id=normal_user.id, period=in_window_1, api_calls=1))
1180 with real_admin_session(super_token) as api:
1181 res = api.AccessStats(admin_pb2.AccessStatsReq(user=normal_user.username))
1183 by_ip = {s.ip_address: s for s in res.stats}
1184 assert "1.2.3.4" in by_ip
1185 assert by_ip["1.2.3.4"].api_call_count == 5
1186 assert by_ip["1.2.3.4"].user_agent == "ua-a"
1187 assert "2001:db8::1" in by_ip
1188 assert by_ip["2001:db8::1"].api_call_count == 3
1189 # NULL ip_address row produces an empty-string ip_address in the proto
1190 assert "" in by_ip
1191 assert by_ip[""].api_call_count == 1
1192 # out-of-window row is excluded by the 90-day default
1193 assert "9.9.9.9" not in by_ip
1195 # explicit end_time should bound the upper end of the window (regression: was >=)
1196 with real_admin_session(super_token) as api:
1197 res = api.AccessStats(
1198 admin_pb2.AccessStatsReq(
1199 user=normal_user.username,
1200 start_time=Timestamp_from_datetime(now() - timedelta(days=5)),
1201 end_time=Timestamp_from_datetime(now()),
1202 )
1203 )
1204 ips = {s.ip_address for s in res.stats}
1205 assert ips == {"1.2.3.4", ""}
1208def test_SetLastDonated(db):
1209 super_user, super_token = generate_user(is_superuser=True)
1210 normal_user, normal_token = generate_user(last_donated=None)
1212 with real_admin_session(super_token) as api:
1213 # user starts with no last_donated
1214 with session_scope() as session:
1215 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1216 assert user.last_donated is None
1218 # can set last_donated
1219 donation_time = now() - timedelta(days=30)
1220 res = api.SetLastDonated(
1221 admin_pb2.SetLastDonatedReq(
1222 user=normal_user.username,
1223 last_donated=Timestamp_from_datetime(donation_time),
1224 )
1225 )
1227 with session_scope() as session:
1228 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1229 assert user.last_donated is not None
1230 # check timestamp is close (within a second)
1231 assert abs((user.last_donated - donation_time).total_seconds()) < 1
1233 # can clear last_donated by not setting the field
1234 res = api.SetLastDonated(admin_pb2.SetLastDonatedReq(user=normal_user.username))
1236 with session_scope() as session:
1237 user = session.execute(select(User).where(User.id == normal_user.id)).scalar_one()
1238 assert user.last_donated is None
1240 # user not found
1241 with pytest.raises(grpc.RpcError) as e:
1242 api.SetLastDonated(admin_pb2.SetLastDonatedReq(user="nonexistent"))
1243 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1244 assert e.value.details() == "Couldn't find that user."
1247def test_admin_actions_level(db):
1248 super_user, super_token = generate_user(is_superuser=True)
1249 normal_user, _ = generate_user()
1251 with real_admin_session(super_token) as api:
1252 # Default level is NORMAL
1253 res = api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=normal_user.username, admin_note="normal note"))
1254 assert res.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
1256 # Explicitly set to DEBUG
1257 res = api.AddAdminNote(
1258 admin_pb2.AddAdminNoteReq(
1259 user=normal_user.username,
1260 admin_note="debug note",
1261 level=admin_pb2.ADMIN_ACTION_LEVEL_DEBUG,
1262 )
1263 )
1264 assert len(res.admin_actions) == 2
1265 assert res.admin_actions[1].level == admin_pb2.ADMIN_ACTION_LEVEL_DEBUG
1267 # Explicitly set to HIGH
1268 res = api.AddAdminNote(
1269 admin_pb2.AddAdminNoteReq(
1270 user=normal_user.username,
1271 admin_note="high note",
1272 level=admin_pb2.ADMIN_ACTION_LEVEL_HIGH,
1273 )
1274 )
1275 assert len(res.admin_actions) == 3
1276 assert res.admin_actions[2].level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH
1279def test_admin_actions_on_mutations(db, push_collector: PushCollector):
1280 super_user, super_token = generate_user(is_superuser=True)
1281 normal_user, _ = generate_user()
1283 original_gender = normal_user.gender
1284 original_birthdate = normal_user.birthdate
1286 with real_admin_session(super_token) as api:
1287 # ChangeUserGender
1288 res = api.ChangeUserGender(admin_pb2.ChangeUserGenderReq(user=normal_user.username, gender="Machine"))
1289 assert any(
1290 a.action_type == "change_gender" and a.note == f"Changed from '{original_gender}' to 'Machine'"
1291 for a in res.admin_actions
1292 )
1294 # ChangeUserBirthdate
1295 res = api.ChangeUserBirthdate(
1296 admin_pb2.ChangeUserBirthdateReq(user=normal_user.username, birthdate="1990-01-01")
1297 )
1298 assert any(
1299 a.action_type == "change_birthdate" and a.note == f"Changed from {original_birthdate} to 1990-01-01"
1300 for a in res.admin_actions
1301 )
1303 # SetPassportSexGenderException
1304 res = api.SetPassportSexGenderException(
1305 admin_pb2.SetPassportSexGenderExceptionReq(user=normal_user.username, passport_sex_gender_exception=True)
1306 )
1307 assert any(
1308 a.action_type == "set_passport_sex_gender_exception" and a.note == "Changed from False to True"
1309 for a in res.admin_actions
1310 )
1312 # SendModNote with notify
1313 res = api.SendModNote(
1314 admin_pb2.SendModNoteReq(
1315 user=normal_user.username, content="Please update your profile", internal_id="test1"
1316 )
1317 )
1318 assert any(
1319 a.action_type == "send_mod_note" and a.note == "Notify user: Yes\n\nPlease update your profile"
1320 for a in res.admin_actions
1321 )
1323 # SendModNote with do_not_notify
1324 res = api.SendModNote(
1325 admin_pb2.SendModNoteReq(
1326 user=normal_user.username,
1327 content="Silent note",
1328 internal_id="test2",
1329 do_not_notify=True,
1330 )
1331 )
1332 assert any(
1333 a.action_type == "send_mod_note" and a.note == "Notify user: No\n\nSilent note" for a in res.admin_actions
1334 )
1336 # DeleteUser
1337 res = api.DeleteUser(admin_pb2.DeleteUserReq(user=normal_user.username))
1338 assert any(a.action_type == "delete_user" for a in res.admin_actions)
1339 assert any(
1340 a.action_type == "delete_user" and a.level == admin_pb2.ADMIN_ACTION_LEVEL_HIGH for a in res.admin_actions
1341 )
1343 # RecoverDeletedUser
1344 res = api.RecoverDeletedUser(admin_pb2.RecoverDeletedUserReq(user=normal_user.username))
1345 assert any(a.action_type == "recover_user" for a in res.admin_actions)
1347 # MarkUserNeedsLocationUpdate
1348 res = api.MarkUserNeedsLocationUpdate(admin_pb2.MarkUserNeedsLocationUpdateReq(user=normal_user.username))
1349 assert any(
1350 a.action_type == "mark_needs_location_update" and a.note == "Marked user as needing location update"
1351 for a in res.admin_actions
1352 )
1354 # SetLastDonated
1355 res = api.SetLastDonated(
1356 admin_pb2.SetLastDonatedReq(
1357 user=normal_user.username,
1358 last_donated=Timestamp_from_datetime(now()),
1359 )
1360 )
1361 assert any(a.action_type == "set_last_donated" for a in res.admin_actions)
1364def test_create_admin_tag(db):
1365 super_user, super_token = generate_user(is_superuser=True)
1367 with real_admin_session(super_token) as api:
1368 res = api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1369 assert res.tag == "test-tag"
1370 assert res.admin_tag_id > 0
1373def test_create_admin_tag_duplicate(db):
1374 super_user, super_token = generate_user(is_superuser=True)
1376 with real_admin_session(super_token) as api:
1377 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1378 with pytest.raises(grpc.RpcError) as e:
1379 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="test-tag"))
1380 assert e.value.code() == grpc.StatusCode.ALREADY_EXISTS
1381 assert e.value.details() == "That admin tag already exists."
1384def test_create_admin_tag_empty(db):
1385 super_user, super_token = generate_user(is_superuser=True)
1387 with real_admin_session(super_token) as api:
1388 with pytest.raises(grpc.RpcError) as e:
1389 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=""))
1390 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1391 assert e.value.details() == "The admin tag cannot be empty."
1393 with pytest.raises(grpc.RpcError) as e:
1394 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag=" "))
1395 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1396 assert e.value.details() == "The admin tag cannot be empty."
1399def test_list_admin_tags(db):
1400 super_user, super_token = generate_user(is_superuser=True)
1402 with real_admin_session(super_token) as api:
1403 # Empty initially
1404 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1405 assert len(res.tags) == 0
1407 # Add some tags
1408 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="bravo"))
1409 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="alpha"))
1411 res = api.ListAdminTags(admin_pb2.ListAdminTagsReq())
1412 assert len(res.tags) == 2
1413 # Ordered alphabetically
1414 assert res.tags[0].tag == "alpha"
1415 assert res.tags[1].tag == "bravo"
1418def test_add_admin_tag_to_user(db):
1419 super_user, super_token = generate_user(is_superuser=True)
1420 normal_user, _ = generate_user()
1422 with real_admin_session(super_token) as api:
1423 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1425 res = api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1426 assert "vip" in res.admin_tags
1427 assert any(a.action_type == "add_tag" and a.tag == "vip" for a in res.admin_actions)
1430def test_add_admin_tag_to_user_duplicate(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"))
1436 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1438 with pytest.raises(grpc.RpcError) as e:
1439 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1440 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1441 assert e.value.details() == "The user already has that admin tag."
1444def test_add_admin_tag_to_user_tag_not_found(db):
1445 super_user, super_token = generate_user(is_superuser=True)
1446 normal_user, _ = generate_user()
1448 with real_admin_session(super_token) as api:
1449 with pytest.raises(grpc.RpcError) as e:
1450 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="nonexistent"))
1451 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1452 assert e.value.details() == "Admin tag not found."
1455def test_remove_admin_tag_from_user(db):
1456 super_user, super_token = generate_user(is_superuser=True)
1457 normal_user, _ = generate_user()
1459 with real_admin_session(super_token) as api:
1460 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1461 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=normal_user.username, tag="vip"))
1463 res = api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1464 assert "vip" not in res.admin_tags
1465 assert any(a.action_type == "remove_tag" and a.tag == "vip" for a in res.admin_actions)
1468def test_remove_admin_tag_from_user_not_assigned(db):
1469 super_user, super_token = generate_user(is_superuser=True)
1470 normal_user, _ = generate_user()
1472 with real_admin_session(super_token) as api:
1473 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1475 with pytest.raises(grpc.RpcError) as e:
1476 api.RemoveAdminTagFromUser(admin_pb2.RemoveAdminTagFromUserReq(user=normal_user.username, tag="vip"))
1477 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1478 assert e.value.details() == "The user does not have that admin tag."
1481def test_search_users_by_admin_tag(db):
1482 super_user, super_token = generate_user(is_superuser=True)
1483 user1, _ = generate_user()
1484 user2, _ = generate_user()
1485 user3, _ = generate_user()
1487 with real_admin_session(super_token) as api:
1488 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="vip"))
1489 api.CreateAdminTag(admin_pb2.CreateAdminTagReq(tag="flagged"))
1491 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user1.username, tag="vip"))
1492 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="vip"))
1493 api.AddAdminTagToUser(admin_pb2.AddAdminTagToUserReq(user=user2.username, tag="flagged"))
1495 # Search for users with "vip" tag
1496 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip"]))
1497 user_ids = {u.user_id for u in res.users}
1498 assert user1.id in user_ids
1499 assert user2.id in user_ids
1500 assert user3.id not in user_ids
1502 # Search for users with both "vip" AND "flagged" tags (AND logic)
1503 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["vip", "flagged"]))
1504 user_ids = {u.user_id for u in res.users}
1505 assert user2.id in user_ids
1506 assert user1.id not in user_ids
1508 # Search for non-existent tag returns no results
1509 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_tags=["nonexistent"]))
1510 assert len(res.users) == 0
1513def test_search_users_by_admin_note(db):
1514 super_user, super_token = generate_user(is_superuser=True)
1515 user1, _ = generate_user()
1516 user2, _ = generate_user()
1518 with real_admin_session(super_token) as api:
1519 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="suspicious activity"))
1520 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="normal user"))
1522 # Search by admin action log content (ilike)
1523 res = api.SearchUsers(admin_pb2.SearchUsersReq(admin_action_log="%suspicious%"))
1524 user_ids = {u.user_id for u in res.users}
1525 assert user1.id in user_ids
1526 assert user2.id not in user_ids
1529def test_ListAdminActions_empty(db):
1530 super_user, super_token = generate_user(is_superuser=True)
1532 with real_admin_session(super_token) as api:
1533 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1534 assert len(res.admin_actions) == 0
1535 assert res.next_page_token == ""
1538def test_ListAdminActions_returns_newest_first_with_target_info(db):
1539 super_user, super_token = generate_user(is_superuser=True)
1540 user1, _ = generate_user()
1541 user2, _ = generate_user()
1543 with real_admin_session(super_token) as api:
1544 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="first note"))
1545 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="second note"))
1546 api.BanUser(admin_pb2.BanUserReq(user=user1.username, admin_note="ban reason"))
1548 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq())
1550 assert len(res.admin_actions) == 3
1551 # Newest first
1552 assert res.admin_actions[0].action_type == "ban"
1553 assert res.admin_actions[0].target_user_id == user1.id
1554 assert res.admin_actions[0].target_username == user1.username
1555 assert res.admin_actions[0].admin_user_id == super_user.id
1556 assert res.admin_actions[0].admin_username == super_user.username
1557 assert res.admin_actions[1].action_type == "note"
1558 assert res.admin_actions[1].target_user_id == user2.id
1559 assert res.admin_actions[2].action_type == "note"
1560 assert res.admin_actions[2].target_user_id == user1.id
1563def test_ListAdminActions_filter_by_admin_and_target(db):
1564 super1, super1_token = generate_user(is_superuser=True)
1565 super2, super2_token = generate_user(is_superuser=True)
1566 user1, _ = generate_user()
1567 user2, _ = generate_user()
1569 with real_admin_session(super1_token) as api:
1570 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super1 to user1"))
1571 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user2.username, admin_note="from super1 to user2"))
1572 with real_admin_session(super2_token) as api:
1573 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user1.username, admin_note="from super2 to user1"))
1575 with real_admin_session(super1_token) as api:
1576 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id))
1577 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super1 to user2"}
1579 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(target_user_id=user1.id))
1580 assert {a.note for a in res.admin_actions} == {"from super1 to user1", "from super2 to user1"}
1582 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(admin_user_id=super1.id, target_user_id=user1.id))
1583 assert [a.note for a in res.admin_actions] == ["from super1 to user1"]
1586def test_ListAdminActions_pagination(db):
1587 super_user, super_token = generate_user(is_superuser=True)
1588 user, _ = generate_user()
1590 with real_admin_session(super_token) as api:
1591 for i in range(3):
1592 api.AddAdminNote(admin_pb2.AddAdminNoteReq(user=user.username, admin_note=f"note {i}"))
1594 res = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2))
1595 assert len(res.admin_actions) == 2
1596 assert res.next_page_token != ""
1597 first_page_notes = [a.note for a in res.admin_actions]
1599 res2 = api.ListAdminActions(admin_pb2.ListAdminActionsReq(page_size=2, page_token=res.next_page_token))
1600 assert len(res2.admin_actions) == 1
1601 assert res2.next_page_token == ""
1603 all_notes = first_page_notes + [a.note for a in res2.admin_actions]
1604 assert set(all_notes) == {"note 0", "note 1", "note 2"}
1607def test_ListUserUploads(db):
1608 super_user, super_token = generate_user(is_superuser=True)
1609 user, _ = generate_user(complete_profile=False)
1610 other_user, _ = generate_user()
1612 with session_scope() as session:
1613 for i in range(3):
1614 session.add(
1615 Upload(
1616 key=f"key{i}",
1617 filename=f"photo{i}.jpg",
1618 creator_user_id=user.id,
1619 credit=f"credit {i}" if i == 0 else None,
1620 )
1621 )
1622 session.add(Upload(key="other_key", filename="other.jpg", creator_user_id=other_user.id))
1624 with real_admin_session(super_token) as api:
1625 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username))
1627 assert len(res.uploads) == 3
1628 assert res.next_page_token == ""
1629 assert {u.filename for u in res.uploads} == {"photo0.jpg", "photo1.jpg", "photo2.jpg"}
1631 upload0 = next(u for u in res.uploads if u.key == "key0")
1632 assert upload0.credit == "credit 0"
1633 assert upload0.full_url.endswith("/img/full/photo0.jpg")
1634 assert upload0.thumbnail_url.endswith("/img/thumbnail/photo0.jpg")
1635 assert upload0.HasField("created")
1638def test_ListUserUploads_pagination(db):
1639 super_user, super_token = generate_user(is_superuser=True)
1640 user, _ = generate_user(complete_profile=False)
1642 with session_scope() as session:
1643 for i in range(3):
1644 session.add(Upload(key=f"key{i}", filename=f"photo{i}.jpg", creator_user_id=user.id))
1646 with real_admin_session(super_token) as api:
1647 res = api.ListUserUploads(admin_pb2.ListUserUploadsReq(user=user.username, page_size=2))
1648 assert len(res.uploads) == 2
1649 assert res.next_page_token != ""
1650 first_page_keys = [u.key for u in res.uploads]
1652 res2 = api.ListUserUploads(
1653 admin_pb2.ListUserUploadsReq(user=user.username, page_size=2, page_token=res.next_page_token)
1654 )
1655 assert len(res2.uploads) == 1
1656 assert res2.next_page_token == ""
1658 all_keys = first_page_keys + [u.key for u in res2.uploads]
1659 assert set(all_keys) == {"key0", "key1", "key2"}
1662def test_ListUserUploads_not_found(db):
1663 super_user, super_token = generate_user(is_superuser=True)
1665 with real_admin_session(super_token) as api:
1666 with pytest.raises(grpc.RpcError) as e:
1667 api.ListUserUploads(admin_pb2.ListUserUploadsReq(user="nonexistent"))
1668 assert e.value.code() == grpc.StatusCode.NOT_FOUND
1671# community invite feature tested in test_events.py
1672# SendBlogPostNotification tested in test_notifications.py
1673# MarkUserNeedsLocationUpdate tested in test_jail.py
1676def _ota_manifest(*, version, fingerprint, created_at="2026-05-31T00:00:00.000Z"):
1677 return {
1678 "id": f"id-{version}",
1679 "createdAt": created_at,
1680 "runtimeVersion": fingerprint,
1681 "launchAsset": {"key": "bundle", "url": f"https://cdn.testing.invalid/{version}/bundle.hbc"},
1682 "assets": [],
1683 "metadata": {},
1684 "extra": {},
1685 }
1688def _ota_signed_multipart(manifest):
1689 # Mimics the signed multipart body the CDN holds (signature header omitted; we only read the JSON).
1690 boundary = "COUCHERS_OTA_BOUNDARY"
1692 def part(name, body, content_type):
1693 return f'--{boundary}\r\ncontent-disposition: form-data; name="{name}"\r\ncontent-type: {content_type}\r\n\r\n{body}\r\n'
1695 body = (
1696 part("manifest", json.dumps(manifest), "application/json; charset=utf-8")
1697 + part("extensions", "{}", "application/json")
1698 + f"--{boundary}--\r\n"
1699 )
1700 return f"multipart/mixed; boundary={boundary}", body.encode()
1703def _patch_ota_cdn(manifests):
1704 # manifests: {version: manifest_dict}. URL is {cdn_root}/{version}/{platform}/manifest.
1705 def fake(url):
1706 version = url.split("/")[-3]
1707 if version not in manifests:
1708 return "multipart/mixed; boundary=COUCHERS_OTA_BOUNDARY", b""
1709 return _ota_signed_multipart(manifests[version])
1711 return patch("couchers.servicers.admin._fetch_signed_manifest", side_effect=fake)
1714def test_CreateOTAPackage(db):
1715 super_user, super_token = generate_user(is_superuser=True)
1717 manifests = {"v1.3.1.aaaa": _ota_manifest(version="v1.3.1.aaaa", fingerprint="ios-fp")}
1718 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1719 res = api.CreateOTAPackage(
1720 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1721 )
1723 assert res.platform == admin_pb2.OTA_PLATFORM_IOS
1724 assert res.fingerprint == "ios-fp"
1725 assert res.version == "v1.3.1.aaaa"
1726 assert res.manifest_id == "id-v1.3.1.aaaa"
1727 assert res.banned is False
1728 assert res.live is True
1729 assert res.creator_user_id == super_user.id
1732def test_CreateOTAPackage_invalid(db):
1733 _, super_token = generate_user(is_superuser=True)
1735 manifests = {"v-incomplete": {"id": "x"}} # on the CDN but missing runtimeVersion / createdAt
1736 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1737 # missing version
1738 with pytest.raises(grpc.RpcError) as e:
1739 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1740 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1742 # nothing published at this version
1743 with pytest.raises(grpc.RpcError) as e:
1744 api.CreateOTAPackage(
1745 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-missing")
1746 )
1747 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1749 # manifest present but missing required fields
1750 with pytest.raises(grpc.RpcError) as e:
1751 api.CreateOTAPackage(
1752 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v-incomplete")
1753 )
1754 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1757def test_CreateOTAPackage_rejects_duplicate_version(db):
1758 _, 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 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa"))
1763 with pytest.raises(grpc.RpcError) as e:
1764 api.CreateOTAPackage(
1765 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.aaaa")
1766 )
1767 assert e.value.code() == grpc.StatusCode.FAILED_PRECONDITION
1770def test_ListOTAPackages(db):
1771 _, super_token = generate_user(is_superuser=True)
1773 manifests = {
1774 "v1.3.1.ios": _ota_manifest(version="v1.3.1.ios", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1775 "v1.3.2.ios": _ota_manifest(version="v1.3.2.ios", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1776 "v1.3.2.android": _ota_manifest(
1777 version="v1.3.2.android", fingerprint="android-fp", created_at="2026-06-01T00:00:00.000Z"
1778 ),
1779 }
1780 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1781 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.ios"))
1782 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.ios"))
1783 api.CreateOTAPackage(
1784 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_ANDROID, version="v1.3.2.android")
1785 )
1787 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1788 # newest (by manifest createdAt) first
1789 assert [p.version for p in res.packages] == ["v1.3.2.android", "v1.3.2.ios", "v1.3.1.ios"]
1790 # only the newest per (platform, fingerprint) is live
1791 live = {p.version: p.live for p in res.packages}
1792 assert live == {"v1.3.2.android": True, "v1.3.2.ios": True, "v1.3.1.ios": False}
1794 ios = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(platform=admin_pb2.OTA_PLATFORM_IOS))
1795 assert [p.version for p in ios.packages] == ["v1.3.2.ios", "v1.3.1.ios"]
1798def test_BanOTAPackage(db):
1799 super_user, super_token = generate_user(is_superuser=True)
1801 manifests = {
1802 "v1.3.1.good": _ota_manifest(
1803 version="v1.3.1.good", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"
1804 ),
1805 "v1.3.2.bad": _ota_manifest(version="v1.3.2.bad", fingerprint="ios-fp", created_at="2026-05-31T00:00:00.000Z"),
1806 }
1807 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1808 api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1.good"))
1809 second = api.CreateOTAPackage(
1810 admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.2.bad")
1811 )
1812 assert second.live is True
1814 banned = api.BanOTAPackage(
1815 admin_pb2.BanOTAPackageReq(ota_package_id=second.ota_package_id, reason="bad bundle")
1816 )
1817 assert banned.banned is True
1818 assert banned.banned_reason == "bad bundle"
1819 assert banned.banned_by_user_id == super_user.id
1820 assert banned.live is False
1822 # banning the newest stops new check-ins getting it; the previous one becomes live again
1823 res = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq(include_banned=True))
1824 live = {p.version: p.live for p in res.packages}
1825 assert live == {"v1.3.2.bad": False, "v1.3.1.good": True}
1827 # banned packages are excluded by default
1828 non_banned = api.ListOTAPackages(admin_pb2.ListOTAPackagesReq())
1829 assert [p.version for p in non_banned.packages] == ["v1.3.1.good"]
1832def test_BanOTAPackage_requires_reason(db):
1833 _, super_token = generate_user(is_superuser=True)
1835 manifests = {
1836 "v1.3.1": _ota_manifest(version="v1.3.1", fingerprint="ios-fp", created_at="2026-05-30T00:00:00.000Z"),
1837 }
1838 with _patch_ota_cdn(manifests), real_admin_session(super_token) as api:
1839 pkg = api.CreateOTAPackage(admin_pb2.CreateOTAPackageReq(platform=admin_pb2.OTA_PLATFORM_IOS, version="v1.3.1"))
1840 with pytest.raises(grpc.RpcError) as e:
1841 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id))
1842 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1843 with pytest.raises(grpc.RpcError) as e:
1844 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=pkg.ota_package_id, reason=" "))
1845 assert e.value.code() == grpc.StatusCode.INVALID_ARGUMENT
1848def test_BanOTAPackage_not_found(db):
1849 _, super_token = generate_user(is_superuser=True)
1851 with real_admin_session(super_token) as api:
1852 with pytest.raises(grpc.RpcError) as e:
1853 api.BanOTAPackage(admin_pb2.BanOTAPackageReq(ota_package_id=123456, reason="never mind"))
1854 assert e.value.code() == grpc.StatusCode.NOT_FOUND