Coverage for app/backend/src/tests/test_admin_export.py: 100%
295 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-19 17:13 +0000
1import json
2from datetime import date, timedelta
3from typing import Any
5import grpc
6import pytest
7from sqlalchemy import func, update
9from couchers.db import session_scope
10from couchers.models import (
11 FriendRelationship,
12 FriendStatus,
13 HostRequest,
14 Invoice,
15 InvoiceType,
16 ModerationObjectType,
17 ModerationState,
18 ModerationVisibility,
19 PhotoGalleryItem,
20 Upload,
21 User,
22)
23from couchers.proto import (
24 admin_pb2,
25 conversations_pb2,
26 discussions_pb2,
27 events_pb2,
28 messages_pb2,
29 public_trips_pb2,
30 references_pb2,
31 requests_pb2,
32 threads_pb2,
33)
34from couchers.utils import not_none, now, today
35from tests.fixtures.api import events as events_api
36from tests.fixtures.db import generate_user, make_friends, make_user_block, make_user_invisible
37from tests.fixtures.sessions import (
38 conversations_session,
39 discussions_session,
40 public_trips_session,
41 real_admin_session,
42 references_session,
43 requests_session,
44 threads_session,
45)
46from tests.test_communities import create_community
47from tests.test_public_trips import VALID_DESCRIPTION as TRIP_DESCRIPTION
48from tests.test_requests import valid_request_text
49from tests.utils import local_date_time
52def _export(super_token: str, user: User) -> Any:
53 with real_admin_session(super_token) as api:
54 res = api.ExportUserData(admin_pb2.ExportUserDataReq(user=user.username))
55 assert res.user_id == user.id
56 return json.loads(res.json)
59def test_ExportUserData(db):
60 super_user, super_token = generate_user(is_superuser=True)
61 user, _ = generate_user(complete_profile=False, about_me="I like couches", pronouns="she/her")
63 with real_admin_session(super_token) as api:
64 res = api.ExportUserData(admin_pb2.ExportUserDataReq(user=user.username))
66 assert res.user_id == user.id
67 assert res.filename.startswith(f"couchers-data-export-{user.username}-")
68 assert res.filename.endswith(".json")
70 export = json.loads(res.json)
71 assert export["export_format_version"] == 1
72 assert export["account"]["user_id"] == str(user.id)
73 assert export["account"]["username"] == user.username
74 assert export["account"]["email"] == user.email
75 assert export["account"]["pronouns"] == "she/her"
76 assert export["profile"]["about_me"] == "I like couches"
77 assert export["home"]["about_place"] == user.about_place
79 with real_admin_session(super_token) as api:
80 details = api.GetUserDetails(admin_pb2.GetUserDetailsReq(user=user.username))
81 assert [action.action_type for action in details.admin_actions] == ["export_user_data"]
82 assert details.admin_actions[0].level == admin_pb2.ADMIN_ACTION_LEVEL_NORMAL
85def test_ExportUserData_not_found(db):
86 _, super_token = generate_user(is_superuser=True)
88 with real_admin_session(super_token) as api:
89 with pytest.raises(grpc.RpcError) as e:
90 api.ExportUserData(admin_pb2.ExportUserDataReq(user="nonexistentuser"))
91 assert e.value.code() == grpc.StatusCode.NOT_FOUND
94def test_ExportUserData_references_written(db):
95 super_user, super_token = generate_user(is_superuser=True)
96 user, token = generate_user()
97 friend, _ = generate_user()
99 make_friends(user, friend)
101 with references_session(token) as api:
102 api.WriteFriendReference(
103 references_pb2.WriteFriendReferenceReq(
104 to_user_id=friend.id,
105 text="a good friend",
106 private_text="but they snore",
107 was_appropriate=True,
108 rating=0.5,
109 )
110 )
112 export = _export(super_token, user)
114 written = export["references"]["written"]
115 assert len(written) == 1
116 assert written[0]["text"] == "a good friend"
117 assert written[0]["to_user"] == {"user_id": str(friend.id), "username": f"@{friend.username}"}
120def test_ExportUserData_conversations_with_visible_user(db, moderator):
121 super_user, super_token = generate_user(is_superuser=True)
122 user, token = generate_user()
123 other, other_token = generate_user()
125 make_friends(user, other)
127 with conversations_session(token) as api:
128 chat_id = api.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id])).group_chat_id
129 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello from me"))
130 moderator.approve_group_chat(chat_id)
131 with conversations_session(other_token) as api:
132 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello back from them"))
134 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
135 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
136 with requests_session(token) as api:
137 host_request_id = api.CreateHostRequest(
138 requests_pb2.CreateHostRequestReq(
139 host_user_id=other.id,
140 from_date=today_plus_2,
141 to_date=today_plus_3,
142 text=valid_request_text(),
143 )
144 ).host_request_id
145 moderator.approve_host_request(host_request_id)
146 with requests_session(other_token) as api:
147 api.SendHostRequestMessage(
148 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request_id, text="sure, come on over")
149 )
151 export = _export(super_token, user)
153 (chat,) = export["group_chats"]
154 assert chat["creator"] == {"user_id": str(user.id), "username": f"@{user.username}"}
155 texts = [message.get("text") for message in chat["messages"]]
156 assert "hello from me" in texts
157 assert "hello back from them" in texts
158 their_message = next(message for message in chat["messages"] if message.get("text") == "hello back from them")
159 assert their_message["author"] == {"user_id": str(other.id), "username": f"@{other.username}"}
160 assert their_message["time"]
162 (request,) = export["host_requests"]
163 assert request["role"] == "surfer"
164 assert request["host"] == {"user_id": str(other.id), "username": f"@{other.username}"}
165 request_texts = [message.get("text") for message in request["messages"]]
166 assert valid_request_text() in request_texts
167 assert "sure, come on over" in request_texts
170def test_ExportUserData_conversations_with_invisible_user(db, moderator):
171 super_user, super_token = generate_user(is_superuser=True)
172 user, token = generate_user()
173 other, other_token = generate_user()
175 make_friends(user, other)
177 with conversations_session(token) as api:
178 chat_id = api.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id])).group_chat_id
179 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello from me"))
180 moderator.approve_group_chat(chat_id)
181 with conversations_session(other_token) as api:
182 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello back from them"))
184 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
185 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
186 with requests_session(token) as api:
187 host_request_id = api.CreateHostRequest(
188 requests_pb2.CreateHostRequestReq(
189 host_user_id=other.id,
190 from_date=today_plus_2,
191 to_date=today_plus_3,
192 text=valid_request_text(),
193 )
194 ).host_request_id
195 moderator.approve_host_request(host_request_id)
196 with requests_session(other_token) as api:
197 api.SendHostRequestMessage(
198 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request_id, text="sure, come on over")
199 )
201 make_user_invisible(other.id)
203 export = _export(super_token, user)
205 (chat,) = export["group_chats"]
206 assert [message["text"] for message in chat["messages"] if message.get("text")] == [
207 "hello from me",
208 "hello back from them",
209 ]
210 their_message = next(message for message in chat["messages"] if message.get("text") == "hello back from them")
211 assert their_message["author"] == {"user_id": str(other.id)}
212 assert chat["members"] == [{"user_id": str(user.id), "username": f"@{user.username}"}, {"user_id": str(other.id)}]
214 assert export.get("host_requests", []) == []
215 assert other.username not in json.dumps(export)
218def test_ExportUserData_conversations_with_blocked_user(db, moderator):
219 super_user, super_token = generate_user(is_superuser=True)
220 user, token = generate_user()
221 other, other_token = generate_user()
223 make_friends(user, other)
225 with conversations_session(token) as api:
226 chat_id = api.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id])).group_chat_id
227 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello from me"))
228 moderator.approve_group_chat(chat_id)
229 with conversations_session(other_token) as api:
230 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello back from them"))
232 make_user_block(user, other)
234 export = _export(super_token, user)
236 (chat,) = export["group_chats"]
237 assert [message["text"] for message in chat["messages"] if message.get("text")] == [
238 "hello from me",
239 "hello back from them",
240 ]
241 their_message = next(message for message in chat["messages"] if message.get("text") == "hello back from them")
242 assert their_message["author"] == {"user_id": str(other.id)}
244 # but their own block list still names who they blocked, as it does in the app
245 assert export["blocked_users"] == [
246 {
247 "user": {"user_id": str(other.id), "username": f"@{other.username}"},
248 "time_blocked": export["blocked_users"][0]["time_blocked"],
249 }
250 ]
253def test_ExportUserData_covers_content_across_the_site(db, moderator):
254 super_user, super_token = generate_user(is_superuser=True)
255 user, token = generate_user()
257 with session_scope() as session:
258 node_id = create_community(session, 0, 2, "Export Community", [user], [], None).id
260 with discussions_session(token) as api:
261 discussion = api.CreateDiscussion(
262 discussions_pb2.CreateDiscussionReq(
263 title="export discussion",
264 content="export discussion content",
265 owner_community_id=node_id,
266 )
267 )
269 start_time = now() + timedelta(hours=2)
270 event = events_api.create_event(
271 token=token,
272 title="export event",
273 content="export event content",
274 location=events_pb2.EventLocation(address="Near Null Island", lat=0.1, lng=0.2),
275 start_date_time=local_date_time(start_time),
276 end_date_time=local_date_time(start_time + timedelta(hours=3)),
277 )
279 with threads_session(token) as api:
280 comment_thread_id = api.PostReply(
281 threads_pb2.PostReplyReq(thread_id=discussion.thread.thread_id, content="export comment")
282 ).thread_id
283 api.PostReply(threads_pb2.PostReplyReq(thread_id=comment_thread_id, content="export reply"))
284 api.PostReply(threads_pb2.PostReplyReq(thread_id=event.thread.thread_id, content="export event comment"))
286 with public_trips_session(token) as api:
287 api.CreatePublicTrip(
288 public_trips_pb2.CreatePublicTripReq(
289 community_id=node_id,
290 from_date=(today() + timedelta(days=5)).isoformat(),
291 to_date=(today() + timedelta(days=10)).isoformat(),
292 description=TRIP_DESCRIPTION,
293 )
294 )
296 with session_scope() as session:
297 session.add(Upload(key="export_key", filename="export.jpg", creator_user_id=user.id, credit="me"))
298 session.flush()
299 session.add(
300 PhotoGalleryItem(
301 gallery_id=not_none(user.profile_gallery_id), upload_key="export_key", position=1.0, caption="hi"
302 )
303 )
304 session.add(
305 Invoice(
306 user_id=user.id,
307 amount=25.0,
308 stripe_payment_intent_id="pi_export",
309 stripe_receipt_url="https://stripe.example.org/receipt",
310 invoice_type=InvoiceType.on_platform,
311 )
312 )
314 export = _export(super_token, user)
316 assert [community["name"] for community in export["communities"]] == ["Export Community"]
318 assert [item["title"] for item in export["discussions"]["created"]] == ["export discussion"]
319 comments = {comment["content"]: comment["posted_in"] for comment in export["discussions"]["comments"]}
320 assert comments["export comment"] == {"type": "discussion", "title": "export discussion"}
321 assert comments["export event comment"] == {"type": "event", "title": "export event"}
322 (reply,) = export["discussions"]["replies"]
323 assert reply["content"] == "export reply"
324 assert reply["posted_in"]["title"] == "export discussion"
326 (event,) = export["events"]["created"]
327 assert event["title"] == "export event"
328 assert event["address"] == "Near Null Island"
329 assert event["start"] and event["end"]
330 assert [item["title"] for item in export["events"]["attending"]] == ["export event"]
331 assert export["events"]["attending"][0]["attendee_status"] == "going"
332 assert [item["title"] for item in export["events"]["organizing"]] == ["export event"]
334 (trip,) = export["public_trips"]
335 assert trip["description"] == TRIP_DESCRIPTION
337 photos = {photo["url"]: photo for gallery in export["photos"]["galleries"] for photo in gallery["photos"]}
338 (photo_url,) = [url for url in photos if url.endswith("/img/full/export.jpg")]
339 assert photos[photo_url]["caption"] == "hi"
340 assert photo_url in [upload["url"] for upload in export["photos"]["uploads"]]
342 (invoice,) = export["donations"]
343 assert invoice["amount"] == 25.0
344 assert invoice["receipt_url"] == "https://stripe.example.org/receipt"
347def _add_friend_relationship(from_user: User, to_user: User, status: FriendStatus) -> None:
348 with session_scope() as session:
349 moderation_state = ModerationState(
350 object_type=ModerationObjectType.friend_request,
351 object_id=0,
352 visibility=ModerationVisibility.visible,
353 )
354 session.add(moderation_state)
355 session.flush()
356 relationship = FriendRelationship(
357 from_user_id=from_user.id,
358 to_user_id=to_user.id,
359 status=status,
360 moderation_state_id=moderation_state.id,
361 )
362 session.add(relationship)
363 session.flush()
364 moderation_state.object_id = relationship.id
367def test_ExportUserData_leaves_out_how_others_answered_friend_requests(db):
368 super_user, super_token = generate_user(is_superuser=True)
369 user, _ = generate_user()
370 accepted, _ = generate_user()
371 they_rejected, _ = generate_user()
372 they_cancelled, _ = generate_user()
373 asked_them, _ = generate_user()
374 they_asked, _ = generate_user()
376 make_friends(user, accepted)
377 _add_friend_relationship(user, they_rejected, FriendStatus.rejected)
378 _add_friend_relationship(user, they_cancelled, FriendStatus.cancelled)
379 _add_friend_relationship(user, asked_them, FriendStatus.pending)
380 _add_friend_relationship(they_asked, user, FriendStatus.pending)
382 export = _export(super_token, user)
384 friends = {friend["user"]["username"]: friend for friend in export["friends"]}
385 assert set(friends) == {f"@{accepted.username}", f"@{asked_them.username}", f"@{they_asked.username}"}
387 # a rejected or cancelled request is the other party's answer, and the app never shows it
388 assert they_rejected.username not in json.dumps(export)
389 assert they_cancelled.username not in json.dumps(export)
391 assert friends[f"@{accepted.username}"]["status"] == "friends"
392 assert friends[f"@{asked_them.username}"]["status"] == "request_sent"
393 assert friends[f"@{asked_them.username}"]["time_sent"]
394 # when they sent it isn't this user's to know
395 assert friends[f"@{they_asked.username}"]["status"] == "request_received"
396 assert "time_sent" not in friends[f"@{they_asked.username}"]
399def test_ExportUserData_leaves_out_references_the_user_cannot_see_yet(db, moderator):
400 super_user, super_token = generate_user(is_superuser=True)
401 surfer, surfer_token = generate_user()
402 host, host_token = generate_user()
404 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
405 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
406 with requests_session(surfer_token) as api:
407 host_request_id = api.CreateHostRequest(
408 requests_pb2.CreateHostRequestReq(
409 host_user_id=host.id,
410 from_date=today_plus_2,
411 to_date=today_plus_3,
412 text=valid_request_text(),
413 )
414 ).host_request_id
415 moderator.approve_host_request(host_request_id)
416 with requests_session(host_token) as api:
417 api.RespondHostRequest(
418 requests_pb2.RespondHostRequestReq(
419 host_request_id=host_request_id, status=messages_pb2.HOST_REQUEST_STATUS_ACCEPTED
420 )
421 )
422 with session_scope() as session:
423 session.execute(
424 update(HostRequest)
425 .where(HostRequest.conversation_id == host_request_id)
426 .values(from_date=date.today() - timedelta(days=3), to_date=date.today() - timedelta(days=2))
427 )
429 with references_session(host_token) as api:
430 reference_id = api.WriteHostRequestReference(
431 references_pb2.WriteHostRequestReferenceReq(
432 host_request_id=host_request_id,
433 text="they were a fine guest",
434 was_appropriate=True,
435 rating=0.5,
436 )
437 ).reference_id
438 moderator.approve_reference(reference_id)
440 # the surfer hasn't written theirs back and the window is still open, so the app hides it
441 assert _export(super_token, surfer).get("references", {}).get("received", []) == []
443 with references_session(surfer_token) as api:
444 api.WriteHostRequestReference(
445 references_pb2.WriteHostRequestReferenceReq(
446 host_request_id=host_request_id,
447 text="a fine host",
448 was_appropriate=True,
449 rating=0.5,
450 )
451 )
453 (received,) = _export(super_token, surfer)["references"]["received"]
454 assert received["text"] == "they were a fine guest"
455 assert "rating" not in received
458def test_ExportUserData_leaves_out_moderation_hidden_content(db, moderator):
459 super_user, super_token = generate_user(is_superuser=True)
460 user, token = generate_user()
461 other, other_token = generate_user()
463 make_friends(user, other)
465 with conversations_session(token) as api:
466 chat_id = api.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id])).group_chat_id
467 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello from me"))
468 moderator.approve_group_chat(chat_id)
470 assert "hello from me" in json.dumps(_export(super_token, user))
472 with session_scope() as session:
473 session.execute(
474 update(ModerationState)
475 .where(ModerationState.object_type == ModerationObjectType.group_chat)
476 .where(ModerationState.object_id == chat_id)
477 .values(visibility=ModerationVisibility.hidden)
478 )
480 assert "hello from me" not in json.dumps(_export(super_token, user))
483def test_ExportUserData_for_a_banned_user_is_not_gutted(db, moderator):
484 super_user, super_token = generate_user(is_superuser=True)
485 user, token = generate_user()
486 other, other_token = generate_user()
488 make_friends(user, other)
490 with conversations_session(token) as api:
491 chat_id = api.CreateGroupChat(conversations_pb2.CreateGroupChatReq(recipient_user_ids=[other.id])).group_chat_id
492 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello from me"))
493 moderator.approve_group_chat(chat_id)
494 with conversations_session(other_token) as api:
495 api.SendMessage(conversations_pb2.SendMessageReq(group_chat_id=chat_id, text="hello back from them"))
497 today_plus_2 = (date.today() + timedelta(days=2)).isoformat()
498 today_plus_3 = (date.today() + timedelta(days=3)).isoformat()
499 with requests_session(token) as api:
500 host_request_id = api.CreateHostRequest(
501 requests_pb2.CreateHostRequestReq(
502 host_user_id=other.id,
503 from_date=today_plus_2,
504 to_date=today_plus_3,
505 text=valid_request_text(),
506 )
507 ).host_request_id
508 moderator.approve_host_request(host_request_id)
509 with requests_session(other_token) as api:
510 api.SendHostRequestMessage(
511 requests_pb2.SendHostRequestMessageReq(host_request_id=host_request_id, text="sure, come on over")
512 )
514 with references_session(other_token) as api:
515 reference_id = api.WriteFriendReference(
516 references_pb2.WriteFriendReferenceReq(
517 to_user_id=user.id, text="a good friend", was_appropriate=True, rating=0.5
518 )
519 ).reference_id
520 moderator.approve_reference(reference_id)
522 make_user_invisible(user.id)
524 export = _export(super_token, user)
526 assert export["friends"] == [
527 {"user": {"user_id": str(other.id), "username": f"@{other.username}"}, "status": "friends"}
528 ]
530 (received,) = export["references"]["received"]
531 assert received["text"] == "a good friend"
532 assert received["from_user"] == {"user_id": str(other.id), "username": f"@{other.username}"}
534 (chat,) = export["group_chats"]
535 assert [message["text"] for message in chat["messages"] if message.get("text")] == [
536 "hello from me",
537 "hello back from them",
538 ]
540 (request,) = export["host_requests"]
541 assert request["host"] == {"user_id": str(other.id), "username": f"@{other.username}"}
542 assert [message["text"] for message in request["messages"] if message.get("text")] == [
543 valid_request_text(),
544 "sure, come on over",
545 ]
548def test_ExportUserData_keeps_references_from_writers_who_deleted_their_account(db, moderator):
549 super_user, super_token = generate_user(is_superuser=True)
550 user, token = generate_user()
551 other, other_token = generate_user()
553 make_friends(user, other)
555 with references_session(other_token) as api:
556 reference_id = api.WriteFriendReference(
557 references_pb2.WriteFriendReferenceReq(
558 to_user_id=user.id, text="a good friend", was_appropriate=True, rating=0.5
559 )
560 ).reference_id
561 moderator.approve_reference(reference_id)
563 with session_scope() as session:
564 session.execute(update(User).where(User.id == other.id).values(deleted_at=func.now()))
566 (received,) = _export(super_token, user)["references"]["received"]
567 assert received["text"] == "a good friend"
568 assert received["from_user"] == {"user_id": str(other.id)}