Coverage for webapp / store / views.py: 53%

1from math import floor 

2 

3import requests as api_requests 

4import flask 

5from dateutil import parser 

6from webapp.decorators import exchange_required, login_required 

7import webapp.helpers as helpers 

8import webapp.store.logic as logic 

9from webapp.api import requests 

10 

11from canonicalwebteam.exceptions import StoreApiError 

12from canonicalwebteam.store_api.dashboard import Dashboard 

13from canonicalwebteam.store_api.publishergw import PublisherGW 

14from canonicalwebteam.store_api.devicegw import DeviceGW 

15from canonicalwebteam.snap_recommendations import SnapRecommendations 

16 

17from webapp.api.exceptions import ApiError 

18from webapp.store.snap_details_views import snap_details_views 

19from webapp.helpers import api_publisher_session, api_session 

20from flask.json import jsonify 

21from webapp.extensions import csrf 

22from webapp.store.logic import ( 

23 get_categories, 

24) 

25from cache.cache_utility import redis_cache 

26 

27session = requests.Session() 

28 

29dashboard = Dashboard(api_session) 

30publisher_gateway = PublisherGW("snap", api_publisher_session) 

31device_gateway = DeviceGW("snap", api_session) 

32snap_recommendations = SnapRecommendations(session) 

33 

34 

35def store_blueprint(store_query=None): 

36 store = flask.Blueprint( 

37 "store", 

38 __name__, 

39 template_folder="/templates", 

40 static_folder="/static", 

41 ) 

42 snap_details_views(store) 

43 

44 @store.route("/validation-sets", defaults={"path": ""}) 

45 @store.route("/validation-sets/<path:path>") 

46 @login_required 

47 def validation_sets(path): 

48 return flask.render_template("store/publisher.html") 

49 

50 @store.route("/discover") 

51 def discover(): 

52 return flask.redirect(flask.url_for(".homepage")) 

53 

54 def brand_store_view(): 

55 error_info = {} 

56 status_code = 200 

57 

58 try: 

59 snaps = device_gateway.get_all_items(size=16)["results"] 

60 except (StoreApiError, ApiError): 

61 snaps = [] 

62 

63 for snap in snaps: 

64 if "media" in snap: 

65 snap["icon_url"] = helpers.get_icon(snap["media"]) 

66 

67 return ( 

68 flask.render_template( 

69 "brand-store/store.html", snaps=snaps, error_info=error_info 

70 ), 

71 status_code, 

72 ) 

73 

74 def brand_search_snap(): 

75 status_code = 200 

76 snap_searched = flask.request.args.get("q", default="", type=str) 

77 

78 if not snap_searched: 

79 return flask.redirect(flask.url_for(".homepage")) 

80 

81 size = flask.request.args.get("limit", default=25, type=int) 

82 offset = flask.request.args.get("offset", default=0, type=int) 

83 

84 try: 

85 page = floor(offset / size) + 1 

86 except ZeroDivisionError: 

87 size = 10 

88 page = floor(offset / size) + 1 

89 

90 error_info = {} 

91 searched_results = [] 

92 

93 searched_results = device_gateway.search( 

94 snap_searched, size=size, page=page 

95 ) 

96 

97 snaps_results = searched_results["results"] 

98 

99 for snap in snaps_results: 

100 snap["icon_url"] = helpers.get_icon(snap["media"]) 

101 

102 links = logic.get_pages_details( 

103 flask.request.base_url, 

104 ( 

105 searched_results["_links"] 

106 if "_links" in searched_results 

107 else [] 

108 ), 

109 ) 

110 

111 context = { 

112 "query": snap_searched, 

113 "snaps": snaps_results, 

114 "links": links, 

115 "error_info": error_info, 

116 } 

117 

118 return ( 

119 flask.render_template("brand-store/search.html", **context), 

120 status_code, 

121 ) 

122 

123 @store.route("/store") 

124 def store_view(): 

125 return flask.render_template("store/store.html") 

126 

127 @store.route("/explore") 

128 def explore_view(): 

129 try: 

130 popular_snaps = redis_cache.get( 

131 "explore:popular-snaps", expected_type=list 

132 ) 

133 if not popular_snaps: 

134 popular_snaps = snap_recommendations.get_popular() 

135 redis_cache.set( 

136 "explore:popular-snaps", popular_snaps, ttl=3600 

137 ) 

138 except api_requests.exceptions.RequestException: 

139 popular_snaps = [] 

140 

141 try: 

142 recent_snaps = redis_cache.get( 

143 "explore:recent-snaps", expected_type=list 

144 ) 

145 if not recent_snaps: 

146 recent_snaps = snap_recommendations.get_recent() 

147 redis_cache.set("explore:recent-snaps", recent_snaps, ttl=3600) 

148 except api_requests.exceptions.RequestException: 

149 recent_snaps = [] 

150 

151 try: 

152 trending_snaps = redis_cache.get( 

153 "explore:trending-snaps", expected_type=list 

154 ) 

155 if not trending_snaps: 

