import cherrypy
from cherrypy.test import helper
script_names = ['', '/foo', '/users/fred/blog', '/corp/blog']
[docs]def setup_server():
class SubSubRoot:
@cherrypy.expose
def index(self):
return 'SubSubRoot index'
@cherrypy.expose
def default(self, *args):
return 'SubSubRoot default'
@cherrypy.expose
def handler(self):
return 'SubSubRoot handler'
@cherrypy.expose
def dispatch(self):
return 'SubSubRoot dispatch'
subsubnodes = {
'1': SubSubRoot(),
'2': SubSubRoot(),
}
class SubRoot:
@cherrypy.expose
def index(self):
return 'SubRoot index'
@cherrypy.expose
def default(self, *args):
return 'SubRoot %s' % (args,)
@cherrypy.expose
def handler(self):
return 'SubRoot handler'
def _cp_dispatch(self, vpath):
return subsubnodes.get(vpath[0], None)
subnodes = {
'1': SubRoot(),
'2': SubRoot(),
}
class Root:
@cherrypy.expose
def index(self):
return 'index'
@cherrypy.expose
def default(self, *args):
return 'default %s' % (args,)
@cherrypy.expose
def handler(self):
return 'handler'
def _cp_dispatch(self, vpath):
return subnodes.get(vpath[0])
# -------------------------------------------------------------------------
# DynamicNodeAndMethodDispatcher example.
# This example exposes a fairly naive HTTP api
class User(object):
def __init__(self, id, name):
self.id = id
self.name = name
def __unicode__(self):
return str(self.name)
def __str__(self):
return str(self.name)
user_lookup = {
1: User(1, 'foo'),
2: User(2, 'bar'),
}
def make_user(name, id=None):
if not id:
id = max(*list(user_lookup.keys())) + 1
user_lookup[id] = User(id, name)
return id
@cherrypy.expose
class UserContainerNode(object):
def POST(self, name):
"""Allow the creation of a new Object."""
return 'POST %d' % make_user(name)
def GET(self):
return str(sorted(user_lookup.keys()))
def dynamic_dispatch(self, vpath):
try:
id = int(vpath[0])
except (ValueError, IndexError):
return None
return UserInstanceNode(id)
@cherrypy.expose
class UserInstanceNode(object):
def __init__(self, id):
self.id = id
self.user = user_lookup.get(id, None)
# For all but PUT methods there MUST be a valid user identified
# by self.id
if not self.user and cherrypy.request.method != 'PUT':
raise cherrypy.HTTPError(404)
def GET(self, *args, **kwargs):
"""Return the appropriate representation of the instance."""
return str(self.user)
def POST(self, name):
"""Update the fields of the user instance."""
self.user.name = name
return 'POST %d' % self.user.id
def PUT(self, name):
"""
Create a new user with the specified id, or edit it if it already
exists
"""
if self.user:
# Edit the current user
self.user.name = name
return 'PUT %d' % self.user.id
else:
# Make a new user with said attributes.
return 'PUT %d' % make_user(name, self.id)
def DELETE(self):
"""Delete the user specified at the id."""
id = self.user.id
del user_lookup[self.user.id]
del self.user
return 'DELETE %d' % id
class ABHandler:
class CustomDispatch:
@cherrypy.expose
def index(self, a, b):
return 'custom'
def _cp_dispatch(self, vpath):
"""Make sure that if we don't pop anything from vpath,
processing still works.
"""
return self.CustomDispatch()
@cherrypy.expose
def index(self, a, b=None):
body = ['a:' + str(a)]
if b is not None:
body.append(',b:' + str(b))
return ''.join(body)
@cherrypy.expose
def delete(self, a, b):
return 'deleting ' + str(a) + ' and ' + str(b)
class IndexOnly:
def _cp_dispatch(self, vpath):
"""Make sure that popping ALL of vpath still shows the index
handler.
"""
while vpath:
vpath.pop()
return self
@cherrypy.expose
def index(self):
return 'IndexOnly index'
class DecoratedPopArgs:
"""Test _cp_dispatch with @cherrypy.popargs."""
@cherrypy.expose
def index(self):
return 'no params'
@cherrypy.expose
def hi(self):
return "hi was not interpreted as 'a' param"
DecoratedPopArgs = cherrypy.popargs(
'a', 'b', handler=ABHandler())(DecoratedPopArgs)
class NonDecoratedPopArgs:
"""Test _cp_dispatch = cherrypy.popargs()"""
_cp_dispatch = cherrypy.popargs('a')
@cherrypy.expose
def index(self, a):
return 'index: ' + str(a)
class ParameterizedHandler:
"""Special handler created for each request."""
def __init__(self, a):
self.a = a
@cherrypy.expose
def index(self):
if 'a' in cherrypy.request.params:
raise Exception(
'Parameterized handler argument ended up in '
'request.params')
return self.a
class ParameterizedPopArgs:
"""Test cherrypy.popargs() with a function call handler."""
ParameterizedPopArgs = cherrypy.popargs(
'a', handler=ParameterizedHandler)(ParameterizedPopArgs)
Root.decorated = DecoratedPopArgs()
Root.undecorated = NonDecoratedPopArgs()
Root.index_only = IndexOnly()
Root.parameter_test = ParameterizedPopArgs()
Root.users = UserContainerNode()
md = cherrypy.dispatch.MethodDispatcher('dynamic_dispatch')
for url in script_names:
conf = {
'/': {
'user': (url or '/').split('/')[-2],
},
'/users': {
'request.dispatch': md
},
}
cherrypy.tree.mount(Root(), url, conf)
[docs]class DynamicObjectMappingTest(helper.CPWebCase):
setup_server = staticmethod(setup_server)
[docs] def testObjectMapping(self):
for url in script_names:
self.script_name = url
self.getPage('/')
self.assertBody('index')
self.getPage('/handler')
self.assertBody('handler')
# Dynamic dispatch will succeed here for the subnodes
# so the subroot gets called
self.getPage('/1/')
self.assertBody('SubRoot index')
self.getPage('/2/')
self.assertBody('SubRoot index')
self.getPage('/1/handler')
self.assertBody('SubRoot handler')
self.getPage('/2/handler')
self.assertBody('SubRoot handler')
# Dynamic dispatch will fail here for the subnodes
# so the default gets called
self.getPage('/asdf/')
self.assertBody("default ('asdf',)")
self.getPage('/asdf/asdf')
self.assertBody("default ('asdf', 'asdf')")
self.getPage('/asdf/handler')
self.assertBody("default ('asdf', 'handler')")
# Dynamic dispatch will succeed here for the subsubnodes
# so the subsubroot gets called
self.getPage('/1/1/')
self.assertBody('SubSubRoot index')
self.getPage('/2/2/')
self.assertBody('SubSubRoot index')
self.getPage('/1/1/handler')
self.assertBody('SubSubRoot handler')
self.getPage('/2/2/handler')
self.assertBody('SubSubRoot handler')
self.getPage('/2/2/dispatch')
self.assertBody('SubSubRoot dispatch')
# The exposed dispatch will not be called as a dispatch
# method.
self.getPage('/2/2/foo/foo')
self.assertBody('SubSubRoot default')
# Dynamic dispatch will fail here for the subsubnodes
# so the SubRoot gets called
self.getPage('/1/asdf/')
self.assertBody("SubRoot ('asdf',)")
self.getPage('/1/asdf/asdf')
self.assertBody("SubRoot ('asdf', 'asdf')")
self.getPage('/1/asdf/handler')
self.assertBody("SubRoot ('asdf', 'handler')")
[docs] def testMethodDispatch(self):
# GET acts like a container
self.getPage('/users')
self.assertBody('[1, 2]')
self.assertHeader('Allow', 'GET, HEAD, POST')
# POST to the container URI allows creation
self.getPage('/users', method='POST', body='name=baz')
self.assertBody('POST 3')
self.assertHeader('Allow', 'GET, HEAD, POST')
# POST to a specific instanct URI results in a 404
# as the resource does not exit.
self.getPage('/users/5', method='POST', body='name=baz')
self.assertStatus(404)
# PUT to a specific instanct URI results in creation
self.getPage('/users/5', method='PUT', body='name=boris')
self.assertBody('PUT 5')
self.assertHeader('Allow', 'DELETE, GET, HEAD, POST, PUT')
# GET acts like a container
self.getPage('/users')
self.assertBody('[1, 2, 3, 5]')
self.assertHeader('Allow', 'GET, HEAD, POST')
test_cases = (
(1, 'foo', 'fooupdated', 'DELETE, GET, HEAD, POST, PUT'),
(2, 'bar', 'barupdated', 'DELETE, GET, HEAD, POST, PUT'),
(3, 'baz', 'bazupdated', 'DELETE, GET, HEAD, POST, PUT'),
(5, 'boris', 'borisupdated', 'DELETE, GET, HEAD, POST, PUT'),
)
for id, name, updatedname, headers in test_cases:
self.getPage('/users/%d' % id)
self.assertBody(name)
self.assertHeader('Allow', headers)
# Make sure POSTs update already existings resources
self.getPage('/users/%d' %
id, method='POST', body='name=%s' % updatedname)
self.assertBody('POST %d' % id)
self.assertHeader('Allow', headers)
# Make sure PUTs Update already existing resources.
self.getPage('/users/%d' %
id, method='PUT', body='name=%s' % updatedname)
self.assertBody('PUT %d' % id)
self.assertHeader('Allow', headers)
# Make sure DELETES Remove already existing resources.
self.getPage('/users/%d' % id, method='DELETE')
self.assertBody('DELETE %d' % id)
self.assertHeader('Allow', headers)
# GET acts like a container
self.getPage('/users')
self.assertBody('[]')
self.assertHeader('Allow', 'GET, HEAD, POST')
[docs] def testVpathDispatch(self):
self.getPage('/decorated/')
self.assertBody('no params')
self.getPage('/decorated/hi')
self.assertBody("hi was not interpreted as 'a' param")
self.getPage('/decorated/yo/')
self.assertBody('a:yo')
self.getPage('/decorated/yo/there/')
self.assertBody('a:yo,b:there')
self.getPage('/decorated/yo/there/delete')
self.assertBody('deleting yo and there')
self.getPage('/decorated/yo/there/handled_by_dispatch/')
self.assertBody('custom')
self.getPage('/undecorated/blah/')
self.assertBody('index: blah')
self.getPage('/index_only/a/b/c/d/e/f/g/')
self.assertBody('IndexOnly index')
self.getPage('/parameter_test/argument2/')
self.assertBody('argument2')
