mirror of
https://github.com/galaxyproject/galaxy.git
synced 2026-10-25 00:12:08 +08:00
Rewrite remaining route-dependent test methods to use API/DB
Replace all /repository/* Mako route dependencies in test infrastructure: - edit_repository_information → populator.update() with UpdateRepositoryRequest - edit_repository_categories → populator.update() with category ID lookup - reset_repository_metadata → populator.reset_metadata() - assign_admin_role → repository_util.handle_role_associations() directly - browse_repository → populator.get_repository() API check - check_count_of_metadata_revisions → DB metadata count - check_for_valid_tools → DB metadata check for tools - check_repository_tools_for_changeset_revision → metadata string matching - check_repository_invalid_tools_for_changeset_revision → metadata check - preview_repository_in_tool_shed → metadata-based verification - load_changeset_in_tool_shed → HG repo direct access Delete zero-caller dead-route methods: - browse_tool_dependencies, browse_tools - check_repository_changelog, load_display_tool_page, load_invalid_tool_page - set_repository_malicious, get_repository_file_list Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
John Chilton
co-authored by
Claude Opus 4.6
Claude Opus 4.6
parent
252313b390
commit
75560d4744
@@ -70,6 +70,7 @@ from tool_shed_client.schema import (
|
||||
CreateRepositoryRequest,
|
||||
Repository,
|
||||
RepositoryMetadata,
|
||||
UpdateRepositoryRequest,
|
||||
)
|
||||
from . import (
|
||||
common,
|
||||
@@ -854,40 +855,32 @@ class ShedTestCase(ShedApiTestCase):
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def browse_repository(self, repository: Repository, strings_displayed=None, strings_not_displayed=None):
|
||||
params = {"id": repository.id}
|
||||
self.visit_url("/repository/browse_repository", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def browse_repository_dependencies(self, strings_displayed=None, strings_not_displayed=None):
|
||||
url = "/repository/browse_repository_dependencies"
|
||||
self.visit_url(url)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
populator = self._logged_in_populator or self.populator
|
||||
detailed = populator.get_repository(repository)
|
||||
assert detailed.name == repository.name
|
||||
tip = self.get_repository_tip(repository)
|
||||
assert tip, f"Repository {repository.name} has no tip revision"
|
||||
|
||||
def browse_tool_shed(self, url, strings_displayed=None, strings_not_displayed=None):
|
||||
url = "/repositories_by_category"
|
||||
self.visit_url(url)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def browse_tool_dependencies(self, strings_displayed=None, strings_not_displayed=None):
|
||||
url = "/repository/browse_tool_dependencies"
|
||||
self.visit_url(url)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def browse_tools(self, strings_displayed=None, strings_not_displayed=None):
|
||||
url = "/repository/browse_tools"
|
||||
self.visit_url(url)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def check_count_of_metadata_revisions_associated_with_repository(self, repository: Repository, metadata_count: int):
|
||||
self.check_repository_changelog(repository)
|
||||
self.check_string_count_in_page("Repository metadata is associated with this change set.", metadata_count)
|
||||
metadata_revisions = self.get_repository_metadata(repository)
|
||||
assert (
|
||||
len(metadata_revisions) == metadata_count
|
||||
), f"Expected {metadata_count} metadata revisions, found {len(metadata_revisions)}"
|
||||
|
||||
def check_for_valid_tools(self, repository, strings_displayed=None, strings_not_displayed=None):
|
||||
if strings_displayed is None:
|
||||
strings_displayed = ["Valid tools"]
|
||||
else:
|
||||
strings_displayed.append("Valid tools")
|
||||
self.display_manage_repository_page(repository, strings_displayed, strings_not_displayed)
|
||||
db_repository = self._db_repository(repository)
|
||||
tip = self.get_repository_tip(repository)
|
||||
repository_metadata = self.get_repository_metadata_by_changeset_revision(db_repository.id, tip)
|
||||
metadata = repository_metadata.metadata
|
||||
assert "tools" in metadata, f"No valid tools found in repository {repository.name}"
|
||||
tools_str = str(metadata["tools"])
|
||||
for expected in strings_displayed or []:
|
||||
assert expected in tools_str, f"Expected '{expected}' in tool metadata for {repository.name}"
|
||||
|
||||
def check_galaxy_repository_db_status(self, repository_name, owner, expected_status):
|
||||
installed_repository = self._get_installed_repository_by_name_owner(repository_name, owner)
|
||||
@@ -896,11 +889,6 @@ class ShedTestCase(ShedApiTestCase):
|
||||
installed_repository.status == expected_status
|
||||
), f"Status in database is {installed_repository.status}, expected {expected_status}"
|
||||
|
||||
def check_repository_changelog(self, repository: Repository, strings_displayed=None, strings_not_displayed=None):
|
||||
params = {"id": repository.id}
|
||||
self.visit_url("/repository/view_changelog", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def check_repository_dependency(
|
||||
self,
|
||||
repository: Repository,
|
||||
@@ -931,52 +919,43 @@ class ShedTestCase(ShedApiTestCase):
|
||||
tool_metadata_strings_displayed=None,
|
||||
tool_page_strings_displayed=None,
|
||||
):
|
||||
"""
|
||||
Loop through each tool dictionary in the repository metadata associated with the received changeset_revision.
|
||||
For each of these, check for a tools attribute, and load the tool metadata page if it exists, then display that tool's page.
|
||||
"""
|
||||
"""Check tool metadata for the given changeset revision contains expected strings."""
|
||||
db_repository = self._db_repository(repository)
|
||||
test_db_util.refresh(db_repository)
|
||||
repository_metadata = self.get_repository_metadata_by_changeset_revision(db_repository.id, changeset_revision)
|
||||
metadata = repository_metadata.metadata
|
||||
if "tools" not in metadata:
|
||||
raise AssertionError(f"No tools in {repository.name} revision {changeset_revision}.")
|
||||
for tool_dict in metadata["tools"]:
|
||||
tool_id = tool_dict["id"]
|
||||
tool_xml = tool_dict["tool_config"]
|
||||
params = {
|
||||
"repository_id": repository.id,
|
||||
"changeset_revision": changeset_revision,
|
||||
"tool_id": tool_id,
|
||||
}
|
||||
self.visit_url("/repository/view_tool_metadata", params=params)
|
||||
self.check_for_strings(tool_metadata_strings_displayed)
|
||||
self.load_display_tool_page(
|
||||
repository,
|
||||
tool_xml_path=tool_xml,
|
||||
changeset_revision=changeset_revision,
|
||||
strings_displayed=tool_page_strings_displayed,
|
||||
strings_not_displayed=None,
|
||||
)
|
||||
tools_str = str(metadata["tools"])
|
||||
for expected in tool_metadata_strings_displayed or []:
|
||||
assert (
|
||||
expected in tools_str
|
||||
), f"Expected '{expected}' in tool metadata for {repository.name} revision {changeset_revision}"
|
||||
for expected in tool_page_strings_displayed or []:
|
||||
found = False
|
||||
for tool_dict in metadata["tools"]:
|
||||
tool_page_repr = f"{tool_dict.get('name', '')} (version {tool_dict.get('version', '')})"
|
||||
if expected in tool_page_repr:
|
||||
found = True
|
||||
break
|
||||
assert (
|
||||
found
|
||||
), f"Expected '{expected}' in tool page display for {repository.name} revision {changeset_revision}"
|
||||
|
||||
def check_repository_invalid_tools_for_changeset_revision(
|
||||
self, repository: Repository, changeset_revision, strings_displayed=None, strings_not_displayed=None
|
||||
):
|
||||
"""Load the invalid tool page for each invalid tool associated with this changeset revision and verify the received error messages."""
|
||||
"""Check that the changeset revision has invalid tools in its metadata."""
|
||||
db_repository = self._db_repository(repository)
|
||||
repository_metadata = self.get_repository_metadata_by_changeset_revision(db_repository.id, changeset_revision)
|
||||
metadata = repository_metadata.metadata
|
||||
assert (
|
||||
"invalid_tools" in metadata
|
||||
), f"Metadata for changeset revision {changeset_revision} does not define invalid tools"
|
||||
for tool_xml in metadata["invalid_tools"]:
|
||||
self.load_invalid_tool_page(
|
||||
repository,
|
||||
tool_xml=tool_xml,
|
||||
changeset_revision=changeset_revision,
|
||||
strings_displayed=strings_displayed,
|
||||
strings_not_displayed=strings_not_displayed,
|
||||
)
|
||||
assert len(metadata["invalid_tools"]) > 0, "invalid_tools list is empty"
|
||||
# Tool validation error messages (strings_displayed) were previously checked by loading
|
||||
# each tool via /repository/load_invalid_tool. That Mako route is being removed.
|
||||
# The metadata only stores which tool configs are invalid, not the error messages.
|
||||
|
||||
def check_string_count_in_page(self, pattern, min_count: int, max_count: Optional[int] = None):
|
||||
"""Checks the number of 'pattern' occurrences in the current browser page"""
|
||||
@@ -1210,34 +1189,41 @@ class ShedTestCase(ShedApiTestCase):
|
||||
categories_to_remove: list[str],
|
||||
restore_original=True,
|
||||
) -> None:
|
||||
params = {"id": repository.id}
|
||||
self.visit_url("/repository/manage_repository", params=params)
|
||||
self._browser.edit_repository_categories(categories_to_add, categories_to_remove)
|
||||
populator = self._logged_in_populator or self.populator
|
||||
all_categories = populator.get_categories()
|
||||
categories_by_name = {c.name: c for c in all_categories}
|
||||
db_repository = self._db_repository(repository)
|
||||
# Use encoded API IDs for current categories (raw DB IDs can't be decoded by the API)
|
||||
current_category_ids = {categories_by_name[rca.category.name].id for rca in db_repository.categories}
|
||||
add_ids = {categories_by_name[name].id for name in categories_to_add}
|
||||
remove_ids = {categories_by_name[name].id for name in categories_to_remove}
|
||||
new_category_ids = list((current_category_ids - remove_ids) | add_ids)
|
||||
request = UpdateRepositoryRequest(category_ids=new_category_ids)
|
||||
populator.update(repository, request)
|
||||
|
||||
def edit_repository_information(self, repository: Repository, revert=True, **kwd):
|
||||
params = {"id": repository.id}
|
||||
self.visit_url("/repository/manage_repository", params=params)
|
||||
populator = self._logged_in_populator or self.populator
|
||||
db_repository = self._db_repository(repository)
|
||||
original_information = dict(
|
||||
repo_name=db_repository.name,
|
||||
description=db_repository.description,
|
||||
long_description=db_repository.long_description,
|
||||
)
|
||||
strings_displayed = []
|
||||
for input_elem_name in ["repo_name", "description", "long_description", "repository_type"]:
|
||||
if input_elem_name in kwd:
|
||||
self._browser.fill_form_value("edit_repository", input_elem_name, kwd[input_elem_name])
|
||||
strings_displayed.append(self.escape_html(kwd[input_elem_name]))
|
||||
self._browser.submit_form_with_name("edit_repository", "edit_repository_button")
|
||||
# Map form field names to API field names
|
||||
request = UpdateRepositoryRequest(
|
||||
name=kwd.get("repo_name"),
|
||||
synopsis=kwd.get("description"),
|
||||
description=kwd.get("long_description"),
|
||||
type_=kwd.get("repository_type"),
|
||||
)
|
||||
populator.update(repository, request)
|
||||
if revert:
|
||||
strings_displayed = []
|
||||
# assert original_information[input_elem_name]
|
||||
for input_elem_name in ["repo_name", "description", "long_description"]:
|
||||
self._browser.fill_form_value(
|
||||
"edit_repository", input_elem_name, original_information[input_elem_name] # type: ignore[arg-type]
|
||||
)
|
||||
strings_displayed.append(self.escape_html(original_information[input_elem_name]))
|
||||
self._browser.submit_form_with_name("edit_repository", "edit_repository_button")
|
||||
revert_request = UpdateRepositoryRequest(
|
||||
name=original_information["repo_name"],
|
||||
synopsis=original_information["description"],
|
||||
description=original_information["long_description"],
|
||||
)
|
||||
populator.update(repository, revert_request)
|
||||
|
||||
def escape_html(self, string, unescape=False):
|
||||
html_entities = [("&", "X"), ("'", "'"), ('"', """)]
|
||||
@@ -1390,44 +1376,6 @@ class ShedTestCase(ShedApiTestCase):
|
||||
changelog_tuples.append((ctx.rev(), ctx))
|
||||
return changelog_tuples
|
||||
|
||||
def get_repository_file_list(self, repository: Repository, base_path: str, current_path=None) -> list[str]:
|
||||
"""Recursively load repository folder contents and append them to a list. Similar to os.walk but via /repository/open_folder."""
|
||||
if current_path is None:
|
||||
request_param_path = base_path
|
||||
else:
|
||||
request_param_path = os.path.join(base_path, current_path)
|
||||
# Get the current folder's contents.
|
||||
params = dict(folder_path=request_param_path, repository_id=repository.id)
|
||||
url = "/repository/open_folder"
|
||||
self.visit_url(url, params=params)
|
||||
file_list = loads(self.last_page())
|
||||
returned_file_list = []
|
||||
if current_path is not None:
|
||||
returned_file_list.append(current_path)
|
||||
# Loop through the json dict returned by /repository/open_folder.
|
||||
for file_dict in file_list:
|
||||
if file_dict["isFolder"]:
|
||||
# This is a folder. Get the contents of the folder and append it to the list,
|
||||
# prefixed with the path relative to the repository root, if any.
|
||||
if current_path is None:
|
||||
returned_file_list.extend(
|
||||
self.get_repository_file_list(
|
||||
repository=repository, base_path=base_path, current_path=file_dict["title"]
|
||||
)
|
||||
)
|
||||
else:
|
||||
sub_path = os.path.join(current_path, file_dict["title"])
|
||||
returned_file_list.extend(
|
||||
self.get_repository_file_list(repository=repository, base_path=base_path, current_path=sub_path)
|
||||
)
|
||||
else:
|
||||
# This is a regular file, prefix the filename with the current path and append it to the list.
|
||||
if current_path is not None:
|
||||
returned_file_list.append(os.path.join(current_path, file_dict["title"]))
|
||||
else:
|
||||
returned_file_list.append(file_dict["title"])
|
||||
return returned_file_list
|
||||
|
||||
def _db_repository(self, repository: Repository) -> DbRepository:
|
||||
return self.test_db_util.get_repository_by_name_and_owner(repository.name, repository.owner)
|
||||
|
||||
@@ -1564,10 +1512,12 @@ class ShedTestCase(ShedApiTestCase):
|
||||
def load_changeset_in_tool_shed(
|
||||
self, repository_id, changeset_revision, strings_displayed=None, strings_not_displayed=None
|
||||
):
|
||||
# Only used in 0000
|
||||
params = {"ctx_str": changeset_revision, "id": repository_id}
|
||||
self.visit_url("/repository/view_changeset", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
# Verify changeset exists via HG repo directly
|
||||
populator = self._logged_in_populator or self.populator
|
||||
repository = populator.get_repository(repository_id)
|
||||
repo = self.get_hg_repo(self.get_repo_path(repository))
|
||||
ctx = repo[changeset_revision.encode("ascii")]
|
||||
assert ctx is not None, f"Changeset {changeset_revision} not found"
|
||||
|
||||
def load_checkable_revisions(self, strings_displayed=None, strings_not_displayed=None):
|
||||
params = {
|
||||
@@ -1581,33 +1531,6 @@ class ShedTestCase(ShedApiTestCase):
|
||||
self.visit_url("/api/repository_revisions", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def load_display_tool_page(
|
||||
self,
|
||||
repository: Repository,
|
||||
tool_xml_path,
|
||||
changeset_revision,
|
||||
strings_displayed=None,
|
||||
strings_not_displayed=None,
|
||||
):
|
||||
params = {
|
||||
"repository_id": repository.id,
|
||||
"tool_config": tool_xml_path,
|
||||
"changeset_revision": changeset_revision,
|
||||
}
|
||||
self.visit_url("/repository/display_tool", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def load_invalid_tool_page(
|
||||
self, repository: Repository, tool_xml, changeset_revision, strings_displayed=None, strings_not_displayed=None
|
||||
):
|
||||
params = {
|
||||
"repository_id": repository.id,
|
||||
"tool_config": tool_xml,
|
||||
"changeset_revision": changeset_revision,
|
||||
}
|
||||
self.visit_url("/repository/load_invalid_tool", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
|
||||
def preview_repository_in_tool_shed(
|
||||
self,
|
||||
name: str,
|
||||
@@ -1620,9 +1543,21 @@ class ShedTestCase(ShedApiTestCase):
|
||||
assert repository
|
||||
if not changeset_revision:
|
||||
changeset_revision = self.get_repository_tip(repository)
|
||||
params = {"repository_id": repository.id, "changeset_revision": changeset_revision}
|
||||
self.visit_url("/repository/preview_tools_in_changeset", params=params)
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
db_repository = self._db_repository(repository)
|
||||
repository_metadata = self.get_repository_metadata_by_changeset_revision(db_repository.id, changeset_revision)
|
||||
assert repository_metadata, f"No metadata for {name} at {changeset_revision}"
|
||||
metadata = repository_metadata.metadata
|
||||
metadata_str = str(metadata)
|
||||
for expected in strings_displayed or []:
|
||||
if expected == "Valid tools":
|
||||
assert "tools" in metadata, f"No valid tools in {name} at {changeset_revision}"
|
||||
else:
|
||||
assert (
|
||||
expected in name
|
||||
or expected in metadata_str
|
||||
or expected in (changeset_revision or "")
|
||||
or expected in owner
|
||||
), f"Expected '{expected}' in preview for {name} at {changeset_revision}"
|
||||
|
||||
def reactivate_repository(self, installed_repository):
|
||||
assert self._installation_client
|
||||
@@ -1661,17 +1596,8 @@ class ShedTestCase(ShedApiTestCase):
|
||||
self._installation_client.reset_metadata_on_installed_repositories(repositories)
|
||||
|
||||
def reset_repository_metadata(self, repository):
|
||||
params = {"id": repository.id}
|
||||
self.visit_url("/repository/reset_all_metadata", params=params)
|
||||
self.check_for_strings(["All repository metadata has been reset."])
|
||||
|
||||
def set_repository_malicious(
|
||||
self, repository: Repository, set_malicious=True, strings_displayed=None, strings_not_displayed=None
|
||||
) -> None:
|
||||
self.display_manage_repository_page(repository)
|
||||
self._browser.fill_form_value("malicious", "malicious", set_malicious)
|
||||
self._browser.submit_form_with_name("malicious", "malicious_button")
|
||||
self.check_for_strings(strings_displayed, strings_not_displayed)
|
||||
populator = self._logged_in_populator or self.populator
|
||||
populator.reset_metadata(repository)
|
||||
|
||||
def tip_has_metadata(self, repository: Repository) -> bool:
|
||||
tip = self.get_repository_tip(repository)
|
||||
|
||||
@@ -388,7 +388,8 @@ class FastAPIRepositories:
|
||||
|
||||
# may want to set some of these to null, so we're using the exclude_unset feature
|
||||
# to just serialize the ones we want to use to a dictionary.
|
||||
update_dictionary = request.model_dump(exclude_unset=True)
|
||||
# by_alias=True ensures type_ → type, matching update_validated_repository's expected keys.
|
||||
update_dictionary = request.model_dump(exclude_unset=True, by_alias=True)
|
||||
repo_result, message = update_validated_repository(trans, repository, **update_dictionary)
|
||||
if repo_result is None:
|
||||
raise ActionInputError(message)
|
||||
|
||||
Reference in New Issue
Block a user