Decompression tests.

Author: Lukasa

hyper/http11/response.py

@@ -66,9 +66,9 @@ def __init__(self, code, reason, headers, sock):
         # This 16 + MAX_WBITS nonsense is to force gzip. See this
         # Stack Overflow answer for more:
         # http://stackoverflow.com/a/2695466/1401686
-        if b'gzip' in self.headers.get('content-encoding', []):
+        if b'gzip' in self.headers.get(b'content-encoding', []):
             self._decompressobj = zlib.decompressobj(16 + zlib.MAX_WBITS)
-        elif b'deflate' in self.headers.get('content-encoding', []):
+        elif b'deflate' in self.headers.get(b'content-encoding', []):
             self._decompressobj = DeflateDecoder()
         else:
             self._decompressobj = None

test/test_http11.py

@@ -6,6 +6,7 @@
 Unit tests for hyper's HTTP/1.1 implementation.
 """
 import os
+import zlib
 
 from collections import namedtuple
 from io import BytesIO, StringIO
@@ -17,7 +18,7 @@
 from hyper.http11.response import HTTP11Response
 from hyper.http20.exceptions import ConnectionResetError
 from hyper.common.headers import HTTPHeaderMap
-from hyper.compat import bytes
+from hyper.compat import bytes, zlib_compressobj
 
 
 class TestHTTP11Connection(object):
@@ -344,6 +345,44 @@ def test_response_as_context_manager(self):
 
         assert r._sock == None
 
+    def test_response_transparently_decrypts_gzip(self):
+        d = DummySocket()
+        headers = {b'content-encoding': [b'gzip'], b'connection': [b'close']}
+        r = HTTP11Response(200, 'OK', headers, d)
+
+        c = zlib_compressobj(wbits=24)
+        body = c.compress(b'this is test data')
+        body += c.flush()
+        d._buffer = BytesIO(body)
+
+        assert r.read() == b'this is test data'
+
+    def test_response_transparently_decrypts_real_deflate(self):
+        d = DummySocket()
+        headers = {b'content-encoding': [b'deflate'], b'connection': [b'close']}
+        r = HTTP11Response(200, 'OK', headers, d)
+        c = zlib_compressobj(wbits=zlib.MAX_WBITS)
+        body = c.compress(b'this is test data')
+        body += c.flush()
+        d._buffer = BytesIO(body)
+
+        assert r.read() == b'this is test data'
+
+    def test_response_transparently_decrypts_wrong_deflate(self):
+        c = zlib_compressobj(wbits=-zlib.MAX_WBITS)
+        body = c.compress(b'this is test data')
+        body += c.flush()
+        body_len = ('%s' % len(body)).encode('ascii')
+
+        headers = {
+            b'content-encoding': [b'deflate'], b'content-length': [body_len]
+        }
+        d = DummySocket()
+        d._buffer = BytesIO(body)
+        r = HTTP11Response(200, 'OK', headers, d)
+
+        assert r.read() == b'this is test data'
+
 class DummySocket(object):
     def __init__(self):
         self.queue = []