Coverage for app / backend / src / couchers / servicers / reporting.py: 100%

1import grpc 

2from google.protobuf import empty_pb2 

3from sqlalchemy import select 

4from sqlalchemy.orm import Session 

5 

6from couchers.context import CouchersContext 

7from couchers.models import ContentReport, User 

8from couchers.proto import reporting_pb2, reporting_pb2_grpc 

9from couchers.sql import username_or_id 

10from couchers.tasks import send_content_report_email 

11 

12 

13class Reporting(reporting_pb2_grpc.ReportingServicer): 

14 def Report(self, request: reporting_pb2.ReportReq, context: CouchersContext, session: Session) -> empty_pb2.Empty: 

15 # note no filtering on visibility 

16 author_user = session.execute(select(User).where(username_or_id(request.author_user))).scalar_one_or_none() 

17 

18 if not author_user: 

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

20 

21 content_report = ContentReport( 

22 reporting_user_id=context.user_id, 

23 reason=request.reason, 

24 description=request.description, 

25 content_ref=request.content_ref, 

26 author_user_id=author_user.id, 

27 user_agent=request.user_agent, 

28 page=request.page, 

29 ) 

30 

31 session.add(content_report) 

32 session.flush() 

33 

34 send_content_report_email(session, content_report) 

35 

36 return empty_pb2.Empty()