[Checkins] SVN: Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/ added tests of mapply and openroot
Shane Hathaway
shane at hathawaymix.org
Wed Feb 18 00:56:39 EST 2009
Log message for revision 96668:
added tests of mapply and openroot
Changed:
U Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/openroot.py
A Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/mapply.txt
A Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/openroot.txt
U Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/tests.py
-=-
Modified: Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/openroot.py
===================================================================
--- Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/openroot.py 2009-02-18 05:49:26 UTC (rev 96667)
+++ Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/openroot.py 2009-02-18 05:56:38 UTC (rev 96668)
@@ -14,6 +14,7 @@
from zope.component import getUtility
from zope.interface import implements
+from zope.interface import Interface
from zope.publisher.interfaces import IWSGIApplication
from zope.security.checker import ProxyFactory
@@ -46,9 +47,9 @@
# then we should get the app controller rather than look
# in the database.
if self.app_controller_name in request.traversal_stack:
- root = getUtility(name=self.app_controller_name)
+ root = getUtility(Interface, name=self.app_controller_name)
request.traversed = [(self.app_controller_name, root)]
- return self.app(environ, start_response)
+ return self.next_app(environ, start_response)
# Open the database.
conn = self.database.open()
Added: Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/mapply.txt
===================================================================
--- Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/mapply.txt (rev 0)
+++ Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/mapply.txt 2009-02-18 05:56:38 UTC (rev 96668)
@@ -0,0 +1,113 @@
+
+Tests of the Caller app
+-----------------------
+
+The Caller app is the last stage in a pipeline. It calls the traversed object
+and sets the response according to what the function returned.
+
+ >>> class TestRequest(object):
+ ... def get(self, key, default=None):
+ ... if key == 'a':
+ ... return 6
+ ... elif key == 'b':
+ ... return 8
+ ... else:
+ ... return default
+ >>> class TestResponse(object):
+ ... def getStatusString(self):
+ ... return '200 Ok'
+ ... def getHeaders(self):
+ ... return [('Content-Type', 'text/plain')]
+ ... def setResult(self, data):
+ ... self.data = data
+ ... def consumeBodyIter(self):
+ ... return [self.data]
+ >>> def compute(a, b, c=4):
+ ... return '%d%d%d' % (a, b, c)
+ >>> r = TestRequest()
+ >>> r.traversed = [('obj', compute)]
+ >>> r.response = TestResponse()
+ >>> from zope.pipeline.apps.mapply import Caller
+ >>> app = Caller()
+ >>> app
+ Caller()
+ >>> env = {'zope.request': r}
+ >>> got_headers = []
+ >>> got_status = []
+ >>> def start_response(status, headers, exc_info=None):
+ ... got_status[:] = [status]
+ ... got_headers[:] = list(headers)
+ >>> app(env, start_response)
+ ['684']
+ >>> got_status
+ ['200 Ok']
+ >>> got_headers
+ [('Content-Type', 'text/plain')]
+
+
+Tests of ``mapply()``
+---------------------
+
+The ``mapply`` function calls a function with only the parameters expected
+by the function.
+
+ >>> from zope.pipeline.apps.mapply import mapply
+ >>> def compute(a, b, c=4):
+ ... return '%d%d%d' % (a, b, c)
+ >>> values = {'a':2, 'b':3, 'c':5, 'd':7}
+ >>> mapply(compute, (), values)
+ '235'
+ >>> mapply(compute, (7,), values)
+ '735'
+
+The ``mapply`` function can call a method of an instance.
+
+ >>> class c(object):
+ ... a = 3
+ ... def __call__(self, b, c=4):
+ ... return '%d%d%d' % (self.a, b, c)
+ ... compute = __call__
+ >>> cc = c()
+ >>> mapply(cc, (), values)
+ '335'
+ >>> del values['c']
+ >>> mapply(cc.compute, (), values)
+ '334'
+
+It can also use classic classes.
+
+ >>> class c2:
+ ... """This is a classic class"""
+ >>> c2inst = c2()
+ >>> c2inst.__call__ = cc
+ >>> mapply(c2inst, (), values)
+ '334'
+
+It can call functions that require no parameters.
+
+ >>> def callme():
+ ... return 'no'
+ >>> mapply(callme)
+ 'no'
+
+It refuses to create class instances directly.
+
+ >>> mapply(c2)
+ Traceback (most recent call last):
+ ...
+ TypeError: mapply() can not call class constructors
+
+You have to provide enough parameters.
+
+ >>> mapply(compute, (), {'a': 4})
+ Traceback (most recent call last):
+ ...
+ TypeError: Missing argument to compute(): b
+
+It can provide the request object for functions that need it.
+
+ >>> request = TestRequest()
+ >>> def compute(REQUEST):
+ ... return '%d%d' % (REQUEST.get('a'), REQUEST.get('b'))
+ >>> mapply(compute, (), request)
+ '68'
Added: Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/openroot.txt
===================================================================
--- Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/openroot.txt (rev 0)
+++ Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/openroot.txt 2009-02-18 05:56:38 UTC (rev 96668)
@@ -0,0 +1,77 @@
+
+RootOpener Tests
+----------------
+
+The RootOpener application opens a request-specific connection to the
+root database, calls the rest of the pipeline, then closes the connection
+on the way out.
+
+Set up the database.
+
+ >>> from ZODB.DB import DB
+ >>> from ZODB.DemoStorage import DemoStorage
+ >>> import transaction
+ >>> from persistent.mapping import PersistentMapping
+ >>> db = DB(DemoStorage())
+ >>> conn = db.open()
+ >>> conn.root()['Application'] = PersistentMapping({'x': 1})
+ >>> transaction.commit()
+ >>> conn.close()
+
+Create the RootOpener app around a test app.
+
+ >>> from zope.pipeline.apps.openroot import RootOpener
+ >>> def my_app(environ, start_response):
+ ... status = '200 OK'
+ ... response_headers = [('Content-type','text/plain')]
+ ... start_response(status, response_headers)
+ ... request = environ['zope.request']
+ ... return [repr(request.traversed)]
+ >>> app = RootOpener(my_app, db)
+
+Call the app.
+
+ >>> class TestRequest(object):
+ ... def __init__(self):
+ ... self.traversal_stack = []
+ ... self.traversed = []
+ ... self.annotations = {}
+ >>> request = TestRequest()
+ >>> environ = {'zope.request': request}
+ >>> def start_response(status, headers, exc_info=None):
+ ... pass
+ >>> app(environ, start_response)
+ ["[('Application', {'x': 1})]"]
+
+Try to call the app with no Application at the root.
+
+ >>> conn = db.open()
+ >>> del conn.root()['Application']
+ >>> transaction.commit()
+ >>> conn.close()
+ >>> app(environ, start_response)
+ Traceback (most recent call last):
+ ...
+ SystemError: Zope Application Not Found
+
+Clean up.
+
+ >>> db.close()
+
+
+Tests of ++etc++process
+-----------------------
+
+If the traversal stack contains ++etc++process, the root opener uses
+a utility by that name as the root instead of opening the database.
+
+ >>> from zope.interface import Interface, implements
+ >>> from zope.component import provideUtility
+ >>> class ProcessRoot(object):
+ ... pass
+ >>> provideUtility(ProcessRoot, name='++etc++process', provides=Interface)
+ >>> request = TestRequest()
+ >>> request.traversal_stack.append('++etc++process')
+ >>> environ = {'zope.request': request}
+ >>> app(environ, start_response)
+ ["[('++etc++process', <class 'ProcessRoot'>)]"]
Modified: Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/tests.py
===================================================================
--- Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/tests.py 2009-02-18 05:49:26 UTC (rev 96667)
+++ Sandbox/shane/republish/zope.pipeline/src/zope/pipeline/apps/tests/tests.py 2009-02-18 05:56:38 UTC (rev 96668)
@@ -29,9 +29,12 @@
def test_suite():
return unittest.TestSuite([
- doctest.DocFileSuite('retry.txt', optionflags=flags),
+ doctest.DocFileSuite('mapply.txt', optionflags=flags),
+ doctest.DocFileSuite('openroot.txt', optionflags=flags,
+ setUp=setUp, tearDown=tearDown),
doctest.DocFileSuite('requestsetup.txt', optionflags=flags,
setUp=setUp, tearDown=tearDown),
+ doctest.DocFileSuite('retry.txt', optionflags=flags),
])
if __name__ == '__main__':
More information about the Checkins
mailing list