Refactor tool shed tests for injection of browser.

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."""

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)

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:

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=[])

lib/tool_shed/webapp/templates/webapps/tool_shed/repository/find_tools.mako

@@ -55,7 +55,7 @@
             </div>
             <div style="clear: both"></div>
             <div class="form-row">
-                <input type="submit" value="Search repositories"/>
+                <input type="submit" name="find_tools_submit" value="Search repositories"/>
             </div>
         </form>
     </div>