Coverage for src/couchers/servicers/bugs.py: 100%

1import grpc 

2import requests 

3from sqlalchemy.sql import func 

4 

5from couchers import errors 

6from couchers.config import config 

7from couchers.descriptor_pool import get_descriptors_pb 

8from couchers.models import User 

9from couchers.sql import couchers_select as select 

10from proto import bugs_pb2, bugs_pb2_grpc 

11from proto.google.api import httpbody_pb2 

12 

13 

14class Bugs(bugs_pb2_grpc.BugsServicer): 

15 def _version(self): 

16 return config["VERSION"] 

17 

18 def Version(self, request, context, session): 

19 return bugs_pb2.VersionInfo(version=self._version()) 

20 

21 def ReportBug(self, request, context, session): 

22 if not config["BUG_TOOL_ENABLED"]: 

23 context.abort(grpc.StatusCode.UNAVAILABLE, errors.BUG_TOOL_DISABLED) 

24 

25 repo = config["BUG_TOOL_GITHUB_REPO"] 

26 auth = (config["BUG_TOOL_GITHUB_USERNAME"], config["BUG_TOOL_GITHUB_TOKEN"]) 

27 

28 if context.user_id: 

29 username = session.execute(select(User.username).where(User.id == context.user_id)).scalar_one() 

30 user_details = f"{username} ({context.user_id})" 

31 else: 

32 user_details = "<not logged in>" 

33 

34 issue_title = request.subject 

35 issue_body = ( 

36 f"Subject: {request.subject}\n" 

37 f"Description:\n" 

38 f"{request.description}\n" 

39 f"\n" 

40 f"Results:\n" 

41 f"{request.results}\n" 

42 f"\n" 

43 f"Backend version: {self._version()}\n" 

44 f"Frontend version: {request.frontend_version}\n" 

45 f"User Agent: {request.user_agent}\n" 

46 f"Page: {request.page}\n" 

47 f"User: {user_details}" 

48 ) 

49 issue_labels = ["bug tool"] 

50 

51 json_body = {"title": issue_title, "body": issue_body, "labels": issue_labels} 

52 

53 r = requests.post(f"https://api.github.com/repos/{repo}/issues", auth=auth, json=json_body) 

54 if not r.status_code == 201: 

55 context.abort(grpc.StatusCode.INTERNAL, errors.BUG_TOOL_REQUEST_FAILED) 

56 

57 issue_number = r.json()["number"] 

58 

59 return bugs_pb2.ReportBugRes( 

60 bug_id=f"#{issue_number}", bug_url=f"https://github.com/{repo}/issues/{issue_number}" 

61 ) 

62 

63 def Status(self, request, context, session): 

64 coucher_count = session.execute(select(func.count()).select_from(User).where(User.is_visible)).scalar_one() 

65 

66 return bugs_pb2.StatusRes( 

67 nonce=request.nonce, 

68 version=self._version(), 

69 coucher_count=coucher_count, 

70 ) 

71 

72 def GetDescriptors(self, request, context, session): 

73 return httpbody_pb2.HttpBody( 

74 content_type="application/octet-stream", 

75 data=get_descriptors_pb(), 

76 )