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

1import grpc 

2import requests 

3from google.protobuf import empty_pb2 

4from sqlalchemy.sql import func 

5 

6from couchers import urls 

7from couchers.config import config 

8from couchers.descriptor_pool import get_descriptors_pb 

9from couchers.models import User 

10from couchers.proto import bugs_pb2, bugs_pb2_grpc 

11from couchers.proto.google.api import httpbody_pb2 

12from couchers.sql import couchers_select as select 

13 

14 

15class Bugs(bugs_pb2_grpc.BugsServicer): 

16 def _version(self): 

17 return config["VERSION"] 

18 

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

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

21 

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

23 if not config["BUG_TOOL_ENABLED"]: 

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

25 

26 repo = config["BUG_TOOL_GITHUB_REPO"] 

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

28 

29 if context.is_logged_in(): 

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

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

32 else: 

33 user_details = "<not logged in>" 

34 

35 issue_title = request.subject 

36 issue_body = ( 

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

38 f"Description:\n" 

39 f"{request.description}\n" 

40 f"\n" 

41 f"Results:\n" 

42 f"{request.results}\n" 

43 f"\n" 

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

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

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

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

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

49 f"User: {user_details}" 

50 ) 

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

52 

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

54 

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

56 if not r.status_code == 201: 

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

58 

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

60 

61 return bugs_pb2.ReportBugRes( 

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

63 ) 

64 

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

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

67 

68 return bugs_pb2.StatusRes( 

69 nonce=request.nonce, 

70 version=self._version(), 

71 coucher_count=coucher_count, 

72 ) 

73 

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

75 return httpbody_pb2.HttpBody( 

76 content_type="application/octet-stream", 

77 data=get_descriptors_pb(), 

78 ) 

79 

80 def GeolocationSearchInfo(self, request, context, session): 

81 return empty_pb2.Empty() 

82 

83 def GeolocationClickInfo(self, request, context, session): 

84 return empty_pb2.Empty()