Coverage for webapp/solutions/logic.py: 59%

1import os 

2import logging 

3import requests 

4from flask import session as flask_session 

5from webapp.solutions.auth import login 

6from webapp.packages.logic import get_store_categories 

7from webapp.store.logic import format_slug 

8from webapp.config import CATEGORIES 

9 

10 

11logger = logging.getLogger(__name__) 

12 

13session = requests.Session() 

14 

15SOLUTIONS_API_BASE = os.getenv( 

16 "FLASK_SOLUTIONS_API_BASE", "https://solutions.staging.charmhub.io/api" 

17) 

18 

19 

20def get_solution_categories(): 

21 """ 

22 Return the categories a solution can belong to as 

23 ``[{"slug": ..., "name": ...}]``. Categories mirror charm categories 

24 and are sourced from the store API so the two always match. 

25 """ 

26 try: 

27 store_categories = get_store_categories() 

28 except Exception as e: 

29 logger.warning(f"Failed to load solution categories: {e}") 

30 store_categories = [] 

31 

32 categories = [ 

33 { 

34 "slug": cat.get("slug") or cat["name"], 

35 "name": cat.get("display_name") or cat["name"], 

36 } 

37 for cat in store_categories 

38 if cat.get("slug") or cat.get("name") 

39 ] 

40 

41 # fallback to shared category config when store API is unavailable 

42 if not categories: 

43 categories = [ 

44 {"slug": cat["slug"], "name": cat["name"]} for cat in CATEGORIES 

45 ] 

46 

47 return categories 

48 

49 

50def map_category_slugs_to_display(slugs): 

51 """ 

52 Map stored category slugs to ``[{"slug": ..., "name": ...}]`` for 

53 display. Unknown slugs fall back to a title-cased slug. 

54 """ 

55 if not slugs: 

56 return [] 

57 

58 lookup = {cat["slug"]: cat["name"] for cat in get_solution_categories()} 

59 return [ 

60 {"slug": slug, "name": lookup.get(slug) or format_slug(slug)} 

61 for slug in slugs 

62 if slug 

63 ] 

64 

65 

66def get_cached_token(username): 

67 token_key = f"solutions_token_{username}" 

68 

69 if token_key in flask_session: 

70 return flask_session[token_key] 

71 

72 return refresh_token(username) 

73 

74 

75def refresh_token(username): 

76 token_key = f"solutions_token_{username}" 

77 

78 token = login(username) 

79 flask_session[token_key] = token 

80 

81 return token 

82 

83 

84def make_authenticated_request(method, url, username, **kwargs): 

85 token = get_cached_token(username) 

86 

87 headers = kwargs.get("headers", {}) 

88 headers["Authorization"] = f"Bearer {token}" 

89 kwargs["headers"] = headers 

90 

91 response = session.request(method, url, **kwargs) 

92 

93 if response.status_code == 401: 

94 token = refresh_token(username) 

95 headers["Authorization"] = f"Bearer {token}" 

96 response = session.request(method, url, **kwargs) 

97 

98 return response 

99 

100 

101def get_solution_from_backend(uuid, prefer_authenticated=False): 

102 try: 

103 # First try the authenticated publisher response when available 

104 # so edit forms receive publisher-only fields such as creator contact details 

105 try: 

106 username = flask_session.get("account", {}).get("username") 

107 except RuntimeError: 

108 username = None 

109 

110 if prefer_authenticated and username: 

111 try: 

112 auth_resp = make_authenticated_request( 

113 "GET", 

114 f"{SOLUTIONS_API_BASE}/publisher/solutions/by-hash/{uuid}", 

115 username, 

116 timeout=5, 

117 ) 

118 if auth_resp.status_code == 200: 

119 return auth_resp.json() 

120 except Exception as e: 

121 logger.exception( 

122 f"Failed to fetch authenticated solution data: {e}" 

123 ) 

124 

125 # Then try public preview endpoint for published/bearer-link previews 

126 resp = session.get( 

127 f"{SOLUTIONS_API_BASE}/solutions/preview/{uuid}", timeout=5 

128 ) 

129 if resp.status_code == 200: 

130 return resp.json() 

131 except Exception as e: 

132 logger.exception(f"Failed to fetch solution from backend: {e}") 

133 return None 

134 

135 

136def get_published_solution_by_name(name): 

137 try: 

138 resp = session.get(f"{SOLUTIONS_API_BASE}/solutions/{name}", timeout=5) 

139 if resp.status_code == 200: 

140 return resp.json() 

141 except Exception as e: 

142 logger.exception(f"Failed to fetch published solution by name: {e}") 

143 return None 

144 

145 

146def solution_name_exists(name): 

147 try: 

