Coverage for webapp/publisher/views.py: 82%
283 statements
« prev ^ index » next coverage.py v7.9.1, created at 2025-06-27 22:07 +0000
1from canonicalwebteam.exceptions import StoreApiResponseErrorList
2from flask import (
3 Blueprint,
4 flash,
5 redirect,
6 render_template,
7 request,
8 session,
9 url_for,
10 make_response,
11)
12from flask.json import jsonify
13from webapp.config import DETAILS_VIEW_REGEX
14from webapp.decorators import login_required, cached_redirect
15from webapp.publisher.logic import get_all_architectures, process_releases
16from webapp.observability.utils import trace_function
17from webapp.store_api import publisher_gateway
19publisher = Blueprint(
20 "publisher",
21 __name__,
22 template_folder="/templates",
23 static_folder="/static",
24)
27@trace_function
28@publisher.route("/account/details")
29@login_required
30def get_account_details():
31 return render_template("publisher/account-details.html")
34@trace_function
35@publisher.route(
36 '/<regex("'
37 + DETAILS_VIEW_REGEX
38 + '"):entity_name>/'
39 + '<regex("listing|releases|publicise|collaboration|settings"):path>'
40)
41@login_required
42def get_publisher(entity_name, path):
43 session["developer_token"] = session["account-auth"]
44 package = publisher_gateway.get_package_metadata(session, entity_name)
46 context = {
47 "package": package,
48 }
50 return render_template("publisher/publisher.html", **context)
53@trace_function
54@publisher.route(
55 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>',
56)
57@login_required
58def get_package(entity_name):
59 session["developer_token"] = session["account-auth"]
60 package = publisher_gateway.get_package_metadata(session, entity_name)
62 return jsonify({"data": package, "success": True})
65@trace_function
66@publisher.route(
67 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>',
68 methods=["PATCH"],
69)
70@login_required
71def update_package(entity_name):
72 payload = request.get_json()
74 res = {}
76 try:
77 package = publisher_gateway.update_package_metadata(
78 session["account-auth"], "charm", entity_name, payload
79 )
80 res["data"] = package
81 res["success"] = True
82 res["message"] = ""
83 response = make_response(res, 200)
84 except StoreApiResponseErrorList as error_list:
85 error_messages = [
86 f"{error.get('message', 'Unable to update this charm or bundle')}"
87 for error in error_list.errors
88 ]
89 if "unauthorized" in error_messages:
90 res["message"] = "Package not found"
91 else:
92 res["message"] = " ".join(error_messages)
93 res["success"] = False
94 response = make_response(res, 500)
96 return response
99@trace_function
100@publisher.route("/charms")
101@publisher.route("/bundles")
102@login_required
103def list_page():
104 publisher_charms = publisher_gateway.get_account_packages(
105 session["account-auth"], "charm", include_collaborations=True
106 )
108 page_type = request.path[1:-1]
110 context = {
111 "published": [
112 {**c, "is_owner": c["publisher"]["id"] == session["account"]["id"]}
113 for c in publisher_charms
114 if c["status"] == "published" and c["type"] == page_type
115 ],
116 "registered": [
117 {**c, "is_owner": c["publisher"]["id"] == session["account"]["id"]}
118 for c in publisher_charms
119 if c["status"] == "registered" and c["type"] == page_type
120 ],
121 "page_type": page_type,
122 }
124 return render_template("publisher/list.html", **context)
127@trace_function
128@publisher.route("/accept-invite")
129@login_required
130@cached_redirect
131def accept_invite():
132 return render_template("publisher/accept-invite.html")
135@trace_function
136@publisher.route("/accept-invite", methods=["POST"])
137@login_required
138def accept_post_invite():
139 res = {}
141 try:
142 token = request.form.get("token")
143 package = request.form.get("package")
144 response = publisher_gateway.accept_invite(
145 session["account-auth"], package, token
146 )
148 if response.status_code == 204:
149 res["success"] = True
150 return make_response(res, 200)
151 else:
152 res["success"] = False
153 errors = response.json().get("error-list", [])
154 res["message"] = (
155 errors[0].get("message") if errors else "Unknown error"
156 )
157 return make_response(res, 500)
159 except StoreApiResponseErrorList as error_list:
160 res["success"] = False
161 error_messages = [
162 f"{error.get('message', 'An error occured')}"
163 for error in error_list.errors
164 ]
165 res["message"] = " ".join(error_messages)
166 except Exception:
167 res["success"] = False
168 res["message"] = "An error occured"
170 return make_response(res, 500)
173@trace_function
174@publisher.route("/reject-invite", methods=["POST"])
175@login_required
176def reject_post_invite():
177 res = {}
179 try:
180 token = request.form.get("token")
181 package = request.form.get("package")
182 response = publisher_gateway.reject_invite(
183 session["account-auth"], package, token
184 )
186 if response.status_code == 204:
187 res["success"] = True
188 return make_response(res, 200)
189 else:
190 res["success"] = False
191 res["message"] = "An error occured"
192 return make_response(res, 200)
194 except StoreApiResponseErrorList as error_list:
195 res["success"] = False
196 error_messages = [
197 f"{error.get('message', 'An error occured')}"
198 for error in error_list.errors
199 ]
200 res["message"] = " ".join(error_messages)
201 response = make_response(res, 500)
202 except Exception:
203 res["success"] = False
204 res["message"] = "An error occured"
206 return response
209@trace_function
210@publisher.route(
211 '/api/packages/<regex("'
212 + DETAILS_VIEW_REGEX
213 + '"):entity_name>/collaborators',
214)
215@login_required
216def get_collaborators(entity_name):
217 res = {}
219 try:
220 collaborators = publisher_gateway.get_collaborators(
221 session["account-auth"], entity_name
222 )
223 res["success"] = True
224 res["data"] = collaborators
225 response = make_response(res, 200)
226 except StoreApiResponseErrorList as error_list:
227 error_messages = [
228 f"{error.get('message', 'An error occured')}"
229 for error in error_list.errors
230 ]
231 res["message"] = " ".join(error_messages)
232 res["success"] = False
233 response = make_response(res, 500)
235 return response
238@trace_function
239@publisher.route(
240 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/invites',
241)
242@login_required
243def get_pending_invites(entity_name):
244 res = {}
246 try:
247 invites = publisher_gateway.get_pending_invites(
248 session["account-auth"], entity_name
249 )
250 res["success"] = True
251 res["data"] = invites["invites"]
252 response = make_response(res, 200)
253 except StoreApiResponseErrorList as error_list:
254 error_messages = [
255 f"{error.get('message', 'An error occured')}"
256 for error in error_list.errors
257 ]
258 res["message"] = " ".join(error_messages)
259 res["success"] = False
260 response = make_response(res, 500)
262 return response
265@trace_function
266@publisher.route(
267 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/invites',
268 methods=["POST"],
269)
270@login_required
271def invite_collaborators(entity_name):
272 res = {}
274 try:
275 collaborators = request.form.get("collaborators")
276 result = publisher_gateway.invite_collaborators(
277 session["account-auth"], entity_name, [collaborators]
278 )
279 res["success"] = True
280 res["data"] = result["tokens"]
281 return make_response(res, 200)
282 except StoreApiResponseErrorList as error_list:
283 res["success"] = False
284 messages = [
285 f"{error.get('message', 'An error occurred')}"
286 for error in error_list.errors
287 ]
288 res["message"] = (" ").join(messages)
289 except Exception:
290 res["success"] = False
291 res["message"] = "An error occurred"
293 return make_response(res, 500)
296@trace_function
297@publisher.route(
298 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/invites',
299 methods=["DELETE"],
300)
301@login_required
302def revoke_invite(entity_name):
303 res = {}
305 try:
306 collaborator = request.form.get("collaborator")
307 response = publisher_gateway.revoke_invites(
308 session["account-auth"], entity_name, [collaborator]
309 )
311 if response.status_code == 204:
312 res["success"] = True
313 return make_response(res, 200)
314 else:
315 res["success"] = False
316 res["message"] = "An error occurred"
317 return make_response(res, 500)
319 except StoreApiResponseErrorList as error_list:
320 res["success"] = False
321 messages = [
322 f"{error.get('message', 'An error occurred')}"
323 for error in error_list.errors
324 ]
325 res["message"] = (" ").join(messages)
327 return make_response(res, 500)
330@trace_function
331@publisher.route("/register-name")
332@login_required
333def register_name():
334 entity_name = request.args.get("entity_name", default="", type=str)
336 invalid_name_str = request.args.get(
337 "invalid_name", default="False", type=str
338 )
339 invalid_name = invalid_name_str == "True"
341 reserved_name_str = request.args.get(
342 "reserved_name", default="False", type=str
343 )
344 reserved_name = reserved_name_str == "True"
346 already_registered_str = request.args.get(
347 "already_registered", default="False", type=str
348 )
349 already_registered = already_registered_str == "True"
351 already_owned_str = request.args.get(
352 "already_owned", default="False", type=str
353 )
354 already_owned = already_owned_str == "True"
356 context = {
357 "entity_name": entity_name,
358 "reserved_name": reserved_name,
359 "invalid_name": invalid_name,
360 "already_owned": already_owned,
361 "already_registered": already_registered,
362 }
363 return render_template("publisher/register-name.html", **context)
366@trace_function
367@publisher.route("/register-name", methods=["POST"])
368@login_required
369def post_register_name():
370 VALID_TYPES = {"charm", "bundle"}
371 data = {
372 "name": request.form["name"],
373 "type": request.form["type"],
374 "private": True if request.form.get("private") == "private" else False,
375 }
377 if data["type"] not in VALID_TYPES:
378 flash("Invalid type specified.", "negative")
379 return redirect(url_for(".register_name"))
381 try:
382 result = publisher_gateway.register_package_name(
383 session["account-auth"], data
384 )
385 if result:
386 flash(
387 f"Your {data['type']} name has been successfully registered.",
388 "positive",
389 )
390 except StoreApiResponseErrorList as api_response_error_list:
391 for error in api_response_error_list.errors:
392 if error["code"] == "api-error":
393 return redirect(
394 url_for(
395 ".register_name",
396 entity_name=data["name"],
397 invalid_name=True,
398 )
399 )
400 elif error["code"] == "reserved-name":
401 return redirect(
402 url_for(
403 ".register_name",
404 entity_name=data["name"],
405 reserved_name=True,
406 )
407 )
408 elif error["code"] == "already-registered":
409 return redirect(
410 url_for(
411 ".register_name",
412 entity_name=data["name"],
413 already_registered=True,
414 )
415 )
416 elif error["code"] == "already-owned":
417 return redirect(
418 url_for(
419 ".register_name",
420 entity_name=data["name"],
421 already_owned=True,
422 )
423 )
425 if data["type"] == "charm":
426 return redirect("/charms")
427 elif data["type"] == "bundle":
428 return redirect("/bundles")
431@trace_function
432@publisher.route("/register-name-dispute")
433@login_required
434def register_name_dispute():
435 entity_name = request.args.get("entity-name", type=str)
437 if not entity_name:
438 return redirect(url_for(".register_name", entity_name=entity_name))
440 return render_template(
441 "publisher/register-name-dispute/index.html", entity_name=entity_name
442 )
445@trace_function
446@publisher.route("/register-name-dispute/thank-you")
447@login_required
448def register_name_dispute_thank_you():
449 entity_name = request.args.get("entity-name", type=str)
451 if not entity_name:
452 return redirect(url_for(".register_name", entity_name=entity_name))
454 return render_template(
455 "publisher/register-name-dispute/thank-you.html",
456 entity_name=entity_name,
457 )
460@trace_function
461@publisher.route("/packages/<package_name>", methods=["DELETE"])
462@login_required
463def delete_package(package_name):
464 resp = publisher_gateway.unregister_package_name(
465 session["account-auth"], package_name
466 )
467 if resp.status_code == 200:
468 return ("", 200)
469 return (
470 jsonify({"error": resp.json["error-list"][0]["message"]}),
471 resp.status_code,
472 )
475@trace_function
476@publisher.route("/<charm_name>/create-track", methods=["POST"])
477@login_required
478def post_create_track(charm_name):
479 track_name = request.form.get("track-name")
480 version_pattern = request.form.get("version-pattern")
481 auto_phasing_percentage = request.form.get("auto-phasing-percentage")
483 if auto_phasing_percentage is not None:
484 auto_phasing_percentage = float(auto_phasing_percentage)
486 response = publisher_gateway.create_track(
487 session,
488 charm_name,
489 track_name,
490 version_pattern,
491 auto_phasing_percentage,
492 )
493 if response.status_code == 201:
494 return response.json(), response.status_code
495 if response.status_code == 409:
496 return (
497 jsonify({"error": "Track already exists."}),
498 response.status_code,
499 )
500 if "error-list" in response.json():
501 return (
502 jsonify({"error": response.json()["error-list"][0]["message"]}),
503 response.status_code,
504 )
505 return response.json(), response.status_code
508@trace_function
509@publisher.route(
510 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/releases',
511)
512@login_required
513def get_releases(entity_name: str):
514 res = {}
516 try:
517 release_data = publisher_gateway.get_releases(
518 session["account-auth"], entity_name
519 )
520 res["success"] = True
522 res["data"] = {}
524 res["data"]["releases"] = process_releases(
525 release_data["channel-map"],
526 release_data["package"]["channels"],
527 release_data["revisions"],
528 )
529 res["data"]["all_architectures"] = get_all_architectures(
530 res["data"]["releases"]
531 )
532 response = make_response(res, 200)
534 except StoreApiResponseErrorList as error_list:
535 error_messages = [
536 f"{error.get('message', 'An error occured')}"
537 for error in error_list.errors
538 ]
539 res["message"] = " ".join(error_messages)
540 res["success"] = False
541 response = make_response(res, 500)
543 return response