Coverage for src/couchers/servicers/jail.py: 98%

1import logging 

2 

3import grpc 

4 

5from couchers import errors 

6from couchers.constants import GUIDELINES_VERSION, TOS_VERSION 

7from couchers.models import ModNote, User 

8from couchers.servicers.account import mod_note_to_pb 

9from couchers.sql import couchers_select as select 

10from couchers.utils import create_coordinate, now 

11from proto import jail_pb2, jail_pb2_grpc 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16def _get_jail_info(session, user): 

17 pending_notes = ( 

18 session.execute(select(ModNote).where(ModNote.user_id == user.id).where(ModNote.is_pending)).scalars().all() 

19 ) 

20 res = jail_pb2.JailInfoRes( 

21 has_not_accepted_tos=user.accepted_tos < TOS_VERSION, 

22 has_not_added_location=user.is_missing_location, 

23 has_not_accepted_community_guidelines=user.accepted_community_guidelines < GUIDELINES_VERSION, 

24 has_pending_mod_notes=len(pending_notes) > 0, 

25 pending_mod_notes=[mod_note_to_pb(note) for note in pending_notes], 

26 ) 

27 

28 # if any of the bools in res are true, we're jailed 

29 jailed = False 

30 for field in res.DESCRIPTOR.fields: 

31 if getattr(res, field.name): 

32 jailed = True 

33 res.jailed = jailed 

34 

35 # double check 

36 assert user.is_jailed == jailed 

37 

38 return res 

39 

40 

41class Jail(jail_pb2_grpc.JailServicer): 

42 """ 

43 The Jail servicer. 

44 

45 API calls allowed for users who need to complete some tasks before being 

46 fully active 

47 """ 

48 

49 def JailInfo(self, request, context, session): 

50 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

51 return _get_jail_info(session, user) 

52 

53 def AcceptTOS(self, request, context, session): 

54 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

55 

56 if not request.accept: 

57 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANT_UNACCEPT_TOS) 

58 

59 user.accepted_tos = TOS_VERSION 

60 session.commit() 

61 

62 return _get_jail_info(session, user) 

63 

64 def SetLocation(self, request, context, session): 

65 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

66 

67 if request.lat == 0 and request.lng == 0: 

68 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.INVALID_COORDINATE) 

69 

70 user.city = request.city 

71 user.geom = create_coordinate(request.lat, request.lng) 

72 user.geom_radius = request.radius 

73 

74 session.commit() 

75 

76 return _get_jail_info(session, user) 

77 

78 def AcceptCommunityGuidelines(self, request, context, session): 

79 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

80 

81 if not request.accept: 

82 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANT_UNACCEPT_COMMUNITY_GUIDELINES) 

83 

84 user.accepted_community_guidelines = GUIDELINES_VERSION 

85 session.commit() 

86 

87 return _get_jail_info(session, user) 

88 

89 def AcknowledgePendingModNote(self, request, context, session): 

90 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one() 

91 

92 note = session.execute( 

93 select(ModNote) 

94 .where(ModNote.user_id == user.id) 

95 .where(ModNote.is_pending) 

96 .where(ModNote.id == request.note_id) 

97 ).scalar_one_or_none() 

98 

99 if not note: 

100 context.abort(grpc.StatusCode.NOT_FOUND, errors.MOD_NOTE_NOT_FOUND) 

101 

102 if not request.acknowledge: 

103 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.MOD_NOTE_NEED_TO_ACKNOWELDGE) 

104 

105 note.acknowledged = now() 

106 session.flush() 

107 

108 return _get_jail_info(session, user)