Add tests for connection leaks on retry/redirect
MODIFY
dummyserver/handlers.py
dummyserver/handlers.py
@@ -165,6 +165,9 @@ class TestingApp(RequestHandler):
headers = [('Location', target)]
return Response(status=status, headers=headers)
+ def not_found(self, request):
+ return Response('Not found', status='404 Not Found')
+
def multi_redirect(self, request):
"Performs a redirect chain based on ``redirect_codes``"
codes = request.params.get('redirect_codes', '200').decode('utf-8')

MODIFY
test/with_dummyserver/test_connectionpool.py
test/with_dummyserver/test_connectionpool.py
@@ -987,5 +987,31 @@ class TestFileBodiesOnRetryOrRedirect(HTTPDummyServerTestCase):
self.assertTrue('Unable to record file position for' in str(e))
+class TestRetryPoolSize(HTTPDummyServerTestCase):
+ def setUp(self):
+ retries = Retry(
+ total = 3,
+ backoff_factor = 0.01,
+ raise_on_status = False,
+ status_forcelist = [404],
+ )
+ self.pool = HTTPConnectionPool(self.host, self.port, maxsize=10, retries=retries, block=True)
+ self.addCleanup(self.pool.close)
+
+ def test_pool_size_retry(self):
+ resp = self.pool.urlopen('GET', '/not_found', preload_content=False)
+ self.assertEquals(self.pool.num_connections, 1)
+
+
+class TestRedirectPoolSize(HTTPDummyServerTestCase):
+ def setUp(self):
+ self.pool = HTTPConnectionPool(self.host, self.port, maxsize=10, retries=5, block=True)
+ self.addCleanup(self.pool.close)
+
+ def test_pool_size_redirect(self):
+ resp = self.pool.urlopen('GET', '/redirect', preload_content=False)
+ self.assertEquals(self.pool.num_connections, 1)
+
+
if __name__ == '__main__':
unittest.main()

Add tests for connection leaks on retry/redirect