Coverage for chalicelib/utils/github_client_v3.py: 49%

1import logging 

2from datetime import datetime 

3 

4import requests 

5from fastapi import HTTPException, status 

6 

7logger = logging.getLogger(__name__) 

8 

9 

10class github_formatters: 

11 

12 @staticmethod 

13 def get_timestamp(ts): 

14 ts = ts[:-1] 

15 pattern = '%Y-%m-%dT%H:%M:%S' 

16 creation = datetime.strptime(ts, pattern) 

17 return int(creation.timestamp() * 1000) 

18 

19 @staticmethod 

20 def label(label): 

21 return { 

22 'id': label["id"], 

23 'name': label["name"], 

24 'description': label["description"], 

25 'color': label["color"] 

26 } 

27 

28 @staticmethod 

29 def comment(comment): 

30 return { 

31 'id': str(comment["id"]), 

32 'message': comment["body"], 

33 'author': str(github_formatters.user(comment["user"])["id"]), 

34 'createdAt': github_formatters.get_timestamp(comment["created_at"]) 

35 } 

36 

37 @staticmethod 

38 def issue(issue): 

39 labels = [github_formatters.label(l) for l in issue["labels"]] 

40 result = { 

41 'id': str(issue["number"]), 

42 'creator': str(github_formatters.user(issue["user"])["id"]), 

43 'assignees': [str(github_formatters.user(a)["id"]) for a in issue["assignees"]], 

44 'title': issue["title"], 

45 'description': issue["body"], 

46 'status': issue["state"], 

47 'createdAt': github_formatters.get_timestamp(issue["created_at"]), 

48 'closed': issue["closed_at"] is not None, 

49 'commentsCount': issue["comments"], 

50 'issueType': [str(l["id"]) for l in labels if l["name"].lower() != "openreplay"], 

51 'labels': [l["name"] for l in labels] 

52 } 

53 return result 

54 

55 @staticmethod 

56 def user(user): 

57 if not user: 

58 return None 

59 result = { 

60 'id': user["id"], 

61 'name': user["login"], 

62 'avatarUrls': {'24x24': user["avatar_url"]}, 

63 'email': "" 

64 } 

65 return result 

66 

67 @staticmethod 

68 def team_to_dict(team): 

69 if not team: 

70 return None 

71 

72 result = {'id': team.id, 'name': team.name, 'members_count': team.members_count} 

73 return result 

74 

75 @staticmethod 

76 def repo(repo): 

77 if not repo: 

78 return None 

79 return { 

80 "id": str(repo["id"]), 

81 "name": repo["name"], 

82 "description": repo["description"], 

83 "creator": str(repo["owner"]["id"]) 

84 } 

85 

86 @staticmethod 

87 def organization(org): 

88 if not org: 

89 return None 

90 return { 

91 "id": org["id"], 

92 "name": org["login"], 

93 "description": org["description"], 

94 "avatarUrls": {'24x42': org["avatar_url"]} 

95 } 

96 

97 

98def get_response_links(response): 

99 links = {} 

100 if "Link" in response.headers: 100 ↛ 101line 100 didn't jump to line 101 because the condition on line 100 was never true

101 link_headers = response.headers["Link"].split(", ") 

102 for link_header in link_headers: 

103 (url, rel) = link_header.split("; ") 

104 url = url[1:-1] 

105 rel = rel[5:-1] 

106 links[rel] = url 

107 return links 

108 

109 

110class githubV3Request: 

111 __base = "https://api.github.com" 

112 

113 def __init__(self, token): 

114 self.__token = token 

115 

116 def __get_request_header(self): 

117 return {"Accept": "application/vnd.github.v3+json", 'Authorization': f'token {self.__token}'} 

118 

119 def get(self, url, params={}): 

120 results = [] 

121 params = {"per_page": 100, **params} 

122 pages = {"next": f"{self.__base}{url}", "last": ""} 

123 while len(pages.keys()) > 0 and pages["next"] != pages["last"]: 123 ↛ 134line 123 didn't jump to line 134 because the condition on line 123 was always true

124 response = requests.get(pages["next"], headers=self.__get_request_header(), params=params) 

125 pages = get_response_links(response) 

126 result = response.json() 

127 if response.status_code != 200: 127 ↛ 131line 127 didn't jump to line 131 because the condition on line 127 was always true

128 logger.warning(f"=>GITHUB Exception") 

129 logger.error(result) 

130 raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=f"GITHUB: {result['message']}") 

131 if isinstance(result, dict): 

132 return result 

133 results += result 

134 return results 

135 

136 def post(self, url, body): 

137 response = requests.post(f"{self.__base}{url}", headers=self.__get_request_header(), json=body) 

138 return response.json()