156 trending_snaps = snap_recommendations.get_trending() 

157 redis_cache.set( 

158 "explore:trending-snaps", trending_snaps, ttl=3600 

159 ) 

160 except api_requests.exceptions.RequestException: 

161 trending_snaps = [] 

162 

163 try: 

164 top_rated_snaps = redis_cache.get( 

165 "explore:top-rated-snaps", expected_type=list 

166 ) 

167 if not top_rated_snaps: 

168 top_rated_snaps = snap_recommendations.get_top_rated() 

169 redis_cache.set( 

170 "explore:top-rated-snaps", top_rated_snaps, ttl=3600 

171 ) 

172 except api_requests.exceptions.RequestException: 

173 top_rated_snaps = [] 

174 

175 try: 

176 categories_results = redis_cache.get( 

177 "explore:categories", expected_type=list 

178 ) 

179 if not categories_results: 

180 categories_results = device_gateway.get_categories() 

181 redis_cache.set( 

182 "explore:categories", categories_results, ttl=3600 

183 ) 

184 except StoreApiError: 

185 categories_results = [] 

186 

187 categories = sorted( 

188 get_categories(categories_results), 

189 key=lambda category: category["slug"], 

190 ) 

191 

192 return flask.render_template( 

193 "explore/index.html", 

194 categories=categories, 

195 popular_snaps=popular_snaps, 

196 recent_snaps=recent_snaps, 

197 trending_snaps=trending_snaps, 

198 top_rated_snaps=top_rated_snaps, 

199 ) 

200 

201 @store.route("/publisher/<regex('[a-z0-9-]*[a-z][a-z0-9-]*'):publisher>") 

202 def publisher_details(publisher): 

203 """ 

204 A view to display the publisher details page for specific publisher. 

205 """ 

206 

207 # 404 for the snap-quarantine publisher 

208 if publisher == "snap-quarantine": 

209 flask.abort(404) 

210 

211 publisher_content_path = flask.current_app.config["CONTENT_DIRECTORY"][ 

212 "PUBLISHER_PAGES" 

213 ] 

214 

215 # special handling for some featured publishers with custom pages 

216 if publisher in ["kde", "snapcrafters", "jetbrains"]: 

217 context = helpers.get_yaml( 

218 publisher_content_path + publisher + ".yaml", typ="safe" 

219 ) 

220 

221 if not context: 

222 flask.abort(404) 

223 

224 popular_snaps = helpers.get_yaml( 

225 publisher_content_path + publisher + "-snaps.yaml", 

226 typ="safe", 

227 ) 

228 

229 context["popular_snaps"] = ( 

230 popular_snaps["snaps"] if popular_snaps else [] 

231 ) 

232 

233 context["snaps"] = [] 

234 snaps_results = [] 

235 try: 

236 snaps_results = device_gateway.find( 

237 publisher=publisher, 

238 fields=[ 

239 "title", 

240 "summary", 

241 "media", 

242 "publisher", 

243 ], 

244 )["results"] 

245 except StoreApiError: 

246 pass # proceed with an empty list 

247 

248 for snap in snaps_results: 

249 item = snap["snap"] 

250 item["package_name"] = snap["name"] 

251 item["icon_url"] = helpers.get_icon(item["media"]) 

252 context["snaps"].append(item) 

253 

254 featured_snaps = [ 

255 snap["package_name"] for snap in context["featured_snaps"] 

256 ] 

257 

258 context["snaps"] = [ 

259 snap 

260 for snap in context["snaps"] 

261 if snap["package_name"] not in featured_snaps 

262 ] 

263 

264 context["snaps_count"] = len(context["snaps"]) + len( 

265 featured_snaps 

266 ) 

267 

268 return flask.render_template( 

269 "store/publisher-details.html", **context 

270 ) 

271 

272 # standard page for all community publishers 

273 status_code = 200 

274 error_info = {} 

275 snaps_results = [] 

276 snaps = [] 

277 snaps_count = 0 

278 publisher_details = {"display-name": publisher, "username": publisher} 

279 

280 snaps_results = device_gateway.find( 

281 publisher=publisher, 

282 fields=[ 

283 "title", 

284 "summary", 

285 "media", 

286 "publisher", 

287 ], 

288 )["results"] 

289 

290 for snap in snaps_results: 

291 item = snap["snap"] 

292 item["package_name"] = snap["name"] 

293 item["icon_url"] = helpers.get_icon(item["media"]) 

294 snaps.append(item) 

295 

296 snaps_count = len(snaps) 

297 

298 if snaps_count > 0: 

299 publisher_details = snaps[0]["publisher"] 

300 

301 context = { 

302 "snaps": snaps, 

303 "snaps_count": snaps_count, 

304 "publisher": publisher_details, 

305 "error_info": error_info, 

306 } 

307 

308 return ( 

309 flask.render_template( 

310 "store/community-publisher-details.html", **context 

311 ), 

312 status_code, 

313 ) 

314 

315 @store.route("/store/categories/<category>") 

