Coverage for src/couchers/servicers/jail.py: 98%
57 statements
« prev ^ index » next coverage.py v7.5.0, created at 2024-10-04 23:02 +0000
« prev ^ index » next coverage.py v7.5.0, created at 2024-10-04 23:02 +0000
1import logging
3import grpc
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
13logger = logging.getLogger(__name__)
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 )
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
35 # double check
36 assert user.is_jailed == jailed
38 return res
41class Jail(jail_pb2_grpc.JailServicer):
42 """
43 The Jail servicer.
45 API calls allowed for users who need to complete some tasks before being
46 fully active
47 """
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)
53 def AcceptTOS(self, request, context, session):
54 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
56 if not request.accept:
57 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANT_UNACCEPT_TOS)
59 user.accepted_tos = TOS_VERSION
60 session.commit()
62 return _get_jail_info(session, user)
64 def SetLocation(self, request, context, session):
65 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
67 if request.lat == 0 and request.lng == 0:
68 context.abort(grpc.StatusCode.INVALID_ARGUMENT, errors.INVALID_COORDINATE)
70 user.city = request.city
71 user.geom = create_coordinate(request.lat, request.lng)
72 user.geom_radius = request.radius
74 session.commit()
76 return _get_jail_info(session, user)
78 def AcceptCommunityGuidelines(self, request, context, session):
79 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
81 if not request.accept:
82 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.CANT_UNACCEPT_COMMUNITY_GUIDELINES)
84 user.accepted_community_guidelines = GUIDELINES_VERSION
85 session.commit()
87 return _get_jail_info(session, user)
89 def AcknowledgePendingModNote(self, request, context, session):
90 user = session.execute(select(User).where(User.id == context.user_id)).scalar_one()
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()
99 if not note:
100 context.abort(grpc.StatusCode.NOT_FOUND, errors.MOD_NOTE_NOT_FOUND)
102 if not request.acknowledge:
103 context.abort(grpc.StatusCode.FAILED_PRECONDITION, errors.MOD_NOTE_NEED_TO_ACKNOWELDGE)
105 note.acknowledged = now()
106 session.flush()
108 return _get_jail_info(session, user)