diff --git a/lib/tool_shed/test/base/browser.py b/lib/tool_shed/test/base/browser.py new file mode 100644 index 00000000000..559ce7468cf --- /dev/null +++ b/lib/tool_shed/test/base/browser.py @@ -0,0 +1,49 @@ +import abc +from typing import ( + List, + Union, +) + +FormValueType = Union[str, bool] + + +class ShedBrowser(metaclass=abc.ABCMeta): + @abc.abstractmethod + def visit_url(self, url: str, allowed_codes: List[int]) -> str: + """Navigate to the supplied URL.""" + + @abc.abstractmethod + def page_content(self) -> str: + """Return the page content for""" + + @abc.abstractmethod + def check_page_for_string(self, patt: str) -> None: + """Looks for 'patt' in the current browser page""" + + @abc.abstractmethod + def check_string_not_in_page(self, patt: str) -> None: + """Looks for 'patt' not being in the current browser page""" + + @abc.abstractmethod + def fill_form_value(self, form_name: str, control_name: str, value: FormValueType): + """Fill in a form value.""" + + @abc.abstractmethod + def submit_form(self, form_no=-1, button="runtool_btn", form=None, **kwd): + """Submit the target button.""" + + @abc.abstractmethod + def submit_form_with_name(self, form_name: str, button="runtool_btn", **kwd): + """Submit the target button.""" + + @abc.abstractproperty + def is_twill(self) -> bool: + """Return whether this is a twill browser.""" + + @abc.abstractmethod + def edit_repository_categories(self, categories_to_add: List[str], categories_to_remove: List[str]) -> None: + """Select some new categories and then restore the component.""" + + @abc.abstractmethod + def grant_users_access(self, usernames: List[str]) -> None: + """Select users to grant access to.""" diff --git a/lib/tool_shed/test/base/twillbrowser.py b/lib/tool_shed/test/base/twillbrowser.py new file mode 100644 index 00000000000..65cf8c48eda --- /dev/null +++ b/lib/tool_shed/test/base/twillbrowser.py @@ -0,0 +1,150 @@ +import tempfile +from typing import ( + Dict, + List, +) + +import twill.commands as tc +from twill.browser import FormElement + +from galaxy.util import smart_str +from .browser import ( + FormValueType, + ShedBrowser, +) + +tc.options["equiv_refresh_interval"] = 0 + + +def visit_url(url: str, allowed_codes: List[int]) -> str: + new_url = tc.go(url) + return_code = tc.browser.code + assert return_code in allowed_codes, "Invalid HTTP return code {}, allowed codes: {}".format( + return_code, + ", ".join(str(code) for code in allowed_codes), + ) + return new_url + + +def page_content() -> str: + return tc.browser.html + + +class TwillShedBrowser(ShedBrowser): + def visit_url(self, url: str, allowed_codes: List[int]) -> str: + return visit_url(url, allowed_codes=allowed_codes) + + def page_content(self) -> str: + """ + Return the last visited page (usually HTML, but can binary data as + well). + """ + return page_content() + + def check_page_for_string(self, patt: str) -> None: + page = self.page_content() + if page.find(patt) == -1: + fname = self.write_temp_file(page) + errmsg = f"no match to '{patt}'\npage content written to '{fname}'\npage: [[{page}]]" + raise AssertionError(errmsg) + + def check_string_not_in_page(self, patt: str) -> None: + page = self.page_content() + if page.find(patt) != -1: + fname = self.write_temp_file(page) + errmsg = f"string ({patt}) incorrectly displayed in page.\npage content written to '{fname}'" + raise AssertionError(errmsg) + + def write_temp_file(self, content, suffix=".html"): + with tempfile.NamedTemporaryFile(suffix=suffix, prefix="twilltestcase-", delete=False) as fh: + fh.write(smart_str(content)) + return fh.name + + def submit_form_with_name(self, form_name: str, button="runtool_btn", **kwd): + forms_by_name: Dict[str, FormElement] = {f.get("name"): f for f in self._show_forms()} + form = forms_by_name[form_name] + self._submit_form(form, button, **kwd) + + def _show_forms(self) -> List[FormElement]: + """Shows form, helpful for debugging new tests""" + return tc.browser.forms + + def submit_form(self, form_no=-1, button="runtool_btn", form=None, **kwd): + if form is None: + try: + form = self._show_forms()[form_no] + except IndexError: + raise ValueError("No form to submit found") + self._submit_form(form, button, **kwd) + + def _submit_form(self, form, button, **kwd): + controls = {c.name: c for c in form.inputs} + form_name = form.get("name") + for control_name, control_value in kwd.items(): + if control_name not in controls: + continue # these cannot be handled safely - cause the test to barf out + if not isinstance(control_value, list): + control_value = [str(control_value)] + control = controls[control_name] + control_type = getattr(control, "type", None) + if control_type in ( + "text", + "textfield", + "submit", + "password", + "TextareaElement", + "checkbox", + "radio", + None, + ): + for cv in control_value: + tc.fv(form_name, control.name, cv) + else: + # Add conditions for other control types here when necessary. + pass + tc.submit(button) + + def fill_form_value(self, form_name: str, control_name: str, value: FormValueType): + value = str(value) + tc.fv(form_name, control_name, value) + + def edit_repository_categories(self, categories_to_add: List[str], categories_to_remove: List[str]) -> None: + """Select some new categories and then restore the component.""" + strings_displayed = [] + strings_not_displayed = [] + for category in categories_to_add: + self.fill_form_value("categories", "category_id", f"+{category}") + strings_displayed.append(f"selected>{category}") + for category in categories_to_remove: + self.fill_form_value("categories", "category_id", f"-{category}") + strings_not_displayed.append(f"selected>{category}") + self.submit_form_with_name("categories", "manage_categories_button") + self._check_for_strings(strings_displayed, strings_not_displayed) + + strings_displayed = [] + strings_not_displayed = [] + for category in categories_to_remove: + self.fill_form_value("categories", "category_id", f"+{category}") + strings_displayed.append(f"selected>{category}") + for category in categories_to_add: + self.fill_form_value("categories", "category_id", f"-{category}") + strings_not_displayed.append(f"selected>{category}") + self.submit_form_with_name("categories", "manage_categories_button") + self._check_for_strings(strings_displayed, strings_not_displayed) + + def grant_users_access(self, usernames: List[str]): + for username in usernames: + self.fill_form_value("user_access", "allow_push", f"+{username}") + self.submit_form_with_name("user_access", "user_access_button") + + @property + def is_twill(self) -> bool: + return True + + def _check_for_strings(self, strings_displayed: List[str], strings_not_displayed: List[str]): + if strings_displayed: + for check_str in strings_displayed: + self.check_page_for_string(check_str) + if strings_not_displayed: + for check_str in strings_not_displayed: + self.check_string_not_in_page(check_str) diff --git a/lib/tool_shed/test/base/twilltestcase.py b/lib/tool_shed/test/base/twilltestcase.py index b88a8f05728..312091e7a11 100644 --- a/lib/tool_shed/test/base/twilltestcase.py +++ b/lib/tool_shed/test/base/twilltestcase.py @@ -22,8 +22,8 @@ from urllib.parse import ( urlparse, ) +import pytest import requests -import twill.commands as tc from mercurial import ( commands, hg, @@ -51,7 +51,6 @@ from galaxy.tool_shed.util.repository_util import check_for_updates from galaxy.util import ( DEFAULT_SOCKET_TIMEOUT, smart_str, - unicodify, ) from galaxy_test.base.api_asserts import assert_status_code_is_ok from galaxy_test.base.api_util import get_admin_api_key @@ -74,12 +73,16 @@ from . import ( test_db_util, ) from .api import ShedApiTestCase +from .browser import ShedBrowser +from .twillbrowser import ( + page_content, + visit_url, +) # Set a 10 minute timeout for repository installation. repository_installation_timeout = 600 log = logging.getLogger(__name__) -tc.options["equiv_refresh_interval"] = 0 class ToolShedInstallationClient(metaclass=abc.ABCMeta): @@ -398,7 +401,7 @@ class GalaxyInteractorToolShedInstallationClient(ToolShedInstallationClient): def _galaxy_token(self): self._visit_galaxy_url("/") - html = self.testcase.last_page() + html = page_content() token_def_index = html.find("session_csrf_token") token_sep_index = html.find(":", token_def_index) token_quote_start_index = html.find('"', token_sep_index) @@ -411,7 +414,7 @@ class GalaxyInteractorToolShedInstallationClient(ToolShedInstallationClient): tool_guid = quote_plus(tool_metadata[0]["guid"], safe="") api_url = f"/api/tools/{tool_guid}" self._visit_galaxy_url(api_url) - tool_dict = loads(self.testcase.last_page()) + tool_dict = loads(page_content()) tool_panel_section = tool_dict["panel_section_name"] return tool_panel_section @@ -425,11 +428,12 @@ class GalaxyInteractorToolShedInstallationClient(ToolShedInstallationClient): ) _wait_for_installation(galaxy_repository, test_db_util.ga_refresh) - def _visit_galaxy_url(self, url, params=None, doseq=False, allowed_codes=None): + def _visit_galaxy_url(self, url, params=None, allowed_codes=None): if allowed_codes is None: allowed_codes = [200] url = f"{self.testcase.galaxy_url}{url}" - self.testcase.visit_url(url, params=params, doseq=doseq, allowed_codes=allowed_codes) + url = self.testcase.join_url_and_params(url, params) + return visit_url(url, allowed_codes) class StandaloneToolShedInstallationClient(ToolShedInstallationClient): @@ -607,11 +611,13 @@ class StandaloneToolShedInstallationClient(ToolShedInstallationClient): return tool_names +@pytest.mark.usefixtures("shed_browser") class ShedTwillTestCase(ShedApiTestCase): """Class of FunctionalTestCase geared toward HTML interactions using the Twill library.""" requires_galaxy: bool = False _installation_client = None + __browser: Optional[ShedBrowser] = None def setUp(self): super().setUp() @@ -639,6 +645,15 @@ class ShedTwillTestCase(ShedApiTestCase): self.__class__._installation_client.setup() self._installation_client = self.__class__._installation_client + @pytest.fixture(autouse=True) + def inject_shed_browser(self, shed_browser: ShedBrowser): + self.__browser = shed_browser + + @property + def _browser(self) -> ShedBrowser: + assert self.__browser + return self.__browser + def check_for_strings(self, strings_displayed=None, strings_not_displayed=None): strings_displayed = strings_displayed or [] strings_not_displayed = strings_not_displayed or [] @@ -660,19 +675,11 @@ class ShedTwillTestCase(ShedApiTestCase): def check_page_for_string(self, patt): """Looks for 'patt' in the current browser page""" - page = unicodify(self.last_page()) - if page.find(patt) == -1: - fname = self.write_temp_file(page) - errmsg = f"no match to '{patt}'\npage content written to '{fname}'\npage: [[{page}]]" - raise AssertionError(errmsg) + self._browser.check_page_for_string(patt) def check_string_not_in_page(self, patt): """Checks to make sure 'patt' is NOT in the page.""" - page = self.last_page() - if page.find(patt) != -1: - fname = self.write_temp_file(page) - errmsg = f"string ({patt}) incorrectly displayed in page.\npage content written to '{fname}'" - raise AssertionError(errmsg) + self._browser.check_string_not_in_page(patt) # Functions associated with user accounts @@ -681,12 +688,12 @@ class ShedTwillTestCase(ShedApiTestCase): # can't find form fields (and hence user can't be logged in). params = dict(cntrller=cntrller, use_panels=False) self.visit_url("/user/create", params) - tc.fv("registration", "email", email) - tc.fv("registration", "redirect", redirect) - tc.fv("registration", "password", password) - tc.fv("registration", "confirm", password) - tc.fv("registration", "username", username) - tc.submit("create_user_button") + self._browser.fill_form_value("registration", "email", email) + self._browser.fill_form_value("registration", "redirect", redirect) + self._browser.fill_form_value("registration", "password", password) + self._browser.fill_form_value("registration", "confirm", password) + self._browser.fill_form_value("registration", "username", username) + self._browser.submit_form_with_name("registration", "create_user_button") previously_created = False username_taken = False invalid_username = False @@ -715,10 +722,7 @@ class ShedTwillTestCase(ShedApiTestCase): Return the last visited page (usually HTML, but can binary data as well). """ - return tc.browser.html - - def last_url(self): - return tc.browser.url + return self._browser.page_content() def user_api_interactor(self, email="test@bx.psu.edu", password="testuser"): return self._api_interactor_by_credentials(email, password) @@ -727,7 +731,12 @@ class ShedTwillTestCase(ShedApiTestCase): return self._get_populator(self.user_api_interactor(email=email, password=password)) def login( - self, email="test@bx.psu.edu", password="testuser", username="admin-user", redirect="", logout_first=True + self, + email: str = "test@bx.psu.edu", + password: str = "testuser", + username: str = "admin-user", + redirect: str = "", + logout_first: bool = True, ): # Clear cookies. if logout_first: @@ -748,69 +757,37 @@ class ShedTwillTestCase(ShedApiTestCase): self.visit_url("/user/logout") self.check_page_for_string("You have been logged out") - def showforms(self): - """Shows form, helpful for debugging new tests""" - return tc.browser.forms - def submit_form(self, form_no=-1, button="runtool_btn", form=None, **kwd): """Populates and submits a form from the keyword arguments.""" # An HTMLForm contains a sequence of Controls. Supported control classes are: # TextControl, FileControl, ListControl, RadioControl, CheckboxControl, SelectControl, # SubmitControl, ImageControl - if form is None: - try: - form = self.showforms()[form_no] - except IndexError: - raise ValueError("No form to submit found") - controls = {c.name: c for c in form.inputs} - form_name = form.get("name") - for control_name, control_value in kwd.items(): - if control_name not in controls: - continue # these cannot be handled safely - cause the test to barf out - if not isinstance(control_value, list): - control_value = [str(control_value)] - control = controls[control_name] - control_type = getattr(control, "type", None) - if control_type in ( - "text", - "textfield", - "submit", - "password", - "TextareaElement", - "checkbox", - "radio", - None, - ): - for cv in control_value: - tc.fv(form_name, control.name, cv) - else: - # Add conditions for other control types here when necessary. - pass - tc.submit(button) + self._browser.submit_form(form_no, button, form, **kwd) - def visit_url(self, url, params=None, doseq=False, allowed_codes=None): - if allowed_codes is None: - allowed_codes = [200] + def join_url_and_params(self, url: str, params, query=None) -> str: if params is None: params = dict() + if query is None: + query = urlparse(url).query + if query: + for query_parameter in query.split("&"): + key, value = query_parameter.split("=") + params[key] = value + if params: + url += f"?{urlencode(params)}" + return url + + def visit_url(self, url: str, params=None, allowed_codes: Optional[List[int]] = None) -> str: parsed_url = urlparse(url) if len(parsed_url.netloc) == 0: url = f"http://{self.host}:{self.port}{parsed_url.path}" else: url = f"{parsed_url.scheme}://{parsed_url.netloc}{parsed_url.path}" - if parsed_url.query: - for query_parameter in parsed_url.query.split("&"): - key, value = query_parameter.split("=") - params[key] = value - if params: - url += f"?{urlencode(params, doseq=doseq)}" - new_url = tc.go(url) - return_code = tc.browser.code - assert return_code in allowed_codes, "Invalid HTTP return code {}, allowed codes: {}".format( - return_code, - ", ".join(str(code) for code in allowed_codes), - ) - return new_url + url = self.join_url_and_params(url, params, query=parsed_url.query) + if allowed_codes is None: + allowed_codes = [200] + + return self._browser.visit_url(url, allowed_codes=allowed_codes) def write_temp_file(self, content, suffix=".html"): with tempfile.NamedTemporaryFile(suffix=suffix, prefix="twilltestcase-", delete=False) as fh: @@ -1220,33 +1197,15 @@ class ShedTwillTestCase(ShedApiTestCase): self.check_for_strings(strings_displayed, strings_not_displayed) def edit_repository_categories( - self, repository: Repository, categories_to_add=None, categories_to_remove=None, restore_original=True + self, + repository: Repository, + categories_to_add: List[str], + categories_to_remove: List[str], + restore_original=True, ) -> None: - categories_to_add = categories_to_add or [] - categories_to_remove = categories_to_remove or [] params = {"id": repository.id} self.visit_url("/repository/manage_repository", params=params) - strings_displayed = [] - strings_not_displayed = [] - for category in categories_to_add: - tc.fv("2", "category_id", f"+{category}") - strings_displayed.append(f"selected>{category}") - for category in categories_to_remove: - tc.fv("2", "category_id", f"-{category}") - strings_not_displayed.append(f"selected>{category}") - tc.submit("manage_categories_button") - self.check_for_strings(strings_displayed, strings_not_displayed) - if restore_original: - strings_displayed = [] - strings_not_displayed = [] - for category in categories_to_remove: - tc.fv("2", "category_id", f"+{category}") - strings_displayed.append(f"selected>{category}") - for category in categories_to_add: - tc.fv("2", "category_id", f"-{category}") - strings_not_displayed.append(f"selected>{category}") - tc.submit("manage_categories_button") - self.check_for_strings(strings_displayed, strings_not_displayed) + self._browser.edit_repository_categories(categories_to_add, categories_to_remove) def edit_repository_information(self, repository: Repository, revert=True, **kwd): params = {"id": repository.id} @@ -1260,17 +1219,21 @@ class ShedTwillTestCase(ShedApiTestCase): strings_displayed = [] for input_elem_name in ["repo_name", "description", "long_description", "repository_type"]: if input_elem_name in kwd: - tc.fv("edit_repository", input_elem_name, kwd[input_elem_name]) + self._browser.fill_form_value("edit_repository", input_elem_name, kwd[input_elem_name]) strings_displayed.append(self.escape_html(kwd[input_elem_name])) - tc.submit("edit_repository_button") - self.check_for_strings(strings_displayed) + self._browser.submit_form_with_name("edit_repository", "edit_repository_button") + # TODO: come back to this (and similar conditional below), the problem is check + # for strings isn't working with with textboxes I think? + if self._browser.is_twill: + self.check_for_strings(strings_displayed) if revert: strings_displayed = [] for input_elem_name in ["repo_name", "description", "long_description"]: - tc.fv("edit_repository", input_elem_name, original_information[input_elem_name]) + self._browser.fill_form_value("edit_repository", input_elem_name, original_information[input_elem_name]) strings_displayed.append(self.escape_html(original_information[input_elem_name])) - tc.submit("edit_repository_button") - self.check_for_strings(strings_displayed) + self._browser.submit_form_with_name("edit_repository", "edit_repository_button") + if self._browser.is_twill: + self.check_for_strings(strings_displayed) def enable_email_alerts(self, repository: Repository, strings_displayed=None, strings_not_displayed=None) -> None: repository_id = repository.id @@ -1562,9 +1525,7 @@ class ShedTwillTestCase(ShedApiTestCase): usernames = usernames or [] self.display_manage_repository_page(repository) self.check_for_strings(strings_displayed, strings_not_displayed) - for username in usernames: - tc.fv("user_access", "allow_push", f"+{username}") - tc.submit("user_access_button") + self._browser.grant_users_access(usernames) self.check_for_strings(post_submit_strings_displayed, post_submit_strings_not_displayed) def _install_repository( @@ -1755,28 +1716,11 @@ class ShedTwillTestCase(ShedApiTestCase): params["galaxy_url"] = self.galaxy_url for field_name, search_string in search_fields.items(): self.visit_url("/repository/find_tools", params=params) - tc.fv("1", "exact_matches", exact_matches) - tc.fv("1", field_name, search_string) - tc.submit() + self._browser.fill_form_value("find_tools", "exact_matches", exact_matches) + self._browser.fill_form_value("find_tools", field_name, search_string) + self._browser.submit_form_with_name("find_tools", "find_tools_submit") self.check_for_strings(strings_displayed, strings_not_displayed) - def set_form_value(self, form, kwd, field_name, field_value): - """ - Set the form field field_name to field_value if it exists, and return the provided dict containing that value. If - the field does not exist in the provided form, return a dict without that index. - """ - form_id = form.attrib.get("id") - controls = [control for control in form.inputs if str(control.name) == field_name] - if len(controls) > 0: - log.debug(f"Setting field {field_name} of form {form_id} to {field_value}.") - tc.formvalue(form_id, field_name, str(field_value)) - kwd[field_name] = str(field_value) - else: - if field_name in kwd: - log.debug("No field %s in form %s, discarding from return value.", field_name, form_id) - del kwd[field_name] - return kwd - def set_repository_deprecated( self, repository: Repository, set_deprecated=True, strings_displayed=None, strings_not_displayed=None ): @@ -1788,8 +1732,8 @@ class ShedTwillTestCase(ShedApiTestCase): self, repository: Repository, set_malicious=True, strings_displayed=None, strings_not_displayed=None ) -> None: self.display_manage_repository_page(repository) - tc.fv("malicious", "malicious", set_malicious) - tc.submit("malicious_button") + 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) def tip_has_metadata(self, repository: Repository) -> bool: diff --git a/lib/tool_shed/test/functional/test_0000_basic_repository_features.py b/lib/tool_shed/test/functional/test_0000_basic_repository_features.py index d9092076a40..029f2987866 100644 --- a/lib/tool_shed/test/functional/test_0000_basic_repository_features.py +++ b/lib/tool_shed/test/functional/test_0000_basic_repository_features.py @@ -353,7 +353,8 @@ class TestBasicRepositoryFeatures(ShedTwillTestCase): repository = self._get_repository_by_name_and_owner(repository_name, common.test_user_1_name) encoded_repository_id = repository.id strings_displayed = ["Invalid+changeset+revision"] - self.visit_url( + view_repo_url = ( f"/repository/view_repository?id={encoded_repository_id}&changeset_revision=nonsensical_changeset" ) + self.visit_url(view_repo_url) self.check_for_strings(strings_displayed=strings_displayed, strings_not_displayed=[]) diff --git a/lib/tool_shed/webapp/templates/webapps/tool_shed/repository/find_tools.mako b/lib/tool_shed/webapp/templates/webapps/tool_shed/repository/find_tools.mako index 6a49ea0a041..a9ea1825c30 100644 --- a/lib/tool_shed/webapp/templates/webapps/tool_shed/repository/find_tools.mako +++ b/lib/tool_shed/webapp/templates/webapps/tool_shed/repository/find_tools.mako @@ -55,7 +55,7 @@