diff --git a/lib/galaxy/webapps/galaxy/api/page_revisions.py b/lib/galaxy/webapps/galaxy/api/page_revisions.py
index 25cbbc3f1ad..56e3bded8dd 100644
--- a/lib/galaxy/webapps/galaxy/api/page_revisions.py
+++ b/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()
diff --git a/lib/galaxy/webapps/galaxy/api/pages.py b/lib/galaxy/webapps/galaxy/api/pages.py
index 92e01fcbebb..aa27c6afc09 100644
--- a/lib/galaxy/webapps/galaxy/api/pages.py
+++ b/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
diff --git a/test/base/api.py b/test/base/api.py
index d1d9a1f7583..aea87bfa811 100644
--- a/test/base/api.py
+++ b/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"
diff --git a/test/functional/api/test_page_revisions.py b/test/functional/api/test_page_revisions.py
new file mode 100644
index 00000000000..2fa91b0992a
--- /dev/null
+++ b/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="NewContent!
" )
+ 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="NewContent!
" )
+ 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="NewContent!
" )
+ 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 )
diff --git a/test/functional/api/test_pages.py b/test/functional/api/test_pages.py
new file mode 100644
index 00000000000..8b46d8b071d
--- /dev/null
+++ b/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="Page!
",
+ )
+ 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"], "Page!
" )
+
+ 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 )