316 def store_category(category): 

317 status_code = 200 

318 error_info = {} 

319 snaps_results = [] 

320 

321 snaps_results = device_gateway.get_category_items( 

322 category=category, size=10, page=1 

323 )["results"] 

324 for snap in snaps_results: 

325 snap["icon_url"] = helpers.get_icon(snap["media"]) 

326 

327 # if the first snap (banner snap) doesn't have an icon, remove the last 

328 # snap from the list to avoid a hanging snap (grid of 9) 

329 if len(snaps_results) == 10 and snaps_results[0]["icon_url"] == "": 

330 snaps_results = snaps_results[:-1] 

331 

332 for index in range(len(snaps_results)): 

333 snaps_results[index] = logic.get_snap_banner_url( 

334 snaps_results[index] 

335 ) 

336 

337 context = { 

338 "category": category, 

339 "has_featured": True, 

340 "snaps": snaps_results, 

341 "error_info": error_info, 

342 } 

343 

344 return ( 

345 flask.render_template("store/_category-partial.html", **context), 

346 status_code, 

347 ) 

348 

349 @store.route("/store/featured-snaps/<category>") 

350 def featured_snaps_in_category(category): 

351 snaps_results = [] 

352 

353 snaps_results = device_gateway.get_category_items( 

354 category=category, size=3, page=1 

355 )["_embedded"]["clickindex:package"] 

356 

357 for snap in snaps_results: 

358 snap["icon_url"] = helpers.get_icon(snap["media"]) 

359 

360 return flask.jsonify(snaps_results) 

361 

362 @store.route("/store/sitemap.xml") 

363 def sitemap(): 

364 sitemap = redis_cache.get("sitemap:xml", expected_type=str) 

365 if sitemap: 

366 response = flask.make_response(sitemap) 

367 response.headers["Content-Type"] = "application/xml" 

368 response.headers["Cache-Control"] = "public, max-age=43200" 

369 return response 

370 

371 base_url = "https://snapcraft.io/store" 

372 

373 snaps = [] 

374 page = 0 

375 url = f"https://api.snapcraft.io/api/v1/snaps/search?page={page}" 

376 while url: 

377 response = session.get(url) 

378 try: 

379 snaps_response = response.json() 

380 except Exception: 

381 continue 

382 

383 for snap in snaps_response["_embedded"]["clickindex:package"]: 

384 try: 

385 last_udpated = ( 

386 parser.parse(snap["last_updated"]) 

387 .replace(tzinfo=None) 

388 .strftime("%Y-%m-%d") 

389 ) 

390 snaps.append( 

391 { 

392 "url": "https://snapcraft.io/" 

393 + snap["package_name"], 

394 "last_udpated": last_udpated, 

395 } 

396 ) 

397 except Exception: 

398 continue 

399 if "next" in snaps_response["_links"]: 

400 url = snaps_response["_links"]["next"]["href"] 

401 else: 

402 url = None 

403 

404 xml_sitemap = flask.render_template( 

405 "sitemap/sitemap.xml", 

406 base_url=base_url, 

407 links=snaps, 

408 ) 

409 

410 # Cache the generated sitemap for 12 hours 

411 redis_cache.set("sitemap:xml", xml_sitemap, ttl=43200) 

412 

413 response = flask.make_response(xml_sitemap) 

414 response.headers["Content-Type"] = "application/xml" 

415 response.headers["Cache-Control"] = "public, max-age=43200" 

416 

417 return response 

418 

419 if store_query: 

420 store.add_url_rule("/", "homepage", brand_store_view) 

421 store.add_url_rule("/search", "search", brand_search_snap) 

422 else: 

423 store.add_url_rule("/store", "homepage", store_view) 

424 

425 store.add_url_rule("/explore", "explore", explore_view) 

426 

427 @store.route("/<snap_name>/create-track", methods=["POST"]) 

428 @login_required 

429 @csrf.exempt 

430 @exchange_required 

431 def post_create_track(snap_name): 

432 track_name = flask.request.form["track-name"] 

433 version_pattern = flask.request.form.get("version-pattern") 

434 auto_phasing_percentage = flask.request.form.get( 

435 "automatic-phasing-percentage" 

436 ) 

437 

438 if auto_phasing_percentage is not None: 

439 auto_phasing_percentage = float(auto_phasing_percentage) 

440 

441 response = publisher_gateway.create_track( 

442 flask.session, 

443 snap_name, 

444 track_name, 

445 version_pattern, 

446 auto_phasing_percentage, 

447 ) 

448 if response.status_code == 201: 

449 return response.json(), response.status_code 

450 if response.status_code == 409: 

451 return ( 

452 jsonify({"error": "Track already exists."}), 

453 response.status_code, 

454 ) 

455 if "error-list" in response.json(): 

456 return ( 

457 jsonify( 

458 {"error": response.json()["error-list"][0]["message"]} 

459 ), 

460 response.status_code, 

461 ) 

462 return response.json(), response.status_code 

463 

464 return store