148 resp = session.get( 

149 f"{SOLUTIONS_API_BASE}/solutions/check-name/{name}", timeout=5 

150 ) 

151 if resp.status_code == 200: 

152 data = resp.json() 

153 return data.get("exists", False) 

154 except Exception as e: 

155 logger.exception(f"Failed to check if solution name exists: {e}") 

156 

157 return False 

158 

159 

160def get_publisher_solutions(username): 

161 try: 

162 resp = make_authenticated_request( 

163 "GET", 

164 f"{SOLUTIONS_API_BASE}/publisher/solutions", 

165 username, 

166 timeout=5, 

167 ) 

168 if resp.status_code == 200: 

169 solutions = resp.json() 

170 return solutions if solutions else [] 

171 

172 except Exception as e: 

173 logger.exception(f"Failed to fetch publisher solutions: {e}") 

174 

175 return [] 

176 

177 

178def register_solution(username, data): 

179 try: 

180 resp = make_authenticated_request( 

181 "POST", 

182 f"{SOLUTIONS_API_BASE}/publisher/solutions", 

183 username, 

184 json=data, 

185 timeout=10, 

186 ) 

187 except Exception as e: 

188 logger.exception(f"Failed to communicate with solutions service: {e}") 

189 return {"error": "Failed to communicate with solutions service"} 

190 

191 if resp.status_code == 201: 

192 return resp.json() 

193 

194 if resp.status_code == 400: 

195 try: 

196 error_data = resp.json() 

197 return ( 

198 error_data 

199 if "error-list" in error_data 

200 else {"error": error_data.get("error", "Invalid request data")} 

201 ) 

202 except Exception as e: 

203 logger.exception(f"Failed to parse error response from API: {e}") 

204 return {"error": f"API error (400): {resp.text}"} 

205 

206 return {"error": f"API error ({resp.status_code}): {resp.text}"} 

207 

208 

209def update_solution(username, name, revision, data, submit=True): 

210 if not submit: 

211 data = {**data, "submit_for_review": False} 

212 

213 try: 

214 resp = make_authenticated_request( 

215 "PATCH", 

216 f"{SOLUTIONS_API_BASE}/publisher/solutions/{name}/{revision}", 

217 username, 

218 json=data, 

219 timeout=10, 

220 ) 

221 except Exception as e: 

222 logger.exception(f"Failed to communicate with solutions service: {e}") 

223 return {"error": "Failed to communicate with solutions service"} 

224 

225 if resp.status_code == 200: 

226 return resp.json() 

227 

228 if resp.status_code == 400: 

229 try: 

230 error_data = resp.json() 

231 return ( 

232 error_data 

233 if "error-list" in error_data 

234 else {"error": error_data.get("error", "Invalid request data")} 

235 ) 

236 except Exception as e: 

237 logger.exception(f"Failed to parse error response from API: {e}") 

238 return {"error": f"API error (400): {resp.text}"} 

239 

240 return {"error": f"API error ({resp.status_code}): {resp.text}"} 

241 

242 

243def get_user_teams_for_solutions(username): 

244 """ 

245 Gets LP groups of a user so they can choose 

246 which group to publish the solution under 

247 """ 

248 try: 

249 resp = make_authenticated_request( 

250 "GET", 

251 f"{SOLUTIONS_API_BASE}/me", 

252 username, 

253 timeout=5, 

254 ) 

255 

256 if resp.status_code == 200: 

257 user_data = resp.json() 

258 teams = user_data.get("user", {}).get("teams", []) 

259 return sorted(teams) 

260 

261 except Exception as e: 

262 logger.exception(f"Failed to fetch user teams for solutions: {e}") 

263 

264 return [] 

265 

266 

267def group_solution_drafts(solutions): 

268 published_names = { 

269 solution["name"] 

270 for solution in solutions 

271 if solution.get("status") == "published" 

272 } 

273 drafts_by_name = { 

274 solution["name"]: solution 

275 for solution in solutions 

276 if solution.get("status") == "draft" and solution.get("revision", 1) > 1 

277 } 

278 

279 for solution in solutions: 

280 draft = drafts_by_name.get(solution["name"]) 

281 if solution.get("status") == "published" and draft: 

282 solution["draft_update"] = { 

283 "hash": draft["hash"], 

284 "revision": draft["revision"], 

285 "last_updated": draft.get("last_updated"), 

286 } 

287 elif solution.get("status") == "draft" and solution.get("revision", 1) > 1: 

288 solution["is_draft_update"] = True 

289 

290 return [ 

291 solution 

292 for solution in solutions 

293 if not ( 

294 solution.get("is_draft_update") 

295 and solution.get("name") in published_names 

296 ) 

297 ]