Add functional tests for pages API.

Touch up pages API in response to test cases. Update pages API to use newer API decorator.

lib/galaxy/webapps/galaxy/api/page_revisions.py

@@ -2,8 +2,9 @@
 API for updating Galaxy Pages
 """
 import logging
-from galaxy import web
+from galaxy.web import _future_expose_api as expose_api
 from galaxy.web.base.controller import SharableItemSecurityMixin, BaseAPIController, SharableMixin
+from galaxy import exceptions
 from galaxy.model.item_attrs import UsesAnnotations
 from galaxy.util.sanitize_html import sanitize_html
 
@@ -12,7 +13,7 @@ log = logging.getLogger( __name__ )
 
 class PageRevisionsController( BaseAPIController, SharableItemSecurityMixin, UsesAnnotations, SharableMixin ):
 
-    @web.expose_api
+    @expose_api
     def index( self, trans, page_id, **kwd ):
         """
         index( self, trans, page_id, **kwd )
@@ -24,14 +25,16 @@ class PageRevisionsController( BaseAPIController, SharableItemSecurityMixin, Use
         :rtype:     list
         :returns:   dictionaries containing different revisions of the page
         """
+        page = self._get_page( trans, page_id )
+        self._verify_page_ownership( trans, page )
+
         r = trans.sa_session.query( trans.app.model.PageRevision ).filter_by( page_id=trans.security.decode_id(page_id) )
         out = []
         for page in r:
-            if self.security_check( trans, page, True, True ):
-                out.append( self.encode_all_ids( trans, page.to_dict(), True) )
+            out.append( self.encode_all_ids( trans, page.to_dict(), True) )
         return out
 
-    @web.expose_api
+    @expose_api
     def create( self, trans, page_id, payload, **kwd ):
         """
         create( self, trans, page_id, payload **kwd )
@@ -46,39 +49,42 @@ class PageRevisionsController( BaseAPIController, SharableItemSecurityMixin, Use
         :rtype:     dictionary
         :returns:   Dictionary with 'success' or 'error' element to indicate the result of the request
         """
-        error_str = ""
+        content = payload.get("content", None)
+        if not content:
+            raise exceptions.ObjectAttributeMissingException("content undefined or empty")
 
-        if not page_id:
-            error_str = "page_id is required"
-        elif not payload.get("content", None):
-            error_str = "content is required"
+        page = self._get_page( trans, page_id )
+        self._verify_page_ownership( trans, page )
+
+        if 'title' in payload:
+            title = payload['title']
         else:
+            title = page.title
 
-            # Create the new stored page
+        content = sanitize_html( content, 'utf-8', 'text/html' )
+
+        page_revision = trans.app.model.PageRevision()
+        page_revision.title = title
+        page_revision.page = page
+        page.latest_revision = page_revision
+        page_revision.content = content
+
+        # Persist
+        session = trans.sa_session
+        session.flush()
+
+        return page_revision.to_dict( view="element" )
+
+    def _get_page( self, trans, page_id ):
+        page = None
+        try:
             page = trans.sa_session.query( trans.app.model.Page ).get( trans.security.decode_id(page_id) )
-            if page is None:
-                return { "error" : "page not found"}
+        except Exception:
+            pass
+        if not page:
+            raise exceptions.ObjectNotFound()
+        return page
 
-            if not self.security_check( trans, page, True, True ):
-                return { "error" : "page not found"}
-
-            if 'title' in payload:
-                title = payload['title']
-            else:
-                title = page.title
-
-            content = payload.get("content", "")
-            content = sanitize_html( content, 'utf-8', 'text/html' )
-
-            page_revision = trans.app.model.PageRevision()
-            page_revision.title = title
-            page_revision.page = page
-            page.latest_revision = page_revision
-            page_revision.content = content
-            # Persist
-            session = trans.sa_session
-            session.flush()
-
-            return { "success" : "revision posted" }
-
-        return { "error" : error_str }
+    def _verify_page_ownership( self, trans, page ):
+        if not self.security_check( trans, page, True, True ):
+            raise exceptions.ItemOwnershipException()

lib/galaxy/webapps/galaxy/api/pages.py

@@ -2,8 +2,9 @@
 API for updating Galaxy Pages
 """
 import logging
-from galaxy import web
+from galaxy.web import _future_expose_api as expose_api
 from galaxy.web.base.controller import SharableItemSecurityMixin, BaseAPIController, SharableMixin
+from galaxy import exceptions
 from galaxy.model.item_attrs import UsesAnnotations
 from galaxy.util.sanitize_html import sanitize_html
 
@@ -12,7 +13,7 @@ log = logging.getLogger( __name__ )
 
 class PagesController( BaseAPIController, SharableItemSecurityMixin, UsesAnnotations, SharableMixin ):
 
-    @web.expose_api
+    @expose_api
     def index( self, trans, deleted=False, **kwd ):
         """
         index( self, trans, deleted=False, **kwd )
@@ -47,7 +48,7 @@ class PagesController( BaseAPIController, SharableItemSecurityMixin, UsesAnnotat
 
         return out
 
-    @web.expose_api
+    @expose_api
     def create( self, trans, payload, **kwd ):
         """
         create( self, trans, payload, **kwd )
@@ -64,45 +65,41 @@ class PagesController( BaseAPIController, SharableItemSecurityMixin, UsesAnnotat
         :returns:   Dictionary return of the Page.to_dict call
         """
         user = trans.get_user()
-        error_str = ""
 
         if not payload.get("title", None):
-            error_str = "Page name is required"
+            raise exceptions.ObjectAttributeMissingException( "Page name is required" )
         elif not payload.get("slug", None):
-            error_str = "Page id is required"
+            raise exceptions.ObjectAttributeMissingException( "Page id is required" )
         elif not self._is_valid_slug( payload["slug"] ):
-            error_str = "Page identifier must consist of only lowercase letters, numbers, and the '-' character"
+            raise exceptions.ObjectAttributeInvalidException( "Page identifier must consist of only lowercase letters, numbers, and the '-' character" )
         elif trans.sa_session.query( trans.app.model.Page ).filter_by( user=user, slug=payload["slug"], deleted=False ).first():
-            error_str = "Page id must be unique"
-        else:
+            raise exceptions.DuplicatedSlugException( "Page slug must be unique" )
 
-            content = payload.get("content", "")
-            content = sanitize_html( content, 'utf-8', 'text/html' )
+        content = payload.get("content", "")
+        content = sanitize_html( content, 'utf-8', 'text/html' )
 
-            # Create the new stored page
-            page = trans.app.model.Page()
-            page.title = payload['title']
-            page.slug = payload['slug']
-            page_annotation = sanitize_html( payload.get( "annotation", "" ), 'utf-8', 'text/html' )
-            self.add_item_annotation( trans.sa_session, trans.get_user(), page, page_annotation )
-            page.user = user
-            # And the first (empty) page revision
-            page_revision = trans.app.model.PageRevision()
-            page_revision.title = payload['title']
-            page_revision.page = page
-            page.latest_revision = page_revision
-            page_revision.content = content
-            # Persist
-            session = trans.sa_session
-            session.add( page )
-            session.flush()
+        # Create the new stored page
+        page = trans.app.model.Page()
+        page.title = payload['title']
+        page.slug = payload['slug']
+        page_annotation = sanitize_html( payload.get( "annotation", "" ), 'utf-8', 'text/html' )
+        self.add_item_annotation( trans.sa_session, trans.get_user(), page, page_annotation )
+        page.user = user
+        # And the first (empty) page revision
+        page_revision = trans.app.model.PageRevision()
+        page_revision.title = payload['title']
+        page_revision.page = page
+        page.latest_revision = page_revision
+        page_revision.content = content
+        # Persist
+        session = trans.sa_session
+        session.add( page )
+        session.flush()
 
-            rval = self.encode_all_ids( trans, page.to_dict(), True )
-            return rval
+        rval = self.encode_all_ids( trans, page.to_dict(), True )
+        return rval
 
-        return { "error" : error_str }
-
-    @web.expose_api
+    @expose_api
     def delete( self, trans, id, **kwd ):
         """
         delete( self, trans, id, **kwd )
@@ -114,22 +111,14 @@ class PagesController( BaseAPIController, SharableItemSecurityMixin, UsesAnnotat
         :rtype:     dict
         :returns:   Dictionary with 'success' or 'error' element to indicate the result of the request
         """
-        page_id = id
-        try:
-            page = trans.sa_session.query(self.app.model.Page).get(trans.security.decode_id(page_id))
-        except Exception, e:
-            return { "error" : "Page with ID='%s' can not be found\n Exception: %s" % (page_id, str( e )) }
+        page = self._get_page( trans, id )
 
-        # check to see if user has permissions to selected workflow
-        if page.user != trans.user and not trans.user_is_admin():
-            return { "error" : "Workflow is not owned by or shared with current user" }
-
-        #Mark a workflow as deleted
+        #Mark a page as deleted
         page.deleted = True
         trans.sa_session.flush()
-        return { "success" : "Deleted", "id" : page_id }
+        return ''  # TODO: Figure out what to return on DELETE, document in guidelines!
 
-    @web.expose_api
+    @expose_api
     def show( self, trans, id, **kwd ):
         """
         show( self, trans, id, **kwd )
@@ -141,8 +130,22 @@ class PagesController( BaseAPIController, SharableItemSecurityMixin, UsesAnnotat
         :rtype:     dict
         :returns:   Dictionary return of the Page.to_dict call with the 'content' field populated by the most recent revision
         """
-        page = trans.sa_session.query( trans.app.model.Page ).get( trans.security.decode_id( id ) )
+        page = self._get_page( trans, id )
         self.security_check( trans, page, check_ownership=False, check_accessible=True)
         rval = self.encode_all_ids( trans, page.to_dict(), True )
         rval['content'] = page.latest_revision.content
         return rval
+
+    def _get_page( self, trans, id ):  # Fetches page object and verifies security.
+        try:
+            page = trans.sa_session.query( trans.app.model.Page ).get( trans.security.decode_id( id ) )
+        except Exception:
+            page = None
+
+        if not page:
+            raise exceptions.ObjectNotFound()
+
+        if page.user != trans.user and not trans.user_is_admin():
+            raise exceptions.ItemOwnershipException()
+
+        return page

test/base/api.py

@@ -60,6 +60,12 @@ class ApiTestCase( TwillTestCase ):
         for key in keys:
             assert key in response, "Response [%s] does not contain key [%s]" % ( response, key )
 
+    def _assert_error_code_is( self, response, error_code ):
+        if hasattr( response, "json" ):
+            response = response.json()
+        self._assert_has_keys( response, "err_code" )
+        self.assertEquals( response[ "err_code" ], int( error_code ) )
+
     def _random_key( self ):  # Used for invalid request testing...
         return "1234567890123456"
 

test/functional/api/test_page_revisions.py

@@ -0,0 +1,35 @@
+from galaxy.exceptions import error_codes
+from functional.api.pages import BasePageApiTestCase
+
+
+class PageRevisionsApiTestCase( BasePageApiTestCase ):
+
+    def test_create( self ):
+        page_json = self._create_valid_page_with_slug( "pr1" )
+        revision_data = dict( content="<p>NewContent!</p>" )
+        page_revision_response = self._post( "pages/%s/revisions" % page_json[ 'id' ], data=revision_data )
+        self._assert_status_code_is( page_revision_response, 200 )
+        page_revision_json = page_revision_response.json()
+        self._assert_has_keys( page_revision_json, 'id', 'content' )
+
+    def test_403_if_create_revision_on_unowned_page( self ):
+        page_json = self._create_valid_page_as( "pr2@bx.psu.edu", "pr2" )
+        revision_data = dict( content="<p>NewContent!</p>" )
+        page_revision_response = self._post( "pages/%s/revisions" % page_json[ 'id' ], data=revision_data )
+        self._assert_status_code_is( page_revision_response, 403 )
+
+    def test_revision_index( self ):
+        page_json = self._create_valid_page_with_slug( "pr3" )
+        revision_data = dict( content="<p>NewContent!</p>" )
+        revisions_url = "pages/%s/revisions" % page_json[ 'id' ]
+        self._post( revisions_url, data=revision_data )
+        revisions_response = self._get( revisions_url )
+        self._assert_status_code_is( revisions_response, 200 )
+        revisions_json = revisions_response.json()
+        assert len( revisions_json ) == 2  # Original revision and new one
+
+    def test_404_if_index_unknown_page( self ):
+        revisions_url = "pages/%s/revisions" % self._random_key()
+        revisions_response = self._get( revisions_url )
+        self._assert_status_code_is( revisions_response, 404 )
+        self._assert_error_code_is( revisions_response, error_codes.USER_OBJECT_NOT_FOUND )

test/functional/api/test_pages.py

@@ -0,0 +1,105 @@
+from galaxy.exceptions import error_codes
+from base import api
+from base.interactor import delete_request
+
+from operator import itemgetter
+
+
+class BasePageApiTestCase( api.ApiTestCase ):
+
+    def _create_valid_page_with_slug( self, slug ):
+        page_request = self._test_page_payload( slug=slug )
+        page_response = self._post( "pages", page_request )
+        self._assert_status_code_is( page_response, 200 )
+        return page_response.json()
+
+    def _create_valid_page_as( self, other_email, slug ):
+        run_as_user = self._setup_user( other_email )
+        page_request = self._test_page_payload( slug=slug )
+        page_request[ "run_as" ] = run_as_user[ "id" ]
+        page_response = self._post( "pages", page_request, admin=True )
+        self._assert_status_code_is( page_response, 200 )
+        return page_response.json()
+
+    def _test_page_payload( self, **kwds ):
+        request = dict(
+            slug="mypage",
+            title="MY PAGE",
+            content="<p>Page!</p>",
+        )
+        request.update( **kwds )
+        return request
+
+
+class PageApiTestCase( BasePageApiTestCase ):
+
+    def test_create( self ):
+        response_json = self._create_valid_page_with_slug( "mypage" )
+        self._assert_has_keys( response_json, "slug", "title", "id" )
+
+    def test_index( self ):
+        create_response_json = self._create_valid_page_with_slug( "indexpage" )
+        assert self._users_index_has_page_with_id( create_response_json[ "id" ] )
+
+    def test_index_doesnt_show_unavailable_pages( self ):
+        create_response_json = self._create_valid_page_as( "others_page_index@bx.psu.edu", "otherspageindex" )
+        assert not self._users_index_has_page_with_id( create_response_json[ "id" ] )
+
+    def test_cannot_create_pages_with_same_slug( self ):
+        page_request = self._test_page_payload( slug="mypage1" )
+        page_response_1 = self._post( "pages", page_request )
+        self._assert_status_code_is( page_response_1, 200 )
+        page_response_2 = self._post( "pages", page_request )
+        self._assert_status_code_is( page_response_2, 400 )
+        self._assert_error_code_is( page_response_2, error_codes.USER_SLUG_DUPLICATE )
+
+    def test_page_requires_name( self ):
+        page_request = self._test_page_payload()
+        del page_request[ 'title' ]
+        page_response = self._post( "pages", page_request )
+        self._assert_status_code_is( page_response, 400 )
+        self._assert_error_code_is( page_response, error_codes.USER_OBJECT_ATTRIBUTE_MISSING )
+
+    def test_page_requires_slug( self ):
+        page_request = self._test_page_payload()
+        del page_request[ 'slug' ]
+        page_response = self._post( "pages", page_request )
+        self._assert_status_code_is( page_response, 400 )
+
+    def test_delete( self ):
+        response_json = self._create_valid_page_with_slug( "testdelete" )
+        delete_response = delete_request( self._api_url( "pages/%s" % response_json[ 'id' ], use_key=True ) )
+        self._assert_status_code_is( delete_response, 200 )
+
+    def test_404_on_delete_unknown_page( self ):
+        delete_response = delete_request( self._api_url( "pages/%s" % self._random_key(), use_key=True ) )
+        self._assert_status_code_is( delete_response, 404 )
+        self._assert_error_code_is( delete_response, error_codes.USER_OBJECT_NOT_FOUND )
+
+    def test_403_on_delete_unowned_page( self ):
+        page_response = self._create_valid_page_as( "others_page@bx.psu.edu", "otherspage" )
+        delete_response = delete_request( self._api_url( "pages/%s" % page_response[ "id" ], use_key=True ) )
+        self._assert_status_code_is( delete_response, 403 )
+        self._assert_error_code_is( delete_response, error_codes.USER_DOES_NOT_OWN_ITEM )
+
+    def test_show( self ):
+        response_json = self._create_valid_page_with_slug( "pagetoshow" )
+        show_response = self._get( "pages/%s" % response_json['id'] )
+        self._assert_status_code_is( show_response, 200 )
+        show_json = show_response.json()
+        self._assert_has_keys( show_json, "slug", "title", "id" )
+        self.assertEquals( show_json["slug"], "pagetoshow" )
+        self.assertEquals( show_json["title"], "MY PAGE" )
+        self.assertEquals( show_json["content"], "<p>Page!</p>" )
+
+    def test_403_on_unowner_show( self ):
+        response_json = self._create_valid_page_as( "others_page_show@bx.psu.edu", "otherspageshow" )
+        show_response = self._get( "pages/%s" % response_json['id'] )
+        self._assert_status_code_is( show_response, 403 )
+        self._assert_error_code_is( show_response, error_codes.USER_DOES_NOT_OWN_ITEM )
+
+    def _users_index_has_page_with_id( self, id ):
+        index_response = self._get( "pages" )
+        self._assert_status_code_is( index_response, 200 )
+        pages = index_response.json()
+        return id in map( itemgetter( "id" ), pages )