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

41 statements  

« prev     ^ index     » next       coverage.py v7.5.0, created at 2024-07-22 16:44 +0000

1import grpc 

2import requests 

3from sqlalchemy.sql import func 

4 

5from couchers import errors 

6from couchers.config import config 

7from couchers.db import session_scope 

8from couchers.descriptor_pool import get_descriptors_pb 

9from couchers.models import User 

10from couchers.sql import couchers_select as select 

11from proto import bugs_pb2, bugs_pb2_grpc 

12from proto.google.api import httpbody_pb2 

13 

14 

15class Bugs(bugs_pb2_grpc.BugsServicer): 

16 def _version(self): 

17 return config["VERSION"] 

18 

19 def Version(self, request, context): 

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

21 

22 def ReportBug(self, request, context): 

23 if not config["BUG_TOOL_ENABLED"]: 

24 context.abort(grpc.StatusCode.UNAVAILABLE, errors.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.user_id: 

30 with session_scope() as session: 

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

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

33 else: 

34 user_details = "<not logged in>" 

35 

36 issue_title = request.subject 

37 issue_body = ( 

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

39 f"Description:\n" 

40 f"{request.description}\n" 

41 f"\n" 

42 f"Results:\n" 

43 f"{request.results}\n" 

44 f"\n" 

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

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

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

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

49 f"User: {user_details}" 

50 ) 

51 issue_labels = ["bug tool"] 

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(grpc.StatusCode.INTERNAL, errors.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): 

66 with session_scope() as session: 

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

68 

69 return bugs_pb2.StatusRes( 

70 nonce=request.nonce, 

71 version=self._version(), 

72 coucher_count=coucher_count, 

73 ) 

74 

75 def GetDescriptors(self, request, context): 

76 return httpbody_pb2.HttpBody( 

77 content_type="application/octet-stream", 

78 data=get_descriptors_pb(), 

79 )