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

50 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 17:19 +0000

1import logging 

2 

3import grpc 

4 

5from couchers import errors 

6from couchers.constants import GUIDELINES_VERSION, TOS_VERSION 

7from couchers.db import session_scope 

8from couchers.models import User 

9from couchers.sql import couchers_select as select 

10from couchers.utils import create_coordinate 

11from proto import jail_pb2, jail_pb2_grpc 

12 

13logger = logging.getLogger(__name__) 

14 

15 

16class Jail(jail_pb2_grpc.JailServicer): 

17 """ 

18 The Jail servicer. 

19 

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

21 fully active 

22 """ 

23 

24 def _get_jail_info(self, user): 

25 res = jail_pb2.JailInfoRes( 

26 has_not_accepted_tos=user.accepted_tos < TOS_VERSION, 

27 has_not_added_location=user.is_missing_location, 

28 has_not_accepted_community_guidelines=user.accepted_community_guidelines < GUIDELINES_VERSION, 

29 ) 

30 

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

32 jailed = False 

33 for field in res.DESCRIPTOR.fields: 

34 if getattr(res, field.name): 

35 jailed = True 

36 res.jailed = jailed 

37 

38 # double check 

39 assert user.is_jailed == jailed 

40 

41 return res 

42 

43 def JailInfo(self, request, context): 

44 with session_scope() as session: 

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

46 return self._get_jail_info(user) 

47 

48 def AcceptTOS(self, request, context): 

49 with session_scope() as session: 

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

51 

52 if not request.accept: 

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

54 

55 user.accepted_tos = TOS_VERSION 

56 session.commit() 

57 

58 return self._get_jail_info(user) 

59 

60 def SetLocation(self, request, context): 

61 with session_scope() as session: 

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

63 

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

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

66 

67 user.city = request.city 

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

69 user.geom_radius = request.radius 

70 

71 session.commit() 

72 

73 return self._get_jail_info(user) 

74 

75 def AcceptCommunityGuidelines(self, request, context): 

76 with session_scope() as session: 

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

78 

79 if not request.accept: 

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

81 

82 user.accepted_community_guidelines = GUIDELINES_VERSION 

83 session.commit() 

84 

85 return self._get_jail_info(user)