Implement framework for testing API...

... if you can call 1 new class a framework. Includes a few test cases to exercise/drive it. These examples include a histories API test (a typical API test) and a general test of the API framework itself (mostly just the run_as functionality).

This includes changes to interactor.py to make it more useful outside the context of tool/workflow testing as well as a tweak to test Galaxy that gets started to allow testing of the run_as feature.

scripts/functional_tests.py

@@ -346,6 +346,7 @@ def main():
     app = None
     if start_server:
         kwargs = dict( admin_users='test@bx.psu.edu',
+                       api_allow_run_as='test@bx.psu.edu',
                        allow_library_path_paste=True,
                        allow_user_creation=True,
                        allow_user_deletion=True,

test/base/api.py

@@ -0,0 +1,85 @@
+# TODO: We don't need all of TwillTestCase, strip down to a common super class
+# shared by API and Twill test cases.
+from .twilltestcase import TwillTestCase
+
+from base.interactor import GalaxyInteractorApi as BaseInteractor
+
+from .api_util import get_master_api_key
+from .api_util import get_user_api_key
+
+from urllib import urlencode
+
+
+TEST_USER = "user@bx.psu.edu"
+
+
+# TODO: Allow these to point at existing Galaxy instances.
+class ApiTestCase( TwillTestCase ):
+
+    def setUp( self ):
+        super( ApiTestCase, self ).setUp( )
+        self.user_api_key = get_user_api_key()
+        self.master_api_key = get_master_api_key()
+        self.galaxy_interactor = ApiTestInteractor( self )
+
+    def _api_url( self, path, params=None, use_key=None ):
+        if not params:
+            params = {}
+        url = "%s/api/%s" % ( self.url, path )
+        if use_key:
+            params[ "key" ] = self.galaxy_interactor.api_key
+        query = urlencode( params )
+        if query:
+            url = "%s?%s" % ( url, query )
+        return url
+
+    def _setup_user( self, email ):
+        self.galaxy_interactor.ensure_user_with_email(email)
+        users = self._get( "users", admin=True ).json()
+        user = [ user for user in users if user["email"] == email ][0]
+        return user
+
+    def _get( self, *args, **kwds ):
+        return self.galaxy_interactor.get( *args, **kwds )
+
+    def _post( self, *args, **kwds ):
+        return self.galaxy_interactor.post( *args, **kwds )
+
+    def _assert_status_code_is( self, response, expected_status_code ):
+        response_status_code = response.status_code
+        if expected_status_code != response_status_code:
+            try:
+                body = response.json()
+            except Exception:
+                body = "INVALID JSON RESPONSE"
+            assertion_message_template = "Request status code (%d) was not expected value %d. Body was %s"
+            assertion_message = assertion_message_template % ( response_status_code, expected_status_code, body )
+            raise AssertionError( assertion_message )
+
+    def _assert_has_keys( self, response, *keys ):
+        for key in keys:
+            assert key in response, "Response [%s] does not contain key [%s]" % ( response, key )
+
+    def _random_key( self ):  # Used for invalid request testing...
+        return "1234567890123456"
+
+    _assert_has_key = _assert_has_keys
+
+
+class ApiTestInteractor( BaseInteractor ):
+    """ Specialized variant of the API interactor (originally developed for
+    tool functional tests) for testing the API generally.
+    """
+
+    def __init__( self, test_case ):
+        super( ApiTestInteractor, self ).__init__( test_case, test_user=TEST_USER )
+
+    # This variant the lower level get and post methods are meant to be used
+    # directly to test API - instead of relying on higher-level constructs for
+    # specific pieces of the API (the way it is done with the variant for tool)
+    # testing.
+    def get( self, *args, **kwds ):
+        return self._get( *args, **kwds )
+
+    def post( self, *args, **kwds ):
+        return self._post( *args, **kwds )

test/functional/api/test_framework.py

@@ -0,0 +1,27 @@
+# This file doesn't test any API in particular but is meant to functionally
+# test the API framework itself.
+from base import api
+
+
+class ApiFrameworkTestCase( api.ApiTestCase ):
+
+    # Next several tests test the API's run_as functionality.
+    def test_user_cannont_run_as( self ):
+        post_data = dict( name="TestHistory1", run_as="another_user" )
+        # Normal user cannot run_as...
+        create_response = self._post( "histories", data=post_data )
+        self._assert_status_code_is( create_response, 403 )
+
+    def test_run_as_invalid_user( self ):
+        post_data = dict( name="TestHistory1", run_as="another_user" )
+        # admin user can run_as, but this user doesn't exist, expect 400.
+        create_response = self._post( "histories", data=post_data, admin=True )
+        self._assert_status_code_is( create_response, 400 )
+
+    def test_run_as_valid_user( self ):
+        run_as_user = self._setup_user( "for_run_as@bx.psu.edu" )
+        post_data = dict( name="TestHistory1", run_as=run_as_user[ "id" ] )
+        # Use run_as with admin user and for another user just created, this
+        # should work.
+        create_response = self._post( "histories", data=post_data, admin=True )
+        self._assert_status_code_is( create_response, 200 )

test/functional/api/test_histories.py

@@ -0,0 +1,26 @@
+from base import api
+# requests.post or something like it if unavailable
+from base.interactor import post_request
+
+
+class HistoriesApiTestCase( api.ApiTestCase ):
+
+    def test_create_history( self ):
+        # Create a history.
+        post_data = dict( name="TestHistory1" )
+        create_response = self._post( "histories", data=post_data ).json()
+        self._assert_has_keys( create_response, "name", "id" )
+        self.assertEquals( create_response[ "name" ], "TestHistory1" )
+        created_id = create_response[ "id" ]
+
+        # Make sure new history appears in index of user's histories.
+        index_response = self._get( "histories" ).json()
+        indexed_history = [ h for h in index_response if h[ "id" ] == created_id ][0]
+        self.assertEquals(indexed_history[ "name" ], "TestHistory1")
+
+    def test_create_anonymous_fails( self ):
+        post_data = dict( name="CannotCreate" )
+        # Using lower-level _api_url will cause key to not be injected.
+        histories_url = self._api_url( "histories" )
+        create_response = post_request( url=histories_url, data=post_data )
+        self._assert_status_code_is( create_response, 403 )