Another round of easy tool shed cleanup.

lib/tool_shed/metadata/repository_metadata_manager.py

@@ -445,6 +445,7 @@ def create_or_update_repository_metadata(self, changeset_revision, metadata_dict
                 includes_tool_dependencies=includes_tool_dependencies,
                 includes_workflows=False,
             )
+        assert repository_metadata
         # Always set the default values for the following columns.  When resetting all metadata
         # on a repository this will reset the values.
         repository_metadata.missing_test_components = False
@@ -492,6 +493,7 @@ def get_parent_id(self, id, old_id, version, guid, changeset_revisions):
             repository_metadata = metadata_util.get_repository_metadata_by_changeset_revision(
                 self.app, id, changeset_revision
             )
+            assert repository_metadata
             metadata = repository_metadata.metadata
             tools_dicts = metadata.get("tools", [])
             for tool_dict in tools_dicts:
@@ -927,6 +929,7 @@ def reset_all_tool_versions(self, repo):
             repository_metadata = metadata_util.get_repository_metadata_by_changeset_revision(
                 self.app, encoded_repository_id, changeset_revision
             )
+            assert repository_metadata
             metadata = repository_metadata.metadata
             tool_dicts = metadata["tools"]
             if index == 0:

lib/tool_shed/structured_app.py

@@ -7,10 +7,12 @@
     from tool_shed.repository_types.registry import Registry as RepositoryTypesRegistry
     from tool_shed.util.hgweb_config import HgWebConfigManager
     from tool_shed.webapp.model import mapping
+    from tool_shed.webapp.security import CommunityRBACAgent
 
 
 class ToolShedApp(BasicSharedApp):
     repository_types_registry: "RepositoryTypesRegistry"
     model: "mapping.ToolShedModelMapping"
     repository_registry: "RepositoryRegistry"
     hgweb_config_manager: "HgWebConfigManager"
+    security_agent: "CommunityRBACAgent"

lib/tool_shed/test/base/populators.py

@@ -21,6 +21,7 @@
     GetOrderedInstallableRevisionsRequest,
     InstallInfo,
     OrderedInstallableRevisions,
+    RepositoriesByCategory,
     Repository,
     RepositoryIndexRequest,
     RepositoryIndexResponse,
@@ -152,6 +153,11 @@ def get_category_with_name(self, name: str) -> Optional[Category]:
         categories = [c for c in self.get_categories() if c.name == name]
         return categories[0] if categories else None
 
+    def repositories_by_category(self, category_id: str) -> RepositoriesByCategory:
+        response = self._api_interactor.get(f"categories/{category_id}/repositories")
+        response.raise_for_status()
+        return RepositoriesByCategory(**response.json())
+
     def has_category_with_name(self, name: str) -> bool:
         return self.get_category_with_name(name) is not None
 

lib/tool_shed/test/base/twilltestcase.py

@@ -564,9 +564,7 @@ def delete_repository(self, repository: Repository) -> None:
         strings_not_displayed: List[str] = []
         self.check_for_strings(strings_displayed, strings_not_displayed)
 
-    def display_installed_jobs_list_page(
-        self, installed_repository, data_manager_names=None, strings_displayed=None, strings_not_displayed=None
-    ):
+    def display_installed_jobs_list_page(self, installed_repository, data_manager_names=None, strings_displayed=None):
         data_managers = installed_repository.metadata_.get("data_manager", {}).get("data_managers", {})
         if data_manager_names:
             if not isinstance(data_manager_names, list):
