diff --git a/lib/tool_shed/test/base/api_util.py b/lib/tool_shed/test/base/api_util.py index 81dad57cb18..e90a495aa69 100644 --- a/lib/tool_shed/test/base/api_util.py +++ b/lib/tool_shed/test/base/api_util.py @@ -99,6 +99,10 @@ class ShedApiInteractor: response.raise_for_status() return response.json() + @property + def hg_url_base(self): + return self.url + def create_user(admin_interactor: ShedApiInteractor, user_dict: Dict[str, Any], assert_ok=True) -> Dict[str, Any]: email = user_dict["email"] diff --git a/lib/tool_shed/test/base/populators.py b/lib/tool_shed/test/base/populators.py index dc2cddf1bdc..7e2f93c3316 100644 --- a/lib/tool_shed/test/base/populators.py +++ b/lib/tool_shed/test/base/populators.py @@ -132,16 +132,24 @@ class ToolShedPopulator: assert_ok=True, start: int = 0, end: Optional[int] = None, + category_id: Optional[str] = None, ) -> Repository: if repository is None: prefix = test_data_path.replace("_", "") - category_id = self.new_category(prefix=prefix).id + if category_id is None: + category_id = self.new_category(prefix=prefix).id repository = self.new_repository(category_id, prefix=prefix) self.setup_test_data_repo_by_id(test_data_path, repository, assert_ok=assert_ok, start=start, end=end) return repository - def setup_column_maker_repo(self, prefix=DEFAULT_PREFIX) -> Repository: - category_id = self.new_category(prefix=prefix).id + def setup_column_maker_repo( + self, + prefix=DEFAULT_PREFIX, + category_id: Optional[str] = None, + ) -> Repository: + if category_id is None: + category_id = self.new_category(prefix=prefix).id + assert category_id repository = self.new_repository(category_id, prefix=prefix) repository_id = repository.id assert repository_id @@ -410,6 +418,9 @@ class ToolShedPopulator: base = url.split("://")[1].split("/")[0] return f"{base}/repos/{repository.owner}/{repository.name}/{tool_id}/{tool_version}" + def new_user(self, username: str, password: str): + return ensure_user_with_email(self._admin_api_interactor, username, password) + def _repository_id(self, has_id: HasRepositoryId) -> str: if isinstance(has_id, Repository): return has_id.id diff --git a/scripts/bootstrap_test_shed.py b/scripts/bootstrap_test_shed.py new file mode 100644 index 00000000000..40f257ded9e --- /dev/null +++ b/scripts/bootstrap_test_shed.py @@ -0,0 +1,249 @@ +"""Script to bootstrap a tool shed server for development. + +- Create categories. +- Create some users. +- Create some repositories +""" + +import argparse +import os +import subprocess +import sys +import tempfile +from typing import ( + Any, + Dict, + List, + Optional, +) + +import requests + +sys.path.insert(1, os.path.abspath(os.path.join(os.path.dirname(__file__), os.pardir, "lib"))) + + +from galaxy.tool_shed.util.hg_util import clone_repository +from tool_shed.test.base.api import ensure_user_with_email +from tool_shed.test.base.api_util import ( + create_user, + ShedApiInteractor, +) +from tool_shed.test.base.populators import ToolShedPopulator +from tool_shed_client.schema import ( + Category, + CreateRepositoryRequest, +) + +DESCRIPTION = "Script to bootstrap a tool shed server for development" +DEFAULT_USER = "jmchilton@gmail.com" +DEFAULT_USER_PASSWORD = "password123" # it is safe because of the 123 + +TEST_CATEGORY_NAME = "Testing Category" +TEST_CATEGORY_DESCRIPTION = "A longer description of the testing category" + +MAIN_SHED_URL = "https://toolshed.g2.bx.psu.edu/" +MAIN_SHED_API = f"{MAIN_SHED_URL}/api" + +CATEGORIES_TO_COPY = ["Data Export", "Climate Analysis", "Materials science"] + + +def main(argv: List[str]) -> None: + arg_parser = _arg_parser() + namespace = arg_parser.parse_args(argv) + populator = init_populator(namespace) + + category = populator.new_category_if_needed( + {"name": "Invalid Test Tools", "description": "A contains a repository with invalid tools."} + ) + populator.setup_bismark_repo(category_id=category.id) + + category = populator.new_category_if_needed({"name": "Test Category", "description": "A longer test description."}) + mirror_main_categories(populator) + mirror_main_users(populator) + + populator.new_user_if_needed({"email": "bob@bobsdomain.com"}) + populator.new_user_if_needed({"email": "alice@alicesdomain.com"}) + populator.new_user_if_needed({"email": "thirduser@threeis.com"}) + + populator.setup_column_maker_repo(prefix="bootstrap", category_id=category.id) + populator.setup_column_maker_repo(prefix="bootstrap2", category_id=category.id) + + main_categories = get_main_categories() + for category in main_categories: + category_id = category["id"] + category_name = category["name"] + if category_name in CATEGORIES_TO_COPY: + local_category = populator.get_category_with_name(category_name) + repos = get_main_repositories_for_category(category_id) + for repo in repos: + mirror_main_repository(populator, repo, local_category.id) + + +def get_main_categories() -> List[Dict[str, Any]]: + main_categories_endpoint = f"{MAIN_SHED_API}/categories" + main_categories = requests.get(main_categories_endpoint).json() + return main_categories + + +def get_main_users() -> List[Dict[str, Any]]: + main_users_endpoint = f"{MAIN_SHED_API}/users" + main_users = requests.get(main_users_endpoint).json() + return main_users + + +def get_main_repositories_for_category(category_id) -> List[Dict[str, Any]]: + main_category_repos_endpoint = f"{MAIN_SHED_API}/categories/{category_id}/repositories" + main_repos_for_category_response = requests.get(main_category_repos_endpoint) + main_repos_for_category = main_repos_for_category_response.json() + assert "repositories" in main_repos_for_category + return main_repos_for_category["repositories"] + + +class RemoteToolShedPopulator(ToolShedPopulator): + """Extend the tool shed populator with some state tracking... + + ... tailored toward bootstrapping dev instances instead of + for tests. + """ + + _categories_by_name: Optional[Dict[str, Category]] = None + _users_by_username: Optional[Dict[str, Dict[str, Any]]] = None + _populators_by_username: Dict[str, "RemoteToolShedPopulator"] = {} + + def __init__(self, admin_interactor: ShedApiInteractor, user_interactor: ShedApiInteractor): + super().__init__(admin_interactor, user_interactor) + + def populator_for_user(self, username): + if username not in self._populators_by_username: + user = self.users_by_username[username] + assert user + mock_email = f"{username}@galaxyproject.org" + password = "testpass" + api_key = self._admin_api_interactor.create_api_key(mock_email, password) + user_interactor = ShedApiInteractor(self._admin_api_interactor.url, api_key) + self._populators_by_username[username] = RemoteToolShedPopulator( + self._admin_api_interactor, user_interactor + ) + return self._populators_by_username[username] + + @property + def categories_by_name(self) -> Dict[str, Category]: + if self._categories_by_name is None: + categories = self.get_categories() + self._categories_by_name = {c.name: c for c in categories} + return self._categories_by_name + + @property + def users_by_username(self) -> Dict[str, Dict[str, Any]]: + if self._users_by_username is None: + users_response = self._api_interactor.get("users") + if users_response.status_code == 400: + error_response = users_response.json() + raise Exception(str(error_response)) + users_response.raise_for_status() + users = users_response.json() + self._users_by_username = {u["username"]: u for u in users} + return self._users_by_username + + def new_category_if_needed(self, as_json: Dict[str, Any]) -> Category: + name = as_json["name"] + description = as_json["description"] + if name in self.categories_by_name: + return self.categories_by_name[name] + return self.new_category(name, description) + + def new_user_if_needed(self, as_json) -> Dict[str, Any]: + if "username" not in as_json: + email = as_json["email"] + as_json["username"] = email.split("@", 1)[0] + username = as_json["username"] + if username in self.users_by_username: + return self.users_by_username[username] + if "email" not in as_json: + mock_email = f"{username}@galaxyproject.org" + as_json["email"] = mock_email + request = {"username": as_json["username"], "email": as_json["email"]} + user = create_user(self._admin_api_interactor, request) + self.users_by_username[username] = user + return user + + +def mirror_main_categories(populator: RemoteToolShedPopulator): + main_categories = get_main_categories() + for category in main_categories: + populator.new_category_if_needed(category) + + +def mirror_main_users(populator: RemoteToolShedPopulator): + main_users = get_main_users() + for user in main_users: + populator.new_user_if_needed(user) + + +def mirror_main_repository(populator: RemoteToolShedPopulator, repository: Dict[str, Any], category_id: str): + # TODO: mirror the user + as_dict = repository.copy() + as_dict["category_ids"] = category_id + as_dict["synopsis"] = repository["description"] + request = CreateRepositoryRequest(**as_dict) + username = repository["owner"] + user_populator = populator.populator_for_user(username) + new_repository = None + try: + new_repository = user_populator.create_repository(request) + except AssertionError as e: + # if the problem is just a repository already + # exists, continue + err_msg = str(e) + if "already own" not in err_msg: + raise + if new_repository: + name = repository["name"] + clone_url = f"{MAIN_SHED_URL}/repos/{username}/{name}" + temp_dir = tempfile.mkdtemp() + clone_repository(clone_url, temp_dir) + url_base = populator._api_interactor.hg_url_base + prefix, rest = url_base.split("://", 1) + target = f"{prefix}://{username}@{rest}/repos/{username}/{name}" + try: + _push_to(temp_dir, target) + except Exception as e: + print(f"Problem cloning repository {e}, continuing bootstrap though...") + pass + populator.reset_metadata(new_repository) + + +def _push_to(repo_path: str, repo_target: str) -> None: + subprocess.check_output(["hg", "push", repo_target], cwd=repo_path) + + +def init_populator(namespace) -> RemoteToolShedPopulator: + admin_interactor = ShedApiInteractor( + namespace.shed_url, + namespace.admin_key, + ) + if namespace.user_key is None: + ensure_user_with_email(admin_interactor, DEFAULT_USER, DEFAULT_USER_PASSWORD) + user_key = admin_interactor.create_api_key(DEFAULT_USER, DEFAULT_USER_PASSWORD) + else: + user_key = namespace.user_key + + user_interactor = ShedApiInteractor(namespace.shed_url, user_key) + return RemoteToolShedPopulator( + admin_interactor, + user_interactor, + ) + + +def _arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(description=DESCRIPTION) + parser.add_argument("-u", "--shed-url", default="http://localhost:9009", help="Tool Shed URL") + parser.add_argument("-a", "--admin-key", default="tsadminkey", help="Tool Shed Admin API Key") + parser.add_argument( + "-k", "--user-key", default=None, help="Tool Shed User API Key (will create a new user if unspecified)" + ) + return parser + + +if __name__ == "__main__": + main(sys.argv[1:])