Warning
This document is for an old release of Galaxy. You can alternatively view this page in the latest release if it exists or view the top of the latest release's documentation.
Source code for tool_shed.webapp.controllers.repository_review
import logging
from sqlalchemy import (
and_,
false,
func,
)
import tool_shed.grids.repository_review_grids as repository_review_grids
import tool_shed.grids.util as grids_util
from galaxy import (
util,
web,
)
from galaxy.web.form_builder import CheckboxField
from galaxy.webapps.base.controller import BaseUIController
from tool_shed.util import (
hg_util,
metadata_util,
repository_util,
review_util,
)
from tool_shed.util import shed_util_common as suc
from tool_shed.util.container_util import STRSEP
from tool_shed.util.web_util import escape
from tool_shed.webapp.framework.decorators import require_login
from tool_shed.webapp.util import ratings_util
log = logging.getLogger(__name__)
[docs]class RepositoryReviewController(BaseUIController, ratings_util.ItemRatings):
component_grid = repository_review_grids.ComponentGrid()
repositories_ready_for_review_grid = repository_review_grids.RepositoriesReadyForReviewGrid()
repositories_reviewed_by_me_grid = repository_review_grids.RepositoriesReviewedByMeGrid()
repositories_with_reviews_grid = repository_review_grids.RepositoriesWithReviewsGrid()
repositories_without_reviews_grid = repository_review_grids.RepositoriesWithoutReviewsGrid()
repository_reviews_by_user_grid = repository_review_grids.RepositoryReviewsByUserGrid()
reviewed_repositories_i_own_grid = repository_review_grids.ReviewedRepositoriesIOwnGrid()
repositories_with_no_tool_tests_grid = repository_review_grids.RepositoriesWithNoToolTestsGrid()
[docs] @web.expose
@require_login("approve repository review")
def approve_repository_review(self, trans, **kwd):
# The value of the received id is the encoded review id.
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
encoded_review_id = kwd["id"]
review = review_util.get_review(trans.app, encoded_review_id)
if kwd.get("approve_repository_review_button", False):
approved_select_field_name = f"{encoded_review_id}{STRSEP}approved"
approved_select_field_value = str(kwd[approved_select_field_name])
review.approved = approved_select_field_value
trans.sa_session.add(review)
trans.sa_session.flush()
message = f"Approved value <b>{escape(approved_select_field_value)}</b> saved for this revision."
repository_id = trans.security.encode_id(review.repository_id)
changeset_revision = review.changeset_revision
return trans.response.send_redirect(
web.url_for(
controller="repository_review",
action="manage_repository_reviews_of_revision",
id=repository_id,
changeset_revision=changeset_revision,
message=message,
status=status,
)
)
[docs] @web.expose
@require_login("browse components")
def browse_components(self, trans, **kwd):
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "create":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="create_component", **kwd)
)
return self.component_grid(trans, **kwd)
[docs] @web.expose
@require_login("browse review")
def browse_review(self, trans, **kwd):
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
review = review_util.get_review(trans.app, kwd["id"])
repository = review.repository
repo = repository.hg_repo
rev, changeset_revision_label = hg_util.get_rev_label_from_changeset_revision(repo, review.changeset_revision)
return trans.fill_template(
"/webapps/tool_shed/repository_review/browse_review.mako",
repository=repository,
changeset_revision_label=changeset_revision_label,
review=review,
message=message,
status=status,
)
[docs] def copy_review(self, trans, review_to_copy, review):
for component_review in review_to_copy.component_reviews:
copied_component_review = trans.model.ComponentReview(
repository_review_id=review.id,
component_id=component_review.component.id,
comment=component_review.comment,
private=component_review.private,
approved=component_review.approved,
rating=component_review.rating,
)
trans.sa_session.add(copied_component_review)
trans.sa_session.flush()
review.approved = review_to_copy.approved
review.rating = review_to_copy.rating
trans.sa_session.add(review)
trans.sa_session.flush()
[docs] @web.expose
@require_login("create component")
def create_component(self, trans, **kwd):
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
name = kwd.get("name", "")
description = kwd.get("description", "")
if kwd.get("create_component_button", False):
if not name or not description:
message = "Enter a valid name and a description"
status = "error"
elif review_util.get_component_by_name(trans.app, name):
message = "A component with that name already exists"
status = "error"
else:
component = trans.app.model.Component(name=name, description=description)
trans.sa_session.add(component)
trans.sa_session.flush()
message = f"Component '{escape(component.name)}' has been created"
status = "done"
trans.response.send_redirect(
web.url_for(
controller="repository_review", action="manage_components", message=message, status=status
)
)
return trans.fill_template(
"/webapps/tool_shed/repository_review/create_component.mako",
name=name,
description=description,
message=message,
status=status,
)
[docs] @web.expose
@require_login("create review")
def create_review(self, trans, **kwd):
# The value of the received id is the encoded repository id.
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
repository_id = kwd.get("id", None)
changeset_revision = kwd.get("changeset_revision", None)
previous_review_id = kwd.get("previous_review_id", None)
create_without_copying = "create_without_copying" in kwd
if repository_id:
if changeset_revision:
# Make sure there is not already a review of the revision by the user.
repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id)
if review_util.get_review_by_repository_id_changeset_revision_user_id(
app=trans.app,
repository_id=repository_id,
changeset_revision=changeset_revision,
user_id=trans.security.encode_id(trans.user.id),
):
message = f"You have already created a review for revision <b>{changeset_revision}</b> of repository <b>{escape(repository.name)}</b>."
status = "error"
else:
# See if there are any reviews for previous changeset revisions that the user can copy.
if (
not create_without_copying
and not previous_review_id
and review_util.has_previous_repository_reviews(trans.app, repository, changeset_revision)
):
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="select_previous_review", **kwd)
)
# A review can be initially performed only on an installable revision of a repository, so make sure we have metadata associated
# with the received changeset_revision.
repository_metadata = metadata_util.get_repository_metadata_by_changeset_revision(
trans.app, repository_id, changeset_revision
)
if repository_metadata:
metadata = repository_metadata.metadata
if metadata:
review = trans.app.model.RepositoryReview(
repository_id=repository_metadata.repository_id,
changeset_revision=changeset_revision,
user_id=trans.user.id,
rating=None,
deleted=False,
)
trans.sa_session.add(review)
trans.sa_session.flush()
if previous_review_id:
review_to_copy = review_util.get_review(trans.app, previous_review_id)
self.copy_review(trans, review_to_copy, review)
review_id = trans.security.encode_id(review.id)
message = f"Begin your review of revision <b>{changeset_revision}</b> of repository <b>{repository.name}</b>."
status = "done"
trans.response.send_redirect(
web.url_for(
controller="repository_review",
action="edit_review",
id=review_id,
message=message,
status=status,
)
)
else:
message = (
"A new review cannot be created for revision <b>%s</b> of repository <b>%s</b>. Select a valid revision and try again."
% (changeset_revision, escape(repository.name))
)
kwd["message"] = message
kwd["status"] = "error"
else:
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="manage_repository_reviews", **kwd)
)
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
[docs] @web.expose
@require_login("edit component")
def edit_component(self, trans, **kwd):
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
id = kwd.get("id", None)
if not id:
message = "No component ids received for editing"
trans.response.send_redirect(
web.url_for(controller="admin", action="manage_categories", message=message, status="error")
)
component = review_util.get_component(trans.app, id)
if kwd.get("edit_component_button", False):
new_description = kwd.get("description", "").strip()
if component.description != new_description:
component.description = new_description
trans.sa_session.add(component)
trans.sa_session.flush()
message = f"The information has been saved for the component named <b>{escape(component.name)}</b>"
status = "done"
return trans.response.send_redirect(
web.url_for(
controller="repository_review", action="manage_components", message=message, status=status
)
)
return trans.fill_template(
"/webapps/tool_shed/repository_review/edit_component.mako",
component=component,
message=message,
status=status,
)
[docs] @web.expose
@require_login("edit review")
def edit_review(self, trans, **kwd):
# The value of the received id is the encoded review id.
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
review_id = kwd.get("id", None)
review = review_util.get_review(trans.app, review_id)
components_dict = {}
for component in review_util.get_components(trans.app):
components_dict[component.name] = dict(component=component, component_review=None)
repository = review.repository
repo = repository.hg_repo
for component_review in review.component_reviews:
if component_review and component_review.component:
component_name = component_review.component.name
if component_name in components_dict:
component_review_dict = components_dict[component_name]
component_review_dict["component_review"] = component_review
components_dict[component_name] = component_review_dict
# Handle a Save button click.
save_button_clicked = False
save_buttons = [f"{comp_name}{STRSEP}review_button" for comp_name in components_dict.keys()]
save_buttons.append("revision_approved_button")
for save_button in save_buttons:
if save_button in kwd:
save_button_clicked = True
break
if save_button_clicked:
# Handle the revision_approved_select_field value.
revision_approved = kwd.get("revision_approved", None)
revision_approved_setting_changed = False
if revision_approved:
revision_approved = str(revision_approved)
if review.approved != revision_approved:
revision_approved_setting_changed = True
review.approved = revision_approved
trans.sa_session.add(review)
trans.sa_session.flush()
saved_component_names = []
for component_name in components_dict.keys():
flushed = False
# Retrieve the review information from the form.
# The star rating form field is a radio button list, so it will not be received if it was not clicked in the form.
# Due to this behavior, default the value to 0.
rating = 0
private = False
for k, v in kwd.items():
if k.startswith(f"{component_name}{STRSEP}"):
component_review_attr = k.replace(f"{component_name}{STRSEP}", "")
if component_review_attr == "component_id":
component_id = str(v)
elif component_review_attr == "comment":
comment = str(v)
elif component_review_attr == "private":
private = CheckboxField.is_checked(v)
elif component_review_attr == "approved":
approved = str(v)
elif component_review_attr == "rating":
rating = int(str(v))
component = review_util.get_component(trans.app, component_id)
component_review = review_util.get_component_review_by_repository_review_id_component_id(
trans.app, review_id, component_id
)
if component_review:
# See if the existing component review should be updated.
if (
component_review.comment != comment
or component_review.private != private
or component_review.approved != approved
or component_review.rating != rating
):
component_review.comment = comment
component_review.private = private
component_review.approved = approved
component_review.rating = rating
trans.sa_session.add(component_review)
trans.sa_session.flush()
flushed = True
saved_component_names.append(component_name)
else:
# See if a new component_review should be created.
if comment or private or approved != trans.model.ComponentReview.approved_states.NO or rating:
component_review = trans.model.ComponentReview(
repository_review_id=review.id,
component_id=component.id,
comment=comment,
approved=approved,
rating=rating,
)
trans.sa_session.add(component_review)
trans.sa_session.flush()
flushed = True
saved_component_names.append(component_name)
if flushed:
# Update the repository rating value to be the average of all component review ratings.
average_rating = (
trans.sa_session.query(func.avg(trans.model.ComponentReview.table.c.rating))
.filter(
and_(
trans.model.ComponentReview.table.c.repository_review_id == review.id,
trans.model.ComponentReview.table.c.deleted == false(),
trans.model.ComponentReview.table.c.approved
!= trans.model.ComponentReview.approved_states.NA,
)
)
.scalar()
)
if average_rating is not None:
review.rating = int(average_rating)
trans.sa_session.add(review)
trans.sa_session.flush()
# Update the information in components_dict.
if component_name in components_dict:
component_review_dict = components_dict[component_name]
component_review_dict["component_review"] = component_review
components_dict[component_name] = component_review_dict
if revision_approved_setting_changed:
message += f"Approved value <b>{review.approved}</b> saved for this revision.<br/>"
if saved_component_names:
message += f"Reviews were saved for components: {', '.join(saved_component_names)}"
if not revision_approved_setting_changed and not saved_component_names:
message += "No changes were made to this review, so nothing was saved."
if review and review.approved:
selected_value = review.approved
else:
selected_value = trans.model.ComponentReview.approved_states.NO
revision_approved_select_field = grids_util.build_approved_select_field(
trans, name="revision_approved", selected_value=selected_value, for_component=False
)
rev, changeset_revision_label = hg_util.get_rev_label_from_changeset_revision(repo, review.changeset_revision)
return trans.fill_template(
"/webapps/tool_shed/repository_review/edit_review.mako",
repository=repository,
review=review,
changeset_revision_label=changeset_revision_label,
revision_approved_select_field=revision_approved_select_field,
components_dict=components_dict,
message=message,
status=status,
)
[docs] @web.expose
@require_login("manage components")
def manage_components(self, trans, **kwd):
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "create":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="create_component", **kwd)
)
elif operation == "edit":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="edit_component", **kwd)
)
if "message" not in kwd:
message = "This is a list of repository components (features) that can be reviewed. You can add new components or change "
message += "the description of an existing component if appropriate. Click on the name link to change the description."
status = "warning"
kwd["message"] = message
kwd["status"] = status
return self.component_grid(trans, **kwd)
[docs] @web.expose
@require_login("manage repositories ready for review")
def manage_repositories_ready_for_review(self, trans, **kwd):
"""
A repository is ready to be reviewed if one of the following conditions is met:
1) It contains no tools
2) It contains tools the tools_functionally_correct flag is set to True. This implies that the repository metadata revision was installed and tested
by the Tool Shed's install and test framework.
"""
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "inspect repository revisions":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="create_review", **kwd)
)
if operation == "view_or_manage_repository":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
message = "Any of these repositories that contain tools have been installed into Galaxy and proven to be functionally correct by executing the tests defined "
message += "for each tool. Repositories that do not contain tools have not been installed into Galaxy. "
kwd["message"] = message
kwd["status"] = "warning"
return self.repositories_ready_for_review_grid(trans, **kwd)
[docs] @web.expose
@require_login("manage repositories reviewed by me")
def manage_repositories_reviewed_by_me(self, trans, **kwd):
# The value of the received id is the encoded repository id.
if "operation" in kwd:
kwd["mine"] = True
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="manage_repositories_with_reviews", **kwd)
)
self.repositories_reviewed_by_me_grid.title = "Repositories reviewed by me"
return self.repositories_reviewed_by_me_grid(trans, **kwd)
[docs] @web.expose
@require_login("manage repositories with invalid tests")
def manage_repositories_with_invalid_tests(self, trans, **kwd):
"""
Display a list of repositories that contain tools, have not yet been reviewed, and have invalid functional tests. Tests are defined as
invalid if they are missing from the tool config or if defined test data is not included in the repository.
"""
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "inspect repository revisions":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="create_review", **kwd)
)
if operation == "view_or_manage_repository":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
message = "These repositories contain tools with missing functional tests or test data. "
kwd["message"] = message
kwd["status"] = "warning"
return self.repositories_with_no_tool_tests_grid(trans, **kwd)
[docs] @web.expose
@require_login("manage repositories with reviews")
def manage_repositories_with_reviews(self, trans, **kwd):
# The value of the received id is the encoded repository id.
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "inspect repository revisions":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="manage_repository_reviews", **kwd)
)
if operation == "view_or_manage_repository":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
return self.repositories_with_reviews_grid(trans, **kwd)
[docs] @web.expose
@require_login("manage repositories without reviews")
def manage_repositories_without_reviews(self, trans, **kwd):
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "inspect repository revisions":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="create_review", **kwd)
)
if operation == "view_or_manage_repository":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
return self.repositories_without_reviews_grid(trans, **kwd)
[docs] @web.expose
@require_login("manage repository reviews")
def manage_repository_reviews(self, trans, mine=False, **kwd):
# The value of the received id is the encoded repository id.
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
repository_id = kwd.get("id", None)
if repository_id:
repository = suc.get_repository_in_tool_shed(trans.app, repository_id)
repo = repository.hg_repo
metadata_revision_hashes = [
metadata_revision.changeset_revision for metadata_revision in repository.metadata_revisions
]
reviewed_revision_hashes = [review.changeset_revision for review in repository.reviews]
reviews_dict = {}
for changeset in hg_util.get_reversed_changelog_changesets(repo):
changeset_revision = str(repo[changeset])
if changeset_revision in metadata_revision_hashes or changeset_revision in reviewed_revision_hashes:
rev, changeset_revision_label = hg_util.get_rev_label_from_changeset_revision(
repo, changeset_revision
)
if changeset_revision in reviewed_revision_hashes:
# Find the review for this changeset_revision
repository_reviews = review_util.get_reviews_by_repository_id_changeset_revision(
trans.app, repository_id, changeset_revision
)
# Determine if the current user can add a review to this revision.
can_add_review = trans.user not in [
repository_review.user for repository_review in repository_reviews
]
repository_metadata = metadata_util.get_repository_metadata_by_changeset_revision(
trans.app, repository_id, changeset_revision
)
if repository_metadata:
repository_metadata_reviews = util.listify(repository_metadata.reviews)
else:
repository_metadata_reviews = []
else:
repository_reviews = []
repository_metadata_reviews = []
can_add_review = True
installable = changeset_revision in metadata_revision_hashes
revision_dict = dict(
changeset_revision_label=changeset_revision_label,
repository_reviews=repository_reviews,
repository_metadata_reviews=repository_metadata_reviews,
installable=installable,
can_add_review=can_add_review,
)
reviews_dict[changeset_revision] = revision_dict
return trans.fill_template(
"/webapps/tool_shed/repository_review/reviews_of_repository.mako",
repository=repository,
reviews_dict=reviews_dict,
mine=mine,
message=message,
status=status,
)
[docs] @web.expose
@require_login("manage repository reviews of revision")
def manage_repository_reviews_of_revision(self, trans, **kwd):
# The value of the received id is the encoded repository id.
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
repository_id = kwd.get("id", None)
changeset_revision = kwd.get("changeset_revision", None)
repository = repository_util.get_repository_in_tool_shed(trans.app, repository_id)
repo = repository.hg_repo
installable = changeset_revision in [
metadata_revision.changeset_revision for metadata_revision in repository.metadata_revisions
]
rev, changeset_revision_label = hg_util.get_rev_label_from_changeset_revision(repo, changeset_revision)
reviews = review_util.get_reviews_by_repository_id_changeset_revision(
trans.app, repository_id, changeset_revision
)
return trans.fill_template(
"/webapps/tool_shed/repository_review/reviews_of_changeset_revision.mako",
repository=repository,
changeset_revision=changeset_revision,
changeset_revision_label=changeset_revision_label,
reviews=reviews,
installable=installable,
message=message,
status=status,
)
[docs] @web.expose
@require_login("repository reviews by user")
def repository_reviews_by_user(self, trans, **kwd):
if "operation" in kwd:
operation = kwd["operation"].lower()
# The value of the received id is the encoded review id.
review = review_util.get_review(trans.app, kwd["id"])
repository = review.repository
kwd["id"] = trans.security.encode_id(repository.id)
if operation == "inspect repository revisions":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="manage_repository_reviews", **kwd)
)
if operation == "view_or_manage_repository":
kwd["changeset_revision"] = review.changeset_revision
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
# The user may not be the current user. The value of the received id is the encoded user id.
user = suc.get_user(trans.app, kwd["id"])
self.repository_reviews_by_user_grid.title = f"All repository revision reviews for user '{user.username}'"
return self.repository_reviews_by_user_grid(trans, **kwd)
[docs] @web.expose
@require_login("reviewed repositories i own")
def reviewed_repositories_i_own(self, trans, **kwd):
# The value of the received id is the encoded repository id.
if "operation" in kwd:
operation = kwd["operation"].lower()
if operation == "inspect repository revisions":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="manage_repository_reviews", **kwd)
)
if operation == "view_or_manage_repository":
return trans.response.send_redirect(
web.url_for(controller="repository_review", action="view_or_manage_repository", **kwd)
)
return self.reviewed_repositories_i_own_grid(trans, **kwd)
[docs] @web.expose
@require_login("select previous review")
def select_previous_review(self, trans, **kwd):
# The value of the received id is the encoded repository id.
message = escape(kwd.get("message", ""))
status = kwd.get("status", "done")
repository = repository_util.get_repository_in_tool_shed(trans.app, kwd["id"])
changeset_revision = kwd.get("changeset_revision", None)
repo = repository.hg_repo
previous_reviews_dict = review_util.get_previous_repository_reviews(trans.app, repository, changeset_revision)
rev, changeset_revision_label = hg_util.get_rev_label_from_changeset_revision(repo, changeset_revision)
return trans.fill_template(
"/webapps/tool_shed/repository_review/select_previous_review.mako",
repository=repository,
changeset_revision=changeset_revision,
changeset_revision_label=changeset_revision_label,
previous_reviews_dict=previous_reviews_dict,
message=message,
status=status,
)
[docs] @web.expose
@require_login("view or manage repository")
def view_or_manage_repository(self, trans, **kwd):
repository = repository_util.get_repository_in_tool_shed(trans.app, kwd["id"])
if trans.user_is_admin or repository.user == trans.user:
return trans.response.send_redirect(web.url_for(controller="repository", action="manage_repository", **kwd))
else:
return trans.response.send_redirect(web.url_for(controller="repository", action="view_repository", **kwd))