Coverage for webapp/integrations/logic.py: 21%

1import re 

2from github import Github 

3from os import getenv 

4 

5from redis_cache.cache_utility import redis_cache 

6from webapp.helpers import get_yaml_loader 

7from webapp.observability.utils import trace_function 

8from webapp.packages.logic import ( 

9 get_package, 

10) 

11 

12GITHUB_TOKEN = getenv("GITHUB_TOKEN") 

13 

14github_client = Github(GITHUB_TOKEN) 

15yaml = get_yaml_loader() 

16 

17 

18class Interfaces: 

19 def __init__(self): 

20 self._repo = None 

21 

22 @property 

23 def repo(self): 

24 if self._repo is None: 

25 self._repo = github_client.get_repo( 

26 "canonical/charmlibs" 

27 ) 

28 return self._repo 

29 

30 @trace_function 

31 def get_interfaces(self): 

32 key = "get-interfaces" 

33 interfaces = redis_cache.get(key, expected_type=list) 

34 if interfaces: 

35 return interfaces 

36 try: 

37 index = self.repo.get_contents("interfaces/index.json") 

38 interfaces = yaml.load(index.decoded_content.decode("utf-8")) 

39 

40 normalized_interfaces = [] 

41 for interface in interfaces or []: 

42 normalized_interfaces.append( 

43 { 

44 "name": interface.get("name", ""), 

45 "status": interface.get("status", ""), 

46 "summary": interface.get("summary", ""), 

47 "description": interface.get("description", ""), 

48 "version": interface.get("version", ""), 

49 "tags": interface.get("tags", []), 

50 "links": { 

51 "library": interface.get("lib_url"), 

52 "documentation": interface.get("docs_url") 

53 }, 

54 } 

55 ) 

56 

57 redis_cache.set(key, normalized_interfaces, ttl=86400) 

58 return normalized_interfaces 

59 except Exception: 

60 return [] 

61 

62 @trace_function 

63 def get_interface_from_path(self, interface_name): 

64 key = f"get-interface-from-path:{interface_name}" 

65 interface = redis_cache.get(key, expected_type=dict) 

66 if interface: 

67 return interface 

68 try: 

69 interface_versions = self.repo.get_contents( 

70 "interfaces/{}/interface".format(interface_name) 

71 ) 

72 except Exception: 

73 return None 

74 versions = [] 

75 if not isinstance(interface_versions, list): 

76 interface_versions = [interface_versions] 

77 for _, version in enumerate(interface_versions): 

78 if version.type == "dir" and version.name.startswith("v"): 

79 versions.append(version.name) 

80 

81 versions.sort() 

82 

83 latest_version = versions.pop() 

84 

85 latest_version_interface = self.repo.get_contents( 

86 "interfaces/{}/interface/{}/interface.yaml".format( 

87 interface_name, latest_version 

88 ) 

89 ).decoded_content.decode("utf-8") 

90 interface = yaml.load(latest_version_interface) 

91 

92 active_providers = [] 

93 active_requirers = [] 

94 

95 if "providers" in interface and interface["providers"]: 

96 active_providers = [] 

97 for provider in interface["providers"]: 

98 try: 

99 get_package(provider["name"], [], False) 

100 active_providers.append(provider) 

101 except Exception: 

102 continue 

103 interface["providers"] = active_providers 

104 

105 if "requirers" in interface and interface["requirers"]: 

106 active_requirers = [] 

107 for requirer in interface["requirers"]: 

108 try: 

109 get_package(requirer["name"], [], False) 

110 active_requirers.append(requirer) 

111 except Exception: 

112 continue 

113 interface["requirers"] = active_requirers 

114 redis_cache.set(key, interface, ttl=86400) 

115 return interface 

116 

117 @trace_function 

118 def get_h_content(self, text, pattern): 

119 start_index = text.index(pattern) 

120 return [start_index, start_index + len(pattern)] 

121 

122 @trace_function 

123 def extract_headings_and_content(self, text, level): 

124 headings = re.findall( 

125 r"^#{" + str(level) + r"}\s.*", text, flags=re.MULTILINE 

126 ) 

127 

128 start_end = { 

129 heading: self.get_h_content(text, heading) for heading in headings 

130 } 

131 result = [] 

132 for i in range(len(headings)): 

133 current_heading = headings[i] 

134 start_index = start_end[current_heading][1] 

135 has_next = i < len(headings) - 1 

136 if has_next: 

137 next_heading = headings[i + 1] 

138 end_index = start_end[next_heading][0] 

139 body = text[start_index:end_index] 

140 else: 

141 body = text[start_index:] 

142 

143 result.append([current_heading.strip(), body.strip()]) 

144 return result 

145 

146 @trace_function 

147 def parse_text(self, interface, version, text): 

148 base_link = ( 

149 "https://github.com/canonical/" 

150 "charmlibs/blob/main/interfaces/{}/interface/v{}" 

151 ).format(interface, version) 

152 pattern = r"\[.*?\]\(.*?\)" 

153 matches = re.findall(pattern, text) 

154 

155 for match in matches: 

156 element_pattern = r"\[(.*?)\]\((.*?)\)" 

157 element_match = re.search(element_pattern, match) 

158 if element_match: 

159 title = element_match.group(1) 

160 url = element_match.group(2) 

161 absolute_url = url 

162 if absolute_url.startswith("./"): 

163 absolute_url = absolute_url.replace("./", base_link + "/") 

164 

165 text = text.replace(match, f"[{title}]({absolute_url})") 

166 

167 return text 

168 

169 @trace_function 

170 def convert_readme(self, interface, version, text, level=2): 

171 headings_and_contents = self.extract_headings_and_content(text, level) 

172 

173 if len(headings_and_contents) == 0: 

174 return [s.strip("\n") for s in text.split("\n") if s.strip("\n")] 

175 

176 resulting_list = [] 

177 

178 for heading, content in headings_and_contents: 

179 strip_char = "{}{}".format("#" * level, " ") 

180 heading = heading.strip(strip_char) 

181 _temp = {"heading": heading, "level": level, "children": []} 

182 

183 children = [] 

184 

185 result = self.convert_readme( 

186 interface, version, content, level + 1 

187 ) 

188 

189 if len(content) > 0: 

190 body = content.split("#")[0].strip() 

191 body_list = body.split("\n\n") 

192 if len(list(filter(None, body_list))) > 0: 

193 children += body_list 

194 

195 if isinstance(result, list) and isinstance(result[0], dict): 

196 children += result 

197 

198 for child in children: 

199 if isinstance(child, list): 

200 _temp["children"] += child 

201 else: 

202 _temp["children"].append(child) 

203 

204 for index, child in list(enumerate(_temp["children"])): 

205 if isinstance(child, str): 

206 _temp["children"][index] = self.parse_text( 

207 interface, version, child 

208 ) 

209 

210 resulting_list.append(_temp) 

211 

212 return resulting_list 

213 

214 @trace_function 

215 def get_interface_name_from_readme(self, text): 

216 name = re.sub(r"[#` \n]", "", text.split("\n##", 1)[0]).split("/")[0] 

217 return name 

218 

219 

220interface_logic = Interfaces()