Added miscellaneous environment-processing utilties, and a 'FileWrapper'

implementation, all with tests.


git-svn-id: svn://svn.eby-sarna.com/svnroot/wsgiref@247 571e12c6-e1fa-0310-aee7-ff1267fa46bd

3 files changed, 371 insertions, 2 deletions

diff --git a/src/wsgiref/tests/__init__.py b/src/wsgiref/tests/__init__.py
index 8ada027..a4eb4b5 100644
--- a/src/wsgiref/tests/__init__.py
+++ b/src/wsgiref/tests/__init__.py
@@ -2,10 +2,10 @@ from unittest import TestSuite, TestCase, makeSuite
 
 def test_suite():
 
-    #from wsgiref.tests import test_xyz
+    from wsgiref.tests import test_util
 
     tests = [
-        #test_xyz.test_suite(),
+        test_util.test_suite(),
     ]
 
     return TestSuite(tests)
diff --git a/src/wsgiref/tests/test_util.py b/src/wsgiref/tests/test_util.py
new file mode 100644
index 0000000..b43f28e
--- /dev/null
+++ b/src/wsgiref/tests/test_util.py
@@ -0,0 +1,205 @@
+from unittest import TestCase, TestSuite, makeSuite
+from wsgiref import util
+from StringIO import StringIO
+
+
+class UtilityTests(TestCase):
+
+    def checkShift(self,sn_in,pi_in,part,sn_out,pi_out):
+        env = {'SCRIPT_NAME':sn_in,'PATH_INFO':pi_in}
+        util.setup_testing_defaults(env)
+        self.assertEqual(util.shift_path_info(env),part)
+        self.assertEqual(env['PATH_INFO'],pi_out)
+        self.assertEqual(env['SCRIPT_NAME'],sn_out)
+        return env
+
+
+    def checkDefault(self, key, value, alt=None):
+        # Check defaulting when empty
+        env = {}
+        util.setup_testing_defaults(env)
+        if isinstance(value,StringIO):
+            self.failUnless(isinstance(env[key],StringIO))
+        else:
+            self.assertEqual(env[key],value)
+
+        # Check existing value
+        env = {key:alt}
+        util.setup_testing_defaults(env)
+        self.failUnless(env[key] is alt)
+
+
+
+
+
+
+
+
+
+
+
+
+    def checkCrossDefault(self,key,value,**kw):
+        util.setup_testing_defaults(kw)
+        self.assertEqual(kw[key],value)
+
+    def checkAppURI(self,uri,**kw):
+        util.setup_testing_defaults(kw)
+        self.assertEqual(util.application_uri(kw),uri)
+
+    def checkReqURI(self,uri,query=1,**kw):
+        util.setup_testing_defaults(kw)
+        self.assertEqual(util.request_uri(kw,query),uri)
+
+    def checkFW(self,text,size,match):
+        sio = StringIO(text)
+        fw = util.FileWrapper(sio,size)
+        n = 0
+        for item in match:
+            self.assertEqual(fw[n],item)
+            n+=1
+        self.assertRaises(IndexError, fw.__getitem__, n)
+
+        try:
+            iter, StopIteration
+        except NameError:
+            pass
+        else:
+            # Only test iter mode under 2.2+
+            sio = StringIO(text)
+            fw = util.FileWrapper(sio,size)
+            self.failUnless(iter(fw) is fw)
+            for item in match:
+                self.assertEqual(fw.next(),item)
+            self.assertRaises(StopIteration, fw.next)
+
+        self.failIf(sio.closed)
+        fw.close()
+        self.failUnless(sio.closed)
+
+
+
+
+    def testSimpleShifts(self):
+        self.checkShift('','/', '', '', '')
+        self.checkShift('','/x', 'x', '/x', '')
+        self.checkShift('/','', None, '/', '')
+        self.checkShift('/a','/x/y', 'x', '/a/x', '/y')
+        self.checkShift('/a','/x/',  'x', '/a/x', '/')
+
+    def testNormalizedShifts(self):
+        self.checkShift('/a/b', '/../y', '..', '/a', '/y')
+        self.checkShift('', '/../y', '..', '', '/y')
+        self.checkShift('/a/b', '//y', 'y', '/a/b/y', '')
+        self.checkShift('/a/b', '//y/', 'y', '/a/b/y', '/')
+        self.checkShift('/a/b', '/./y', 'y', '/a/b/y', '')
+        self.checkShift('/a/b', '/./y/', 'y', '/a/b/y', '/')
+        self.checkShift('/a/b', '///./..//y/.//', '..', '/a', '/y/')
+        self.checkShift('/a/b', '///', '', '/a/b', '')
+        self.checkShift('/a/b', '/.//', '', '/a/b', '')
+        self.checkShift('/a/b', '/x//', 'x', '/a/b/x', '/')
+        self.checkShift('/a/b', '/.', None, '/a/b', '')
+
+    def testDefaults(self):
+        for key, value in [
+            ('SERVER_NAME','127.0.0.1'),
+            ('SERVER_PORT', '80'),
+            ('HTTP_HOST','127.0.0.1'),
+            ('REQUEST_METHOD','GET'),
+            ('SCRIPT_NAME',''),
+            ('PATH_INFO','/'),
+            ('wsgi.version', (1,0)),
+            ('wsgi.run_once', 0),
+            ('wsgi.multithread', 0),
+            ('wsgi.multiprocess', 0),
+            ('wsgi.input', StringIO("")),
+            ('wsgi.errors', StringIO()),
+            ('wsgi.url_scheme','http'),
+        ]:
+            self.checkDefault(key,value)
+
+
+
+
+    def testCrossDefaults(self):
+        self.checkCrossDefault('HTTP_HOST',"foo.bar",SERVER_NAME="foo.bar")
+        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="on")
+        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="1")
+        self.checkCrossDefault('wsgi.url_scheme',"https",HTTPS="yes")
+        self.checkCrossDefault('wsgi.url_scheme',"http",HTTPS="foo")
+        self.checkCrossDefault('SERVER_PORT',"80",HTTPS="foo")
+        self.checkCrossDefault('SERVER_PORT',"443",HTTPS="on")
+
+    def testGuessScheme(self):
+        self.assertEqual(util.guess_scheme({}), "http")
+        self.assertEqual(util.guess_scheme({'HTTPS':"foo"}), "http")
+        self.assertEqual(util.guess_scheme({'HTTPS':"on"}), "https")
+        self.assertEqual(util.guess_scheme({'HTTPS':"yes"}), "https")
+        self.assertEqual(util.guess_scheme({'HTTPS':"1"}), "https")
+
+    def testAppURIs(self):
+        self.checkAppURI("http://127.0.0.1/")
+        self.checkAppURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam")
+        self.checkAppURI("http://spam.example.com/",
+            HTTP_HOST="spam.example.com")
+        self.checkAppURI("http://spam.example.com/",
+            SERVER_NAME="spam.example.com")
+        self.checkAppURI("http://127.0.0.1/",
+            HTTP_HOST="127.0.0.1", SERVER_NAME="spam.example.com")
+        self.checkAppURI("https://127.0.0.1/", HTTPS="on")
+        self.checkAppURI("http://127.0.0.1:8000/", SERVER_PORT="8000")
+
+    def testReqURIs(self):
+        self.checkReqURI("http://127.0.0.1/")
+        self.checkReqURI("http://127.0.0.1/spam", SCRIPT_NAME="/spam")
+        self.checkReqURI("http://127.0.0.1/spammity/spam",
+            SCRIPT_NAME="/spammity", PATH_INFO="/spam")
+        self.checkReqURI("http://127.0.0.1/spammity/spam?say=ni",
+            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni")
+        self.checkReqURI("http://127.0.0.1/spammity/spam", 0,
+            SCRIPT_NAME="/spammity", PATH_INFO="/spam",QUERY_STRING="say=ni")
+
+    def testFileWrapper(self):
+        self.checkFW("xyz"*50, 120, ["xyz"*40,"xyz"*10])
+
+TestClasses = (
+    UtilityTests,
+)
+
+def test_suite():
+    return TestSuite([makeSuite(t,'test') for t in TestClasses])
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/src/wsgiref/util.py b/src/wsgiref/util.py
new file mode 100644
index 0000000..dd6f79c
--- /dev/null
+++ b/src/wsgiref/util.py
@@ -0,0 +1,164 @@
+"""Miscellaneous WSGI-related Utilities"""
+
+import posixpath
+
+__all__ = [
+    'FileWrapper', 'guess_scheme', 'application_uri', 'request_uri',
+    'shift_path_info', 'setup_testing_defaults',
+]
+
+
+class FileWrapper:
+    """Wrapper to convert file-like objects to iterables"""
+
+    def __init__(self, filelike, blksize=8192):
+        self.filelike = filelike
+        self.blksize = blksize
+        if hasattr(filelike,'close'):
+            self.close = filelike.close
+            
+    def __getitem__(self,key):
+        data = self.filelike.read(self.blksize)
+        if data:
+            return data
+        raise IndexError
+
+    def __iter__(self):
+        return self
+        
+    def next(self):
+        data = self.filelike.read(self.blksize)
+        if data:
+            return data
+        raise StopIteration
+
+
+
+
+
+
+
+
+def guess_scheme(environ):
+    """Return a guess for whether 'wsgi.url_scheme' should be 'http' or 'https'
+    """
+    if environ.get("HTTPS") in ('yes','on','1'):
+        return 'https'
+    else:
+        return 'http'
+           
+def application_uri(environ):
+    """Return the application's base URI (no PATH_INFO or QUERY_STRING)"""
+    url = environ['wsgi.url_scheme']+'://'
+    from urllib import quote
+
+    if environ.get('HTTP_HOST'):
+        url += environ['HTTP_HOST']
+    else:
+        url += environ['SERVER_NAME']
+
+    if environ['wsgi.url_scheme'] == 'https':
+        if environ['SERVER_PORT'] != '443':
+           url += ':' + environ['SERVER_PORT']
+    else:
+        if environ['SERVER_PORT'] != '80':
+           url += ':' + environ['SERVER_PORT']
+
+    url += quote(environ.get('SCRIPT_NAME') or '/')
+    return url   
+
+def request_uri(environ,include_query=1):
+    """Return the full request URI, optionally including the query string"""
+    url = application_uri(environ)
+    from urllib import quote
+    path_info = quote(environ.get('PATH_INFO',''))
+    if not environ.get('SCRIPT_NAME'):
+        url += path_info[1:]
+    else:
+        url += path_info
+    if include_query and environ.get('QUERY_STRING'):
+        url += '?' + environ['QUERY_STRING']
+    return url
+
+def shift_path_info(environ):
+    """Shift a name from PATH_INFO to SCRIPT_NAME, returning it
+
+    If there are no remaining path segments in PATH_INFO, return None.
+    Note: 'environ' is modified in-place; use a copy if you need to keep
+    the original PATH_INFO or SCRIPT_NAME.
+    """
+
+    path_info = environ.get('PATH_INFO','')
+
+    if not path_info:
+        return None
+
+    path_parts = path_info.split('/')
+    path_parts[1:-1] = [p for p in path_parts[1:-1] if p and p<>'.']
+    name = path_parts[1]
+    del path_parts[1]
+
+    script_name = environ.get('SCRIPT_NAME','')
+    script_name = posixpath.normpath(script_name+'/'+name)
+    if script_name.endswith('/'):
+        script_name = script_name[:-1]
+
+    environ['SCRIPT_NAME'] = script_name
+    environ['PATH_INFO']   = '/'.join(path_parts)
+
+    # Special case: '/.' on PATH_INFO doesn't get stripped,
+    # because we don't strip the last element of PATH_INFO
+    # if there's only one path part left.  Instead of fixing this
+    # above, we fix it here so that PATH_INFO gets normalized to
+    # an empty string in the environ.
+    if name=='.':
+        name = None
+
+    return name
+
+
+
+
+
+
+def setup_testing_defaults(environ):
+    """Update 'environ' with trivial defaults for testing purposes
+
+    This adds various parameters required for WSGI, including HTTP_HOST,
+    SERVER_NAME, SERVER_PORT, REQUEST_METHOD, SCRIPT_NAME, PATH_INFO,
+    and all of the wsgi.* variables.  It only supplies default values,
+    and does not replace any existing settings for these variables.
+
+    This routine is intended to make it easier for unit tests of WSGI
+    servers and applications to set up dummy environments.  It should *not*
+    be used by actual WSGI servers or applications, since the data is fake!
+    """
+
+    environ.setdefault('SERVER_NAME','127.0.0.1')
+    environ.setdefault('HTTP_HOST',environ['SERVER_NAME'])
+    environ.setdefault('REQUEST_METHOD','GET')
+
+    if 'SCRIPT_NAME' not in environ and 'PATH_INFO' not in environ:
+        environ.setdefault('SCRIPT_NAME','')
+        environ.setdefault('PATH_INFO','/')
+
+    environ.setdefault('wsgi.version', (1,0))
+    environ.setdefault('wsgi.run_once', 0)
+    environ.setdefault('wsgi.multithread', 0)
+    environ.setdefault('wsgi.multiprocess', 0)
+
+    from StringIO import StringIO
+    environ.setdefault('wsgi.input', StringIO(""))
+    environ.setdefault('wsgi.errors', StringIO())
+    environ.setdefault('wsgi.url_scheme',guess_scheme(environ))
+
+    if environ['wsgi.url_scheme']=='http':
+        environ.setdefault('SERVER_PORT', '80')
+    elif environ['wsgi.url_scheme']=='https':
+        environ.setdefault('SERVER_PORT', '443')
+
+
+
+
+
+