Use cherrypy for dev server

CherryPy will support chunked encoding, so I'm using that for the test
endpoint.

Author: ionrock

dev_requirements.txt

@@ -4,7 +4,7 @@ tox
 pytest-cov
 pytest
 mock
-webtest
+cherrypy
 sphinx
 redis
 lockfile

tests/conftest.py

@@ -1,8 +1,11 @@
 from pprint import pformat
 
+import os
+import socket
+
 import pytest
 
-from webtest.http import StopableWSGIServer
+import cherrypy
 
 
 class SimpleApp(object):
@@ -33,13 +36,15 @@ def optional_cacheable_request(self, env, start_response):
         return [pformat(env).encode("utf8")]
 
     def vary_accept(self, env, start_response):
+        response = pformat(env).encode("utf8")
+
         headers = [
             ('Cache-Control', 'max-age=5000'),
             ('Content-Type', 'text/plain'),
             ('Vary', 'Accept-Encoding, Accept'),
         ]
         start_response('200 OK', headers)
-        return [pformat(env).encode("utf8")]
+        return [response]
 
     def update_etag_string(self):
         self.etag_count += 1
@@ -91,11 +96,13 @@ def multiple_choices(self, env, start_response):
 
     def stream(self, env, start_response):
         headers = [
-            ('Cache-Control', 'max-age=5000'),
             ('Content-Type', 'text/plain'),
+            ('Cache-Control', 'max-age=5000'),
         ]
         start_response('200 OK', headers)
-        return (pformat(env).encode("utf8") for i in range(10))
+
+        for i in range(10):
+            yield pformat(i).encode("utf8")
 
     def __call__(self, env, start_response):
         func = self.dispatch(env)
@@ -118,12 +125,38 @@ def server():
 
 @pytest.fixture()
 def url(server):
-    return server.application_url
+    return 'http://%s:%s/' % server.bind_addr
+
+
+def get_free_port():
+    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+    s.bind(('', 0))
+    ip, port = s.getsockname()
+    s.close()
+    ip = os.environ.get('WEBTEST_SERVER_BIND', '127.0.0.1')
+    return ip, port
 
 
 def pytest_namespace():
-    return dict(server=StopableWSGIServer.create(SimpleApp()))
+    cherrypy.tree.graft(SimpleApp(), '/')
+
+    ip, port = get_free_port()
+
+    cherrypy.config.update({
+        'server.socket_host': ip,
+        'server.socket_port': port
+    })
+
+    #turn off logging
+    logger = cherrypy.log.access_log
+    logger.removeHandler(logger.handlers[0])
+
+    cherrypy.server.start()
+    return dict(server=cherrypy.server)
 
 
 def pytest_unconfigure(config):
-    pytest.server.shutdown()
+    try:
+        cherrypy.server.stop()
+    except:
+        pass

tests/test_chunked_response.py

@@ -0,0 +1,50 @@
+"""
+Test for supporting streamed responses (Transfer-Encoding: chunked)
+"""
+from __future__ import print_function, unicode_literals
+
+import pytest
+import requests
+
+from cachecontrol import CacheControl
+
+
+@pytest.fixture()
+def sess():
+    return CacheControl(requests.Session())
+
+
+class TestChunkedResponses(object):
+
+    def test_cache_chunked_response(self, url, sess):
+        """
+        Verify that an otherwise cacheable response is cached when the
+        response is chunked.
+        """
+        url = url + 'stream'
+        r = sess.get(url)
+        from pprint import pprint
+        pprint(dict(r.headers))
+        pprint(dict(r.request.headers))
+        print(r.content)
+        assert r.headers.get('transfer-encoding') == 'chunked'
+
+        r = sess.get(url, headers={'Cache-Control': 'max-age=3600'})
+        assert r.from_cache is True
+
+    def test_stream_is_cached(self, url, sess):
+        resp_1 = sess.get(url + 'stream')
+        content_1 = resp_1.content
+
+        resp_2 = sess.get(url + 'stream')
+        content_2 = resp_1.content
+
+        assert not resp_1.from_cache
+        assert resp_2.from_cache
+        assert content_1 == content_2
+
+    def test_stream_is_not_cached_when_content_is_not_read(self, url, sess):
+        sess.get(url + 'stream', stream=True)
+        resp = sess.get(url + 'stream', stream=True)
+
+        assert not resp.from_cache

