Coverage for webapp/publisher/views.py: 81%
293 statements
« prev ^ index » next coverage.py v7.10.2, created at 2025-08-08 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
18from webapp.utils.emailer import get_emailer
20publisher = Blueprint(
21 "publisher",
22 __name__,
23 template_folder="/templates",
24 static_folder="/static",
25)
28@trace_function
29@publisher.route("/account/details")
30@login_required
31def get_account_details():
32 return render_template("publisher/account-details.html")
35@trace_function
36@publisher.route(
37 '/<regex("'
38 + DETAILS_VIEW_REGEX
39 + '"):entity_name>/'
40 + '<regex("listing|releases|publicise|collaboration|settings"):path>'
41)
42@login_required
43def get_publisher(entity_name, path):
44 session["developer_token"] = session["account-auth"]
45 package = publisher_gateway.get_package_metadata(session, entity_name)
47 context = {
48 "package": package,
49 }
51 return render_template("publisher/publisher.html", **context)
54@trace_function
55@publisher.route(
56 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>',
57)
58@login_required
59def get_package(entity_name):
60 session["developer_token"] = session["account-auth"]
61 package = publisher_gateway.get_package_metadata(session, entity_name)
63 return jsonify({"data": package, "success": True})
66@trace_function
67@publisher.route(
68 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>',
69 methods=["PATCH"],
70)
71@login_required
72def update_package(entity_name):
73 payload = request.get_json()
75 res = {}
77 try:
78 package = publisher_gateway.update_package_metadata(
79 session["account-auth"], "charm", entity_name, payload
80 )
81 res["data"] = package
82 res["success"] = True
83 res["message"] = ""
84 response = make_response(res, 200)
85 except StoreApiResponseErrorList as error_list:
86 error_messages = [
87 f"{error.get('message', 'Unable to update this charm or bundle')}"
88 for error in error_list.errors
89 ]
90 if "unauthorized" in error_messages:
91 res["message"] = "Package not found"
92 else:
93 res["message"] = " ".join(error_messages)
94 res["success"] = False
95 response = make_response(res, 500)
97 return response
100@trace_function
101@publisher.route("/charms")
102@publisher.route("/bundles")
103@login_required
104def list_page():
105 publisher_charms = publisher_gateway.get_account_packages(
106 session["account-auth"], "charm", include_collaborations=True
107 )
109 page_type = request.path[1:-1]
111 context = {
112 "published": [
113 {**c, "is_owner": c["publisher"]["id"] == session["account"]["id"]}
114 for c in publisher_charms
115 if c["status"] == "published" and c["type"] == page_type
116 ],
117 "registered": [
118 {**c, "is_owner": c["publisher"]["id"] == session["account"]["id"]}
119 for c in publisher_charms
120 if c["status"] == "registered" and c["type"] == page_type
121 ],
122 "page_type": page_type,
123 }
125 return render_template("publisher/list.html", **context)
128@trace_function
129@publisher.route("/accept-invite")
130@login_required
131@cached_redirect
132def accept_invite():
133 return render_template("publisher/accept-invite.html")
136@trace_function
137@publisher.route("/accept-invite", methods=["POST"])
138@login_required
139def accept_post_invite():
140 res = {}
142 try:
143 token = request.form.get("token")
144 package = request.form.get("package")
145 response = publisher_gateway.accept_invite(
146 session["account-auth"], package, token
147 )
149 if response.status_code == 204:
150 res["success"] = True
151 return make_response(res, 200)
152 else:
153 res["success"] = False
154 errors = response.json().get("error-list", [])
155 res["message"] = (
156 errors[0].get("message") if errors else "Unknown error"
157 )
158 return make_response(res, 500)
160 except StoreApiResponseErrorList as error_list:
161 res["success"] = False
162 error_messages = [
163 f"{error.get('message', 'An error occured')}"
164 for error in error_list.errors
165 ]
166 res["message"] = " ".join(error_messages)
167 except Exception:
168 res["success"] = False
169 res["message"] = "An error occured"
171 return make_response(res, 500)
174@trace_function
175@publisher.route("/reject-invite", methods=["POST"])
176@login_required
177def reject_post_invite():
178 res = {}
180 try:
181 token = request.form.get("token")
182 package = request.form.get("package")
183 response = publisher_gateway.reject_invite(
184 session["account-auth"], package, token
185 )
187 if response.status_code == 204:
188 res["success"] = True
189 return make_response(res, 200)
190 else:
191 res["success"] = False
192 res["message"] = "An error occured"
193 return make_response(res, 200)
195 except StoreApiResponseErrorList as error_list:
196 res["success"] = False
197 error_messages = [
198 f"{error.get('message', 'An error occured')}"
199 for error in error_list.errors
200 ]
201 res["message"] = " ".join(error_messages)
202 response = make_response(res, 500)
203 except Exception:
204 res["success"] = False
205 res["message"] = "An error occured"
207 return response
210@trace_function
211@publisher.route(
212 '/api/packages/<regex("'
213 + DETAILS_VIEW_REGEX
214 + '"):entity_name>/collaborators',
215)
216@login_required
217def get_collaborators(entity_name):
218 res = {}
220 try:
221 collaborators = publisher_gateway.get_collaborators(
222 session["account-auth"], entity_name
223 )
224 res["success"] = True
225 res["data"] = collaborators
226 response = make_response(res, 200)
227 except StoreApiResponseErrorList as error_list:
228 error_messages = [
229 f"{error.get('message', 'An error occured')}"
230 for error in error_list.errors
231 ]
232 res["message"] = " ".join(error_messages)
233 res["success"] = False
234 response = make_response(res, 500)
236 return response
239@trace_function
240@publisher.route(
241 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/invites',
242)
243@login_required
244def get_pending_invites(entity_name):
245 res = {}
247 try:
248 invites = publisher_gateway.get_pending_invites(
249 session["account-auth"], entity_name
250 )
251 res["success"] = True
252 res["data"] = invites["invites"]
253 response = make_response(res, 200)
254 except StoreApiResponseErrorList as error_list:
255 error_messages = [
256 f"{error.get('message', 'An error occured')}"
257 for error in error_list.errors
258 ]
259 res["message"] = " ".join(error_messages)
260 res["success"] = False
261 response = make_response(res, 500)
263 return response
266@trace_function
267@publisher.route(
268 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/invites',
269 methods=["POST"],
270)
271@login_required
272def invite_collaborators(entity_name):
273 res = {}
275 try:
276 collaborators = request.form.get("collaborators")
277 if not collaborators:
278 res["success"] = False
279 res["message"] = "No collaborators provided"
280 return make_response(res, 400)
282 result = publisher_gateway.invite_collaborators(
283 session["account-auth"], entity_name, [collaborators]
284 )
286 token = result["tokens"][0]["token"]
288 invite_link = (
289 f"https://charmhub.io/accept-invite?package={entity_name}"
290 f"&token={token}"
291 )
292 emailer = get_emailer()
293 emailer.send_email_template(
294 template_path="emails/collaborator-invite.html",
295 to_email=collaborators,
296 context={
297 "charm_name": entity_name,
298 "invite_link": invite_link,
299 },
300 subject=(
301 f"You have been invited to as a collaborator on "
302 f"{entity_name} in Charmhub"
303 ),
304 )
306 res["success"] = True
307 res["data"] = result["tokens"]
308 return make_response(res, 200)
309 except StoreApiResponseErrorList as error_list:
310 res["success"] = False
311 messages = [
312 f"{error.get('message', 'An error occurred')}"
313 for error in error_list.errors
314 ]
315 res["message"] = (" ").join(messages)
316 except Exception:
317 res["success"] = False
318 res["message"] = "An error occurred"
319 raise
321 return make_response(res, 500)
324@trace_function
325@publisher.route(
326 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/invites',
327 methods=["DELETE"],
328)
329@login_required
330def revoke_invite(entity_name):
331 res = {}
333 try:
334 collaborator = request.form.get("collaborator")
335 response = publisher_gateway.revoke_invites(
336 session["account-auth"], entity_name, [collaborator]
337 )
339 if response.status_code == 204:
340 res["success"] = True
341 return make_response(res, 200)
342 else:
343 res["success"] = False
344 res["message"] = "An error occurred"
345 return make_response(res, 500)
347 except StoreApiResponseErrorList as error_list:
348 res["success"] = False
349 messages = [
350 f"{error.get('message', 'An error occurred')}"
351 for error in error_list.errors
352 ]
353 res["message"] = (" ").join(messages)
355 return make_response(res, 500)
358@trace_function
359@publisher.route("/register-name")
360@login_required
361def register_name():
362 entity_name = request.args.get("entity_name", default="", type=str)
364 invalid_name_str = request.args.get(
365 "invalid_name", default="False", type=str
366 )
367 invalid_name = invalid_name_str == "True"
369 reserved_name_str = request.args.get(
370 "reserved_name", default="False", type=str
371 )
372 reserved_name = reserved_name_str == "True"
374 already_registered_str = request.args.get(
375 "already_registered", default="False", type=str
376 )
377 already_registered = already_registered_str == "True"
379 already_owned_str = request.args.get(
380 "already_owned", default="False", type=str
381 )
382 already_owned = already_owned_str == "True"
384 context = {
385 "entity_name": entity_name,
386 "reserved_name": reserved_name,
387 "invalid_name": invalid_name,
388 "already_owned": already_owned,
389 "already_registered": already_registered,
390 }
391 return render_template("publisher/register-name.html", **context)
394@trace_function
395@publisher.route("/register-name", methods=["POST"])
396@login_required
397def post_register_name():
398 VALID_TYPES = {"charm", "bundle"}
399 data = {
400 "name": request.form["name"],
401 "type": request.form["type"],
402 "private": True if request.form.get("private") == "private" else False,
403 }
405 if data["type"] not in VALID_TYPES:
406 flash("Invalid type specified.", "negative")
407 return redirect(url_for(".register_name"))
409 try:
410 result = publisher_gateway.register_package_name(
411 session["account-auth"], data
412 )
413 if result:
414 flash(
415 f"Your {data['type']} name has been successfully registered.",
416 "positive",
417 )
418 except StoreApiResponseErrorList as api_response_error_list:
419 for error in api_response_error_list.errors:
420 if error["code"] == "api-error":
421 return redirect(
422 url_for(
423 ".register_name",
424 entity_name=data["name"],
425 invalid_name=True,
426 )
427 )
428 elif error["code"] == "reserved-name":
429 return redirect(
430 url_for(
431 ".register_name",
432 entity_name=data["name"],
433 reserved_name=True,
434 )
435 )
436 elif error["code"] == "already-registered":
437 return redirect(
438 url_for(
439 ".register_name",
440 entity_name=data["name"],
441 already_registered=True,
442 )
443 )
444 elif error["code"] == "already-owned":
445 return redirect(
446 url_for(
447 ".register_name",
448 entity_name=data["name"],
449 already_owned=True,
450 )
451 )
453 if data["type"] == "charm":
454 return redirect("/charms")
455 elif data["type"] == "bundle":
456 return redirect("/bundles")
459@trace_function
460@publisher.route("/register-name-dispute")
461@login_required
462def register_name_dispute():
463 entity_name = request.args.get("entity-name", type=str)
465 if not entity_name:
466 return redirect(url_for(".register_name", entity_name=entity_name))
468 return render_template(
469 "publisher/register-name-dispute/index.html", entity_name=entity_name
470 )
473@trace_function
474@publisher.route("/register-name-dispute/thank-you")
475@login_required
476def register_name_dispute_thank_you():
477 entity_name = request.args.get("entity-name", type=str)
479 if not entity_name:
480 return redirect(url_for(".register_name", entity_name=entity_name))
482 return render_template(
483 "publisher/register-name-dispute/thank-you.html",
484 entity_name=entity_name,
485 )
488@trace_function
489@publisher.route("/packages/<package_name>", methods=["DELETE"])
490@login_required
491def delete_package(package_name):
492 resp = publisher_gateway.unregister_package_name(
493 session["account-auth"], package_name
494 )
495 if resp.status_code == 200:
496 return ("", 200)
497 return (
498 jsonify({"error": resp.json["error-list"][0]["message"]}),
499 resp.status_code,
500 )
503@trace_function
504@publisher.route("/<charm_name>/create-track", methods=["POST"])
505@login_required
506def post_create_track(charm_name):
507 track_name = request.form.get("track-name")
508 version_pattern = request.form.get("version-pattern")
509 auto_phasing_percentage = request.form.get("auto-phasing-percentage")
511 if auto_phasing_percentage is not None:
512 auto_phasing_percentage = float(auto_phasing_percentage)
514 response = publisher_gateway.create_track(
515 session,
516 charm_name,
517 track_name,
518 version_pattern,
519 auto_phasing_percentage,
520 )
521 if response.status_code == 201:
522 return response.json(), response.status_code
523 if response.status_code == 409:
524 return (
525 jsonify({"error": "Track already exists."}),
526 response.status_code,
527 )
528 if "error-list" in response.json():
529 return (
530 jsonify({"error": response.json()["error-list"][0]["message"]}),
531 response.status_code,
532 )
533 return response.json(), response.status_code
536@trace_function
537@publisher.route(
538 '/api/packages/<regex("' + DETAILS_VIEW_REGEX + '"):entity_name>/releases',
539)
540@login_required
541def get_releases(entity_name: str):
542 res = {}
544 try:
545 release_data = publisher_gateway.get_releases(
546 session["account-auth"], entity_name
547 )
548 res["success"] = True
550 res["data"] = {}
552 res["data"]["releases"] = process_releases(
553 release_data["channel-map"],
554 release_data["package"]["channels"],
555 release_data["revisions"],
556 )
557 res["data"]["all_architectures"] = get_all_architectures(
558 res["data"]["releases"]
559 )
560 response = make_response(res, 200)
562 except StoreApiResponseErrorList as error_list:
563 error_messages = [
564 f"{error.get('message', 'An error occured')}"
565 for error in error_list.errors
566 ]
567 res["message"] = " ".join(error_messages)
568 res["success"] = False
569 response = make_response(res, 500)
571 return response