Coverage for app/backend/src/couchers/servicers/blocking.py: 94%

1import grpc 

2from google.protobuf import empty_pb2 

3from sqlalchemy import exists, false, select 

4from sqlalchemy.orm import Session 

5from sqlalchemy.sql import not_, or_, union 

6 

7from couchers import urls 

8from couchers.context import CouchersContext 

9from couchers.models import Upload, User, UserBlock 

10from couchers.models.uploads import get_avatar_photo_subquery 

11from couchers.proto import blocking_pb2, blocking_pb2_grpc 

12 

13 

14def is_not_visible( 

15 session: Session, user1_id: int | None, user2_id: int | None, *, ignore_shadow: bool = False 

16) -> bool: 

17 """ 

18 Check if users are not visible to each other (due to block or because either account is deleted/banned/shadowed). 

19 """ 

20 hidden_users = select(User.id).where(or_(User.id == user1_id, User.id == user2_id)).where(not_(User.is_visible)) 

21 shadowed_target = select(User.id).where(false()) 

22 if not ignore_shadow: 

23 shadowed_target = select(User.id).where(User.id == user2_id).where(User.shadowed_at.is_not(None)) 

24 if user1_id is not None: 

25 shadowed_target = shadowed_target.where(User.id != user1_id) 

26 # if either user_id is empty, just check if either user is hidden (as they can't block each other) 

27 if not user1_id or not user2_id: 

28 return ( 

29 session.execute(select(union(hidden_users, shadowed_target).subquery()).limit(1)).one_or_none() is not None 

30 ) 

31 else: 

32 blocked_users = ( 

33 select(UserBlock.blocked_user_id) 

34 .where(UserBlock.blocking_user_id == user1_id) 

35 .where(UserBlock.blocked_user_id == user2_id) 

36 ) 

37 blocking_users = ( 

38 select(UserBlock.blocking_user_id) 

39 .where(UserBlock.blocking_user_id == user2_id) 

40 .where(UserBlock.blocked_user_id == user1_id) 

41 ) 

42 return ( 

43 session.execute( 

44 select(union(blocked_users, blocking_users, hidden_users, shadowed_target).subquery()).limit(1) 

45 ).one_or_none() 

46 is not None 

47 ) 

48 

49 

50class Blocking(blocking_pb2_grpc.BlockingServicer): 

51 def BlockUser( 

52 self, request: blocking_pb2.BlockUserReq, context: CouchersContext, session: Session 

53 ) -> empty_pb2.Empty: 

54 blockee = session.execute( 

55 select(User).where(User.is_visible).where(User.username == request.username) 

56 ).scalar_one_or_none() 

57 

58 if not blockee: 58 ↛ 59line 58 didn't jump to line 59 because the condition on line 58 was never true

59 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found") 

60 

61 if context.user_id == blockee.id: 

62 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "cant_block_self") 

63 

64 if session.execute( 

65 select( 

66 exists() 

67 .where(UserBlock.blocking_user_id == context.user_id) 

68 .where(UserBlock.blocked_user_id == blockee.id) 

69 ) 

70 ).scalar_one(): 

71 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_already_blocked") 

72 else: 

73 user_block = UserBlock( 

74 blocking_user_id=context.user_id, 

75 blocked_user_id=blockee.id, 

76 ) 

77 session.add(user_block) 

78 session.commit() 

79 

80 return empty_pb2.Empty() 

81 

82 def UnblockUser( 

83 self, request: blocking_pb2.UnblockUserReq, context: CouchersContext, session: Session 

84 ) -> empty_pb2.Empty: 

85 blockee = session.execute( 

86 select(User).where(User.is_visible).where(User.username == request.username) 

87 ).scalar_one_or_none() 

88 

89 if not blockee: 89 ↛ 90line 89 didn't jump to line 90 because the condition on line 89 was never true

90 context.abort_with_error_code(grpc.StatusCode.NOT_FOUND, "user_not_found") 

91 

92 user_block = session.execute( 

93 select(UserBlock) 

94 .where(UserBlock.blocking_user_id == context.user_id) 

95 .where(UserBlock.blocked_user_id == blockee.id) 

96 ).scalar_one_or_none() 

97 if not user_block: 

98 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "user_not_blocked") 

99 

100 session.delete(user_block) 

101 session.commit() 

102 

103 return empty_pb2.Empty() 

104 

105 def GetBlockedUsers( 

106 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

107 ) -> blocking_pb2.GetBlockedUsersRes: 

108 avatar_photo_subquery = get_avatar_photo_subquery() 

109 

110 blocked_users = session.execute( 

111 select(User.username, User.name, Upload.filename) 

112 .join(UserBlock, UserBlock.blocked_user_id == User.id) 

113 .outerjoin( 

114 avatar_photo_subquery, 

115 avatar_photo_subquery.c.gallery_id == User.profile_gallery_id, 

116 ) 

117 .outerjoin(Upload, Upload.key == avatar_photo_subquery.c.upload_key) 

118 .where(User.is_visible) 

119 .where(UserBlock.blocking_user_id == context.user_id) 

120 ).all() 

121 

122 return blocking_pb2.GetBlockedUsersRes( 

123 blocked_users=[ 

124 blocking_pb2.BlockedUser( 

125 username=blocked_user.username, 

126 name=blocked_user.name, 

127 avatar_thumbnail_url=urls.media_url(filename=blocked_user.filename, size="thumbnail") 

128 if blocked_user.filename 

129 else None, 

130 ) 

131 for blocked_user in blocked_users 

132 ] 

133 )