tests/test_etag.py

@@ -26,9 +26,9 @@ class TestETag(object):
     """
 
     @pytest.fixture()
-    def sess(self, server):
-        self.etag_url = urljoin(server.application_url, '/etag')
-        self.update_etag_url = urljoin(server.application_url, '/update_etag')
+    def sess(self, url):
+        self.etag_url = urljoin(url, '/etag')
+        self.update_etag_url = urljoin(url, '/update_etag')
         self.cache = DictCache()
         sess = CacheControl(
             requests.Session(),
@@ -89,9 +89,9 @@ class TestDisabledETags(object):
     response has an ETag.
     """
     @pytest.fixture()
-    def sess(self, server):
-        self.etag_url = urljoin(server.application_url, '/etag')
-        self.update_etag_url = urljoin(server.application_url, '/update_etag')
+    def sess(self, server, url):
+        self.etag_url = urljoin(url, '/etag')
+        self.update_etag_url = urljoin(url, '/update_etag')
         self.cache = DictCache()
         sess = CacheControl(requests.Session(),
                             cache=self.cache,
@@ -126,9 +126,9 @@ class TestReleaseConnection(object):
     method is not called we consume the response (which should be
     empty according to the HTTP spec) and release the connection.
     """
-    def test_not_modified_releases_connection(self, server):
+    def test_not_modified_releases_connection(self, server, url):
         sess = CacheControl(requests.Session())
-        etag_url = urljoin(server.application_url, '/etag')
+        etag_url = urljoin(url, '/etag')
         sess.get(etag_url)
 
         resp = Mock(status=304, headers={})

tests/test_max_age.py

@@ -18,8 +18,8 @@ def loads(self, request, data):
 class TestMaxAge(object):
 
     @pytest.fixture()
-    def sess(self, server):
-        self.url = server.application_url
+    def sess(self, url):
+        self.url = url
         self.cache = DictCache()
         sess = Session()
         sess.mount(

tests/test_server_http_version.py

@@ -0,0 +1,8 @@
+import requests
+
+
+def test_http11(url):
+    resp = requests.get(url)
+
+    # Making sure our test server speaks HTTP/1.1
+    assert resp.raw._fp.version == 11

tests/test_storage_filecache.py

@@ -25,8 +25,8 @@ def randomdata():
 class TestStorageFileCache(object):
 
     @pytest.fixture()
-    def sess(self, server, tmpdir):
-        self.url = server.application_url
+    def sess(self, url, tmpdir):
+        self.url = url
         self.cache = FileCache(str(tmpdir))
         sess = CacheControl(requests.Session(), cache=self.cache)
         return sess

tests/test_vary.py

@@ -5,17 +5,24 @@
 from cachecontrol.cache import DictCache
 from cachecontrol.compat import urljoin
 
+from pprint import pprint
+
 
 class TestVary(object):
 
     @pytest.fixture()
-    def sess(self, server):
-        self.url = urljoin(server.application_url, '/vary_accept')
+    def sess(self, url):
+        self.url = urljoin(url, '/vary_accept')
         self.cache = DictCache()
         sess = CacheControl(requests.Session(), cache=self.cache)
         return sess
 
     def cached_equal(self, cached, resp):
+        # remove any transfer-encoding headers as they don't apply to
+        # a cached value
+        if 'chunked' in resp.raw.headers.get('transfer-encoding', ''):
+            resp.raw.headers.pop('transfer-encoding')
+
         checks = [
             cached._fp.getvalue() == resp.content,
             cached.headers == resp.raw.headers,
@@ -25,6 +32,10 @@ def cached_equal(self, cached, resp):
             cached.strict == resp.raw.strict,
             cached.decode_content == resp.raw.decode_content,
         ]
+
+        print(checks)
+        pprint(dict(cached.headers))
+        pprint(dict(resp.raw.headers))
         return all(checks)
 
     def test_vary_example(self, sess):