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

1from typing import cast 

2 

3import grpc 

4import requests 

5from google.protobuf import empty_pb2 

6from sqlalchemy.orm import Session 

7from sqlalchemy.sql import func 

8 

9from couchers import urls 

10from couchers.config import config 

11from couchers.context import CouchersContext 

12from couchers.descriptor_pool import get_descriptors_pb 

13from couchers.models import User 

14from couchers.proto import bugs_pb2, bugs_pb2_grpc 

15from couchers.proto.google.api import httpbody_pb2 

16from couchers.sql import couchers_select as select 

17 

18 

19class Bugs(bugs_pb2_grpc.BugsServicer): 

20 def _version(self) -> str: 

21 return cast(str, config["VERSION"]) 

22 

23 def Version(self, request: empty_pb2.Empty, context: CouchersContext, session: Session) -> bugs_pb2.VersionInfo: 

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

25 

26 def ReportBug( 

27 self, request: bugs_pb2.ReportBugReq, context: CouchersContext, session: Session 

28 ) -> bugs_pb2.ReportBugRes: 

29 if not config["BUG_TOOL_ENABLED"]: 

30 context.abort_with_error_code(grpc.StatusCode.UNAVAILABLE, "bug_tool_disabled") 

31 

32 repo = config["BUG_TOOL_GITHUB_REPO"] 

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

34 

35 if context.is_logged_in(): 

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

37 user_details = f"[@{username}]({urls.user_link(username=username)}) ({context.user_id})" 

38 else: 

39 user_details = "<not logged in>" 

40 

41 issue_title = request.subject 

42 issue_body = ( 

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

44 f"Description:\n" 

45 f"{request.description}\n" 

46 f"\n" 

47 f"Results:\n" 

48 f"{request.results}\n" 

49 f"\n" 

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

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

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

53 f"Screen resolution: {request.screen_resolution.width}x{request.screen_resolution.height}\n" 

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

55 f"User: {user_details}" 

56 ) 

57 issue_labels = ["bug tool", "bug: triage needed"] 

58 

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

60 

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

62 if not r.status_code == 201: 

63 context.abort_with_error_code(grpc.StatusCode.INTERNAL, "bug_tool_request_failed") 

64 

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

66 

67 return bugs_pb2.ReportBugRes( 

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

69 ) 

70 

71 def Status(self, request: bugs_pb2.StatusReq, context: CouchersContext, session: Session) -> bugs_pb2.StatusRes: 

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

73 

74 return bugs_pb2.StatusRes( 

75 nonce=request.nonce, 

76 version=self._version(), 

77 coucher_count=coucher_count, 

78 ) 

79 

80 def GetDescriptors( 

81 self, request: empty_pb2.Empty, context: CouchersContext, session: Session 

82 ) -> httpbody_pb2.HttpBody: 

83 return httpbody_pb2.HttpBody( 

84 content_type="application/octet-stream", 

85 data=get_descriptors_pb(), 

86 ) 

87 

88 def GeolocationSearchInfo( 

89 self, request: bugs_pb2.GeolocationSearchInfoReq, context: CouchersContext, session: Session 

90 ) -> empty_pb2.Empty: 

91 return empty_pb2.Empty() 

92 

93 def GeolocationClickInfo( 

94 self, request: bugs_pb2.GeolocationClickInfoReq, context: CouchersContext, session: Session 

95 ) -> empty_pb2.Empty: 

96 return empty_pb2.Empty()