@@ -580,7 +578,7 @@ def display_installed_jobs_list_page(
         for data_manager_name in data_manager_names:
             params = {"id": data_managers[data_manager_name]["guid"]}
             self.visit_galaxy_url("/data_manager/jobs_list", params=params)
-            self.check_for_strings(strings_displayed, strings_not_displayed)
+            self.check_for_strings(strings_displayed)
 
     def display_installed_repository_manage_json(self, installed_repository):
         params = {"id": self.security.encode_id(installed_repository.id)}

lib/tool_shed/test/functional/test_shed_repositories.py

@@ -10,15 +10,13 @@ def test_create(self):
         populator = self.populator
         category_id = populator.new_category(prefix="testcreate").id
 
-        response = self.api_interactor.get(f"categories/{category_id}/repositories")
-        api_asserts.assert_status_code_is_ok(response)
-        repos = response.json()["repositories"]
+        repos_by_category = populator.repositories_by_category(category_id)
+        repos = repos_by_category.repositories
         assert len(repos) == 0
 
         populator.new_repository(category_id)
-        response = self.api_interactor.get(f"categories/{category_id}/repositories")
-        api_asserts.assert_status_code_is_ok(response)
-        repos = response.json()["repositories"]
+        repos_by_category = populator.repositories_by_category(category_id)
+        repos = repos_by_category.repositories
         assert len(repos) == 1
 
     def test_update_repository(self):

lib/tool_shed/util/metadata_util.py

@@ -1,5 +1,9 @@
 import logging
 from operator import itemgetter
+from typing import (
+    Optional,
+    TYPE_CHECKING,
+)
 
 from sqlalchemy import and_
 
@@ -11,6 +15,12 @@
 from galaxy.util.tool_shed.common_util import parse_repository_dependency_tuple
 from tool_shed.util.hg_util import changeset2rev
 
+if TYPE_CHECKING:
+    from tool_shed.structured_app import ToolShedApp
+    from tool_shed.webapp.model import RepositoryMetadata
+    from tool_shed.webapp.model.mapping import ToolShedModelMapping
+
+
 log = logging.getLogger(__name__)
 
 
@@ -229,18 +239,27 @@ def get_repository_dependency_tups_from_repository_metadata(app, repository_meta
     return dependency_tups
 
 
-def get_repository_metadata_by_changeset_revision(app, id, changeset_revision):
+def get_repository_metadata_by_changeset_revision(
+    app: "ToolShedApp", id: str, changeset_revision: str
+) -> Optional["RepositoryMetadata"]:
     """Get metadata for a specified repository change set from the database."""
+    decoded_id = app.security.decode_id(id)
+    return repository_metadata_by_changeset_revision(app.model, decoded_id, changeset_revision)
+
+
+def repository_metadata_by_changeset_revision(
+    model_mapping: "ToolShedModelMapping", id: int, changeset_revision: str
+) -> Optional["RepositoryMetadata"]:
     # Make sure there are no duplicate records, and return the single unique record for the changeset_revision.
     # Duplicate records were somehow created in the past.  The cause of this issue has been resolved, but we'll
     # leave this method as is for a while longer to ensure all duplicate records are removed.
-    sa_session = app.model.session
+    sa_session = model_mapping.context
     all_metadata_records = (
-        sa_session.query(app.model.RepositoryMetadata)
+        sa_session.query(model_mapping.RepositoryMetadata)
         .filter(
             and_(
-                app.model.RepositoryMetadata.table.c.repository_id == app.security.decode_id(id),
-                app.model.RepositoryMetadata.table.c.changeset_revision == changeset_revision,
+                model_mapping.RepositoryMetadata.table.c.repository_id == id,
+                model_mapping.RepositoryMetadata.table.c.changeset_revision == changeset_revision,
             )
         )
         .all()

lib/tool_shed/util/repository_util.py

@@ -232,7 +232,9 @@ def create_repository(
     return repository, message
 
 
-def generate_sharable_link_for_repository_in_tool_shed(repository, changeset_revision=None):
+def generate_sharable_link_for_repository_in_tool_shed(
+    repository: "Repository", changeset_revision: Optional[str] = None
+) -> str:
     """Generate the URL for sharing a repository that is in the tool shed."""
     base_url = web.url_for("/", qualified=True).rstrip("/")
     sharable_url = f"{base_url}/view/{repository.user.username}/{repository.name}"
@@ -361,6 +363,7 @@ def get_repositories_by_category(
         for changeset, changehash in repository.installable_revisions(app):
             encoded_id = app.security.encode_id(repository.id)
             metadata = get_repository_metadata_by_changeset_revision(app, encoded_id, changehash)
+            assert metadata
             repository_dict["metadata"][f"{changeset}:{changehash}"] = metadata.to_dict(
                 value_mapper=default_value_mapper
             )
@@ -436,7 +439,7 @@ def handle_role_associations(app: "ToolShedApp", role, repository, **kwd):
     return associations_dict
 
 
-def change_repository_name_in_hgrc_file(hgrc_file, new_name):
+def change_repository_name_in_hgrc_file(hgrc_file: str, new_name: str) -> None:
     config = configparser.ConfigParser()
     config.read(hgrc_file)
     config.set("web", "name", new_name)

lib/tool_shed/util/search_util.py

@@ -63,13 +63,6 @@ def in_tool_dict(tool_dict, exact_matches_checked, tool_id=None, tool_name=None,
     return found
 
 
-def in_workflow_dict(workflow_dict, exact_matches_checked, workflow_name):
-    workflow_dict_workflow_name = workflow_dict["name"].lower()
-    return (workflow_name == workflow_dict_workflow_name) or (
-        not exact_matches_checked and workflow_dict_workflow_name.find(workflow_name) >= 0
-    )
-
-
 def make_same_length(list1, list2):
     # If either list is 1 item, we'll append to it until its length is the same as the other.
     if len(list1) == 1:

lib/tool_shed_client/schema/__init__.py

@@ -218,6 +218,11 @@ class RepositoryIndexRequest(BaseModel):
     deleted: str = "false"
 
 
+class RepositoriesByCategory(Category):
+    repository_count: int
+    repositories: List[Repository]
+
+
 class RepositoryIndexResponse(BaseModel):
     __root__: List[Repository]