Coverage for app / backend / src / couchers / servicers / bugs.py: 97%

1import json 

2import time 

3from datetime import UTC, datetime 

4from typing import cast 

5 

6import grpc 

7import requests 

8from google.protobuf import empty_pb2 

9from sqlalchemy import insert, select 

10from sqlalchemy.orm import Session 

11from sqlalchemy.sql import func 

12 

13from couchers import urls 

14from couchers.config import config 

15from couchers.constants import STABLE_THRESHOLD_SECONDS 

16from couchers.context import CouchersContext 

17from couchers.descriptor_pool import get_descriptors_pb 

18from couchers.models import User 

19from couchers.models.logging import EventLog, EventSource 

20from couchers.proto import bugs_pb2, bugs_pb2_grpc 

21from couchers.proto.google.api import httpbody_pb2 

22 

23_start_time = time.monotonic() 

24 

25 

26class Bugs(bugs_pb2_grpc.BugsServicer): 

27 def _version(self) -> str: 

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

29 

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

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

32 

33 def ReportBug( 

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

35 ) -> bugs_pb2.ReportBugRes: 

36 if not config["BUG_TOOL_ENABLED"]: 

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

38 

39 repo = config["BUG_TOOL_GITHUB_REPO"] 

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

41 

42 if context.is_logged_in(): 

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

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

45 else: 

46 user_details = "<not logged in>" 

47 

48 issue_title = request.subject 

49 issue_body = ( 

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

51 f"Description:\n" 

52 f"{request.description}\n" 

53 f"\n" 

54 f"Results:\n" 

55 f"{request.results}\n" 

56 f"\n" 

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

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

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

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

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

62 f"User: {user_details}" 

63 ) 

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

65 

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

67 

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

69 if not r.status_code == 201: 

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

71 

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

73 

74 return bugs_pb2.ReportBugRes( 

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

76 ) 

77 

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

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

80 

81 return bugs_pb2.StatusRes( 

82 nonce=request.nonce, 

83 version=self._version(), 

84 coucher_count=coucher_count, 

85 stable=time.monotonic() - _start_time >= STABLE_THRESHOLD_SECONDS, 

86 ) 

87 

88 def GetDescriptors( 

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

90 ) -> httpbody_pb2.HttpBody: 

91 return httpbody_pb2.HttpBody( 

92 content_type="application/octet-stream", 

93 data=get_descriptors_pb(), 

94 ) 

95 

96 def ReportDiagnostics( 

97 self, request: bugs_pb2.ReportDiagnosticsReq, context: CouchersContext, session: Session 

98 ) -> empty_pb2.Empty: 

99 if len(request.infos) > 100: 

100 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "too_many_diagnostic_infos") 

101 

102 events = [] 

103 for info in request.infos: 

104 try: 

105 properties = json.loads(info.properties_json) 

106 except (json.JSONDecodeError, ValueError): 

107 context.abort_with_error_code(grpc.StatusCode.INVALID_ARGUMENT, "invalid_diagnostics_json") 

108 

109 occurred = info.occurred.ToDatetime(tzinfo=UTC) if info.HasField("occurred") else datetime.now(UTC) 

110 

111 events.append( 

112 { 

113 "event_type": info.tag, 

114 "user_id": context._user_id, 

115 "sofa": context._sofa, 

116 "version": request.frontend_version, 

117 "properties": properties, 

118 "value": info.value, 

119 "source": EventSource.frontend, 

120 "occurred": occurred, 

121 } 

122 ) 

123 

124 if events: 

125 session.execute(insert(EventLog), events) 

126 

127 return empty_pb2.Empty() 

128 

129 def GeolocationSearchInfo( 

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

131 ) -> empty_pb2.Empty: 

132 return empty_pb2.Empty() 

133 

134 def GeolocationClickInfo( 

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

136 ) -> empty_pb2.Empty: 

137 return empty_pb2.Empty()