Lynn Rees avatar Lynn Rees committed 2753387

[svn]

Comments (0)

Files changed (3)

branches/0.6/trunk/setup.py

       author='L. C. Rees',
       author_email='lcrees@gmail.com',
       license='BSD',
-      py_modules=['urlrelay'],
+      py_modules=['urlrelay'],
+      test_suite='test_urlrelay',
       packages = [],
       zip_safe = True,
       keywords='WSGI URL dispatch relay route middleware web HTTP',

branches/0.6/trunk/test_urlrelay.py

+import re
+import urlrelay
+import unittest
+
+def dummy_sr(status, headers, exc_info=None):
+    pass
+
+@urlrelay.url(r'^/$')
+def index(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    return ['index']
+
+@urlrelay.url(r'^/handle$', 'GET')
+def get_handle(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    return ['GET']
+
+@urlrelay.url(r'^/handle$', 'POST')
+def post_handle(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    return ['POST']
+
+@urlrelay.url(r'^/handle$', 'PUT')
+def put_handle(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    return ['PUT']    
+
+@urlrelay.url(r'^/argtest/(?P<kwarg1>\w+)/(?P<kwarg2>\w+)/(\w+)/(\w+)$')
+def args(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    args, kwargs = environ['wsgiorg.routing_args']
+    return [' '.join([args[0], args[1], kwargs['kwarg1'], kwargs['kwarg2']])]
+
+@urlrelay.url(r'^/methtest/(?P<kwarg1>\w+)/(?P<kwarg2>\w+)/(\w+)/(\w+)$', 'GET')
+def meth_get(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    args, kwargs = environ['wsgiorg.routing_args']
+    return [' '.join(['GET', args[0], args[1], kwargs['kwarg1'], kwargs['kwarg2']])] 
+
+@urlrelay.url(r'^/methtest/(?P<kwarg1>\w+)/(?P<kwarg2>\w+)/(\w+)/(\w+)$', 'POST')
+def meth_get(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    args, kwargs = environ['wsgiorg.routing_args']
+    return [' '.join(['POST', args[0], args[1], kwargs['kwarg1'], kwargs['kwarg2']])] 
+
+@urlrelay.url(r'^/methtest/(?P<kwarg1>\w+)/(?P<kwarg2>\w+)/(\w+)/(\w+)$', 'PUT')
+def meth_get(environ, start_response):
+    start_response('200 Ok', [('Content-Type', 'text/plain')])
+    args, kwargs = environ['wsgiorg.routing_args']
+    return [' '.join(['PUT', args[0], args[1], kwargs['kwarg1'], kwargs['kwarg2']])]
+
+urlrelay.register('^/notfound$', 'urlrelay._handler')
+urlrelay.register('^/notfound2$', 'urlrelay._handler', 'GET')
+urlrelay.register('^/notfound2$', 'urlrelay._handler', 'PUT')
+urlrelay.register('^/notfound2$', 'urlrelay._handler', 'POST')
+urlrelay.register('^/notfound4$', '_handler')
+urlrelay.register('^/notfound3$', '_handler', 'GET')
+urlrelay.register('^/notfound3$', '_handler', 'PUT')
+urlrelay.register('^/notfound3$', '_handler', 'POST')
+
+class UrlRelayTest(unittest.TestCase):      
+
+    def test_inmem_root(self):
+        '''Checks simple url.'''
+        environ = {'PATH_INFO':'/', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'index')
+
+    def test_inmem_method_get(self):
+        '''Checks url + get method.'''
+        environ = {'PATH_INFO':'/handle', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'GET')
+
+    def test_inmem_method_post(self):
+        '''Checks url + post method.'''
+        environ = {'PATH_INFO':'/handle', 'REQUEST_METHOD':'POST'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'POST')        
+
+    def test_inmem_method_put(self):
+        '''Checks url + put method.'''
+        environ = {'PATH_INFO':'/handle', 'REQUEST_METHOD':'PUT'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'PUT')
+
+    def test_inmem_argextract(self):
+        '''Checks arg/kwarg extraction from URL.'''
+        environ = {'PATH_INFO':'/argtest/kwarg1/kwarg2/arg1/arg2',
+            'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'arg1 arg2 kwarg1 kwarg2')
+
+    def test_inmem_method_extract_get(self):
+        '''Checks arg/kwarg extraction from URL + get method.'''
+        environ = {'PATH_INFO':'/methtest/kwarg1/kwarg2/arg1/arg2',
+            'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'GET arg1 arg2 kwarg1 kwarg2')
+
+    def test_inmem_method_extract_post(self):
+        '''Checks arg/kwarg extraction from URL + post method.'''
+        environ = {'PATH_INFO':'/methtest/kwarg1/kwarg2/arg1/arg2',
+            'REQUEST_METHOD':'POST'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'POST arg1 arg2 kwarg1 kwarg2')
+
+    def test_inmem_method_extract_put(self):
+        '''Checks arg/kwarg extraction from URL + put method.'''
+        environ = {'PATH_INFO':'/methtest/kwarg1/kwarg2/arg1/arg2',
+            'REQUEST_METHOD':'PUT'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'PUT arg1 arg2 kwarg1 kwarg2')
+
+    def test_offdisk_method(self):
+        '''Checks loading handler off of a disk.'''
+        environ = {'PATH_INFO':'/notfound',
+            'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound was not found on this server.')
+
+    def test_offdisk_method_get(self):
+        '''Checks loading handler off of a disk + get method.'''
+        environ = {'PATH_INFO':'/notfound2',
+            'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound2 was not found on this server.')
+
+    def test_offdisk_method_put(self):
+        '''Checks loading handler off of a disk + put method.'''
+        environ = {'PATH_INFO':'/notfound2',
+            'REQUEST_METHOD':'PUT'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound2 was not found on this server.')
+
+    def test_offdisk_method_post(self):
+        '''Checks loading handler off of a disk + post method.'''
+        environ = {'PATH_INFO':'/notfound2',
+            'REQUEST_METHOD':'POST'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound2 was not found on this server.')         
+
+    def test_offdisk_method_modpath(self):
+        '''Checks loading handler off of a disk.'''
+        environ = {'PATH_INFO':'/notfound4',
+            'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(modpath='urlrelay')(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound4 was not found on this server.')
+
+    def test_offdisk_method_get_modpath(self):
+        '''Checks loading handler off of a disk + get method.'''
+        environ = {'PATH_INFO':'/notfound3',
+            'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(modpath='urlrelay')(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound3 was not found on this server.')
+
+    def test_offdisk_method_put_modpath(self):
+        '''Checks loading handler off of a disk + put method.'''
+        environ = {'PATH_INFO':'/notfound3',
+            'REQUEST_METHOD':'PUT'}
+        result = urlrelay.URLRelay(modpath='urlrelay')(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound3 was not found on this server.')
+
+    def test_offdisk_method_post_modpath(self):
+        '''Checks loading handler off of a disk + post method.'''
+        environ = {'PATH_INFO':'/notfound3',
+            'REQUEST_METHOD':'POST'}
+        result = urlrelay.URLRelay(modpath='urlrelay')(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /notfound3 was not found on this server.')
+
+    def test_inmem_defaultapp(self):
+        '''Checks using default app url.'''
+        environ = {'PATH_INFO':'/plt', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(default=index)(environ, dummy_sr)
+        self.assertEqual(result[0], 'index')        
+
+    def test_inmem_defaultapp_args(self):
+        '''Checks arg/kwarg extraction from URL + put method.'''
+        environ = {'PATH_INFO':'/methkwarg1', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(default=args, kwargs={'kwarg1':'kwarg1',
+            'kwarg2':'kwarg2'}, args=('arg1','arg2'))(environ, dummy_sr)
+        self.assertEqual(result[0], 'arg1 arg2 kwarg1 kwarg2')
+
+    def test_inmem_defaultapp(self):
+        '''Checks using default app url.'''
+        environ = {'PATH_INFO':'/plt', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(default=index)(environ, dummy_sr)
+        self.assertEqual(result[0], 'index')        
+
+    def test_ondisk_defaultapp(self):
+        '''Checks arg/kwarg extraction from URL + put method.'''
+        environ = {'PATH_INFO':'/methkwarg1', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(default='urlrelay._handler')(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /methkwarg1 was not found on this server.')
+
+    def test_ondisk_defaultapp_modpath(self):
+        '''Checks arg/kwarg extraction from URL + put method.'''
+        environ = {'PATH_INFO':'/methkwarg1', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(default='_handler', modpath='urlrelay')(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /methkwarg1 was not found on this server.')
+
+    def test_handler_override(self):
+        '''Checks overriding the default 404 handler.'''
+        def handle(environ, start_response):                
+            start_response('404 Not Found', [('content-type', 'text/plain')])
+            return ['404']
+        environ = {'PATH_INFO':'/methkwarg1', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay(handler=handle)(environ, dummy_sr)
+        self.assertEqual(result[0], '404')
+
+    def test_notfound(self):
+        '''Checks that default 404 handler responds.'''
+        environ = {'PATH_INFO':'/methkwarg1', 'REQUEST_METHOD':'GET'}
+        result = urlrelay.URLRelay()(environ, dummy_sr)
+        self.assertEqual(result[0], 'Requested URL /methkwarg1 was not found on this server.')
+
+    def test_nonregistry_paths_index(self):
+        '''Checks use of non-global path registry.'''
+        environ = {'PATH_INFO':'/', 'REQUEST_METHOD':'GET'}        
+        tpaths = ((r'^/$', index), (r'^/handle$', {'GET':get_handle,'POST':post_handle,
+            'PUT':put_handle}))
+        result = urlrelay.URLRelay(paths=tpaths)(environ, dummy_sr)
+        self.assertEqual(result[0], 'index')
+
+    def test_nonregistry_paths_get(self):
+        '''Checks use of non-global path registry.'''
+        environ = {'PATH_INFO':'/handle', 'REQUEST_METHOD':'GET'}        
+        tpaths = ((r'^/$', index), (r'^/handle$', {'GET':get_handle,'POST':post_handle,
+            'PUT':put_handle}))
+        result = urlrelay.URLRelay(paths=tpaths)(environ, dummy_sr)
+        self.assertEqual(result[0], 'GET')
+
+    def test_nonregistry_paths_post(self):
+        '''Checks use of non-global path registry.'''
+        environ = {'PATH_INFO':'/handle', 'REQUEST_METHOD':'POST'}        
+        tpaths = ((r'^/$', index), (r'^/handle$', {'GET':get_handle,'POST':post_handle,
+            'PUT':put_handle}))
+        result = urlrelay.URLRelay(paths=tpaths)(environ, dummy_sr)
+        self.assertEqual(result[0], 'POST')
+
+    def test_nonregistry_paths_put(self):
+        '''Checks use of non-global path registry.'''
+        environ = {'PATH_INFO':'/handle', 'REQUEST_METHOD':'PUT'}        
+        tpaths = ((r'^/$', index), (r'^/handle$', {'GET':get_handle,'POST':post_handle,
+            'PUT':put_handle}))
+        result = urlrelay.URLRelay(paths=tpaths)(environ, dummy_sr)
+        self.assertEqual(result[0], 'PUT')
+
+    def test_noncallable_in_registry(self):
+        '''Checks that non callables or module strings are not allowed in
+        non-global path registry.'''
+        environ = {'PATH_INFO':'/', 'REQUEST_METHOD':'PUT'}
+        def tempfunc(): 
+            test = urlrelay.URLRelay(paths=((r'^/$', []),))
+            test(environ, dummy_sr)
+        self.assertRaises(AssertionError, tempfunc)           
+        
+
+if __name__ == '__main__': unittest.main()
+

branches/0.6/trunk/urlrelay.py

 __revision__ = '0.6'
 
 import re
-import sys
 
 __all__ = ['URLRelay', 'url', 'register']
 
+
+class _Registry(object):
+
+    '''Maintains order of URL preference while updating the central URL path
+    registry.'''
+
+    _register = list()
+
+    def __iter__(self):
+        '''Iterator for registry.'''
+        return iter(self._register)
+
+    def add(self, pattern, mapping):
+        '''Add tuple to registry.'''
+        self._register.append((pattern, mapping))
+
+    def get(self):
+        '''Returns current registry.'''
+        return tuple(self._register)
+
+
 # URL registry and cache
-_reg, _cache = dict(), dict()
+_reg, _cache = _Registry(), dict()
 
 def _handler(environ, start_response):
     '''Default HTTP 404 handler.'''
     @param method HTTP method (default: None)
     '''
     if method is None:
-        _reg[pattern] = application
+        _reg.add(pattern, application)
     # Handle URL/method combinations
     else:
-        methdict = _reg.get(pattern, {})
-        methdict[method] = application
-        _reg[pattern] = methdict
+        # Update existing registry entry
+        for entry in _reg:
+            if entry[0] == pattern:
+                entry[1][method] = application
+                return
+        # Add new registry entry
+        _reg.add(pattern, {method:application})
 
 def url(pattern, method=None):
     '''Decorator for registering a path pattern /application pair.
     def decorator(application):
         register(pattern, application, method)
         return application
-    return decorator      
+    return decorator
 
 
 class URLRelay(object):
     def __init__(self, **k):
         # Add any iterable of pairs consisting of a path pattern and either a
         # callback name or a dictionary of HTTP method/callback names
-        if 'paths' in k: _reg.update(dict((u[0], u[1]) for u in k.get('urls')))
-        self.paths = tuple((re.compile(p), _reg[p]) for p in _reg)
+        self.paths = tuple((re.compile(u), v) for u, v in k.get('paths', _reg.get()))
         # Shortcut for full module search path
         self.modpath = k.get('modpath', '')
         # 404 handler
         self.akey = k.get('akey', 'wsgize.args')
         # URL <-> callable mapping Cache 
         self.cache = k.get('cache', _cache)
-        # Custom additions to PYTHONPATH
-        paths = k.get('paths', [])
-        for path in paths: sys.path.append(path)
 
     def __call__(self, env, start_response):
         try:
                 assert hasattr(app, '__call__')
                 # Extract any positional or keywords arguments in the path
                 args, kwargs = search.groups(), search.groupdict()
+                # Eliminate keyword values in args
+                args = tuple(i for i in args if i not in kwargs.itervalues())                
                 # Cache callable, positional and keyword arguments
                 self.cache[key] = (app, args, kwargs)                
                 return app, args, kwargs
             # Return defaults if no matching path and default app is set
             if self.default is not None:
-                return self.default, self.args, self.kwargs
+                default = self.default
+                if isinstance(default, basestring):
+                    default = self.getapp(default)
+                return default, self.args, self.kwargs
             # Raise ImportError
             else:
                 raise ImportError()
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.