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.event_log import log_event 

8from couchers.models import ContentReport, User 

9from couchers.proto import reporting_pb2, reporting_pb2_grpc 

10from couchers.sql import username_or_id 

11from couchers.tasks import send_content_report_email 

12 

13 

14class Reporting(reporting_pb2_grpc.ReportingServicer): 

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

16 # note no filtering on visibility 

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

18 

19 if not author_user: 

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

21 

22 content_report = ContentReport( 

23 reporting_user_id=context.user_id, 

24 reason=request.reason, 

25 description=request.description, 

26 content_ref=request.content_ref, 

27 author_user_id=author_user.id, 

28 user_agent=request.user_agent, 

29 page=request.page, 

30 ) 

31 

32 session.add(content_report) 

33 session.flush() 

34 

35 send_content_report_email(session, content_report) 

36 

37 log_event( 

38 context, 

39 session, 

40 "content.reported", 

41 { 

42 "author_user_id": author_user.id, 

43 "reason": request.reason, 

44 "content_ref": request.content_ref, 

45 "page": request.page, 

46 }, 

47 ) 

48 

49 return empty_pb2.Empty()