Coverage for chalicelib/core/issue_tracking/github_issue.py: 19%

1from chalicelib.core.issue_tracking.base_issue import BaseIntegrationIssue 

2from chalicelib.utils import github_client_v3 

3from chalicelib.utils.github_client_v3 import github_formatters as formatter 

4 

5 

6class GithubIntegrationIssue(BaseIntegrationIssue): 

7 def __init__(self, token): 

8 self.__client = github_client_v3.githubV3Request(token) 

9 super(GithubIntegrationIssue, self).__init__("GITHUB", token) 

10 

11 def get_current_user(self): 

12 return formatter.user(self.__client.get("/user")) 

13 

14 def get_meta(self, repoId): 

15 current_user = self.get_current_user() 

16 try: 

17 users = self.__client.get(f"/repositories/{repoId}/collaborators") 

18 except Exception as e: 

19 users = [] 

20 users = [formatter.user(u) for u in users] 

21 if current_user not in users: 

22 users.insert(0, current_user) 

23 meta = { 

24 'users': users, 

25 'issueTypes': [formatter.label(l) for l in 

26 self.__client.get(f"/repositories/{repoId}/labels")] 

27 } 

28 

29 return meta 

30 

31 def create_new_assignment(self, integration_project_id, title, description, assignee, 

32 issue_type): 

33 repoId = integration_project_id 

34 assignees = [assignee] 

35 labels = [str(issue_type)] 

36 

37 metas = self.get_meta(repoId) 

38 real_assignees = [] 

39 for a in assignees: 

40 for u in metas["users"]: 

41 if a == str(u["id"]): 

42 real_assignees.append(u["name"]) 

43 break 

44 real_labels = ["OpenReplay"] 

45 for l in labels: 

46 found = False 

47 for ll in metas["issueTypes"]: 

48 if l == str(ll["id"]): 

49 real_labels.append(ll["name"]) 

50 found = True 

51 break 

52 if not found: 

53 real_labels.append(l) 

54 issue = self.__client.post(f"/repositories/{repoId}/issues", body={"title": title, "body": description, 

55 "assignees": real_assignees, 

56 "labels": real_labels}) 

57 return formatter.issue(issue) 

58 

59 def get_by_ids(self, saved_issues): 

60 results = [] 

61 for i in saved_issues: 

62 results.append(self.get(integration_project_id=i["integrationProjectId"], assignment_id=i["id"])) 

63 return {"issues": results} 

64 

65 def get(self, integration_project_id, assignment_id): 

66 repoId = integration_project_id 

67 issueNumber = assignment_id 

68 issue = self.__client.get(f"/repositories/{repoId}/issues/{issueNumber}") 

69 issue = formatter.issue(issue) 

70 if issue["commentsCount"] > 0: 

71 issue["comments"] = [formatter.comment(c) for c in 

72 self.__client.get(f"/repositories/{repoId}/issues/{issueNumber}/comments")] 

73 return issue 

74 

75 def comment(self, integration_project_id, assignment_id, comment): 

76 repoId = integration_project_id 

77 issueNumber = assignment_id 

78 commentCreated = self.__client.post(f"/repositories/{repoId}/issues/{issueNumber}/comments", 

79 body={"body": comment}) 

80 return formatter.comment(commentCreated) 

81 

82 def get_metas(self, integration_project_id): 

83 current_user = self.get_current_user() 

84 try: 

85 users = self.__client.get(f"/repositories/{integration_project_id}/collaborators") 

86 except Exception as e: 

87 users = [] 

88 users = [formatter.user(u) for u in users] 

89 if current_user not in users: 

90 users.insert(0, current_user) 

91 

92 return {"provider": self.provider.lower(), 

93 'users': users, 

94 'issueTypes': [formatter.label(l) for l in 

95 self.__client.get(f"/repositories/{integration_project_id}/labels")] 

96 } 

97 

98 def get_projects(self): 

99 repos = self.__client.get("/user/repos") 

100 return [formatter.repo(r) for r in repos]