Changeset 774

Timestamp:

11/01/05 01:23:34

Author:

fumanchu

Message:

Fix for #357 (Pythonic access to Accept-* request headers). New cptools.getAccept function, which returns a list of AcceptValue? objects (sorted in descending priority). GzipFilter rewritten to use it.

Files:

• trunk/cherrypy/lib/cptools.py (modified) (1 diff)
• trunk/cherrypy/lib/filter/gzipfilter.py (modified) (1 diff)
• trunk/cherrypy/test/test_core.py (modified) (3 diffs)
• trunk/cherrypy/test/test_gzip_filter.py (modified) (1 diff)

trunk/cherrypy/lib/cptools.py

@@ -186 +186 @@
-        raise cherrypy.HTTPError(416, message)
-    
+    return result
+
+
+class AcceptValue(object):
+    """A value (with parameters) from an Accept-* request header."""
+    
+    def __init__(self, value, params=None):
+        self.value = value
+        if params is None:
+            params = {}
+        self.params = params
+    
+    def qvalue(self):
+        val = self.params.get("q", "1")
+        if isinstance(val, AcceptValue):
+            val = val.value
+        return float(val)
+    qvalue = property(qvalue, doc="The qvalue, or priority, of this value.")
+    
+    def __str__(self):
+        p = [";%s=%s" % (k, v) for k, v in self.params.iteritems()]
+        return "%s%s" % (self.value, "".join(p))
+    
+    def __cmp__(self, other):
+        # If you sort a list of AcceptValue objects, they will be listed in
+        # priority order; that is, the most preferred value will be first.
+        diff = cmp(other.qvalue, self.qvalue)
+        if diff == 0:
+            diff = cmp(str(other), str(self))
+        return diff
+
+
+def getAccept(headername='Accept'):
+    """Return a list of AcceptValues from an Accept header, or None."""
+    
+    r = cherrypy.request.headerMap.get(headername)
+    if not r:
+        return None
+    
+    result = []
+    for capability in r.split(","):
+        # The first "q" parameter (if any) separates the initial
+        # parameter(s) (if any) from the accept-params.
+        atoms = re.split(r'; *q *=', capability, 1)
+        capvalue = atoms.pop(0).strip()
+        if atoms:
+            qvalue = atoms[0].strip()
+            if headername == 'Accept':
+                # The qvalue for an Accept header can have extensions.
+                atoms = [x.strip() for x in qvalue.split(";")]
+                qvalue = atoms.pop(0).strip()
+                ext = {}
+                for atom in atoms:
+                    atom = atom.split("=", 1)
+                    key = atom.pop(0).strip()
+                    if atom:
+                        val = atom[0].strip()
+                    else:
+                        val = ""
+                    ext[key] = val
+                qvalue = AcceptValue(qvalue, ext)
+            params = {"q": qvalue}
+        else:
+            params = {}
+        
+        if headername == 'Accept':
+            # The media-range may have parameters (before the qvalue).
+            atoms = [x.strip() for x in capvalue.split(";")]
+            capvalue = atoms.pop(0).strip()
+            for atom in atoms:
+                atom = atom.split("=", 1)
+                key = atom.pop(0).strip()
+                if atom:
+                    val = atom[0].strip()
+                else:
+                    val = ""
+                params[key] = val
+        
+        result.append(AcceptValue(capvalue, params))
+    
+    result.sort()
-    return result
-

trunk/cherrypy/lib/filter/gzipfilter.py

@@ -17 +17 @@
-            return
-        
-
+
+
-            # Response body is empty (might be a 304 for instance)
-            return
-        
-
-
-
-
-
+
-            # Return a generator that compresses the page
+            response.headerMap['Content-Encoding'] = 'gzip'
-            level = cherrypy.config.get('gzipFilter.compresslevel', 9)
-
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
-    
-    def write_gzip_header(self):

trunk/cherrypy/test/test_core.py

@@ -175 +175 @@
-        path = os.path.join(os.getcwd(), os.path.dirname(__file__))
-        return cptools.serveFile(os.path.join(path, "static/index.html"))
+
+
+class Accept(Test):
+    
+    def get_accept(self, headername):
+        return "\n".join([str(x) for x in cptools.getAccept(headername)])
-
-
@@ -572 +578 @@
-               "[Errno 2] No such file or directory: 'nonexistent.html'")
-        self.assertInBody(msg)
-
-    
-    def testRanges(self):
@@ -619 +624 @@
-        self.assertHeader("Content-Range", "bytes */14")
-    
+    def testAccept(self):
+        h = [('Accept', 'audio/*; q=0.2, audio/basic')]
+        self.getPage("/accept/get_accept?headername=Accept", h)
+        self.assertStatus("200 OK")
+        self.assertBody("audio/basic\n"
+                        "audio/*;q=0.2")
+        
+        h = [('Accept', 'text/plain; q=0.5, text/html, text/x-dvi; q=0.8, text/x-c')]
+        self.getPage("/accept/get_accept?headername=Accept", h)
+        self.assertStatus("200 OK")
+        self.assertBody("text/x-c\n"
+                        "text/html\n"
+                        "text/x-dvi;q=0.8\n"
+                        "text/plain;q=0.5")
+        
+        # Test that more specific media ranges get priority.
+        # Note that the highest priority will be first in the list.
+        h = [('Accept', 'text/*, text/html, text/html;level=1, */*')]
+        self.getPage("/accept/get_accept?headername=Accept", h)
+        self.assertStatus("200 OK")
+        self.assertBody("text/html;level=1\n"
+                        "text/html\n"
+                        "text/*\n"
+                        "*/*")
+        
+        # Test Accept-Charset
+        h = [('Accept-Charset', 'iso-8859-5, unicode-1-1;q=0.8')]
+        self.getPage("/accept/get_accept?headername=Accept-Charset", h)
+        self.assertStatus("200 OK")
+        self.assertBody("iso-8859-5\n"
+                        "unicode-1-1;q=0.8")
+        
+        # Test Accept-Encoding
+        h = [('Accept-Encoding', 'gzip;q=1.0, identity; q=0.5, *;q=0')]
+        self.getPage("/accept/get_accept?headername=Accept-Encoding", h)
+        self.assertStatus("200 OK")
+        self.assertBody("gzip;q=1.0\n"
+                        "identity;q=0.5\n"
+                        "*;q=0")
+        
+        # Test Accept-Language
+        h = [('Accept-Language', 'da, en-gb;q=0.8, en;q=0.7')]
+        self.getPage("/accept/get_accept?headername=Accept-Language", h)
+        self.assertStatus("200 OK")
+        self.assertBody("da\n"
+                        "en-gb;q=0.8\n"
+                        "en;q=0.7")
+    
-    def testHeaderCaseSensitivity(self):
-        # Tests that each header only appears once, regardless of case.

trunk/cherrypy/test/test_gzip_filter.py

@@ -47 +47 @@
-        self.assertInBody(zbuf.getvalue()[:3])
-        
+        # Test when gzip is denied.
+        self.getPage('/', headers=[("Accept-Encoding", "identity")])
+        self.assertBody("Hello, world")
+        self.getPage('/', headers=[("Accept-Encoding", "gzip;q=0")])
+        self.assertBody("Hello, world")
+        self.getPage('/', headers=[("Accept-Encoding", "*;q=0")])
+        self.assertStatus("406 Not Acceptable")
+        self.assertErrorPage(406, "identity, gzip")
+        
-        # Test for ticket #147
-        helper.webtest.ignored_exceptions.append(IndexError)