Expose cert and proxies as attributes of the last request.
When adding timeout and allow_redirects we added the ability to fetch some of the connection parameters from the request. Given we have made that link we should also expose the verify, cert and proxies so users can test that. Particularly testing that cert has been passed through to a request is something i've come across a number of times. Change-Id: I2c60be44769607c32eaf68ad697656b85cae0737
This commit is contained in:
parent
459e13cb24
commit
780eb4b4fe
@ -10,6 +10,7 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import copy
|
||||
import json
|
||||
import weakref
|
||||
|
||||
@ -36,8 +37,14 @@ class _RequestObjectProxy(object):
|
||||
self._matcher = None
|
||||
self._url_parts_ = None
|
||||
self._qs = None
|
||||
|
||||
# All of these params should always exist but we use a default
|
||||
# to make the test setup easier.
|
||||
self._timeout = kwargs.pop('timeout', None)
|
||||
self._allow_redirects = kwargs.pop('allow_redirects', None)
|
||||
self._verify = kwargs.pop('verify', None)
|
||||
self._cert = kwargs.pop('cert', None)
|
||||
self._proxies = copy.deepcopy(kwargs.pop('proxies', {}))
|
||||
|
||||
def __getattr__(self, name):
|
||||
return getattr(self._request, name)
|
||||
@ -80,6 +87,18 @@ class _RequestObjectProxy(object):
|
||||
def allow_redirects(self):
|
||||
return self._allow_redirects
|
||||
|
||||
@property
|
||||
def verify(self):
|
||||
return self._verify
|
||||
|
||||
@property
|
||||
def cert(self):
|
||||
return self._cert
|
||||
|
||||
@property
|
||||
def proxies(self):
|
||||
return self._proxies
|
||||
|
||||
@classmethod
|
||||
def _create(cls, *args, **kwargs):
|
||||
return cls(requests.Request(*args, **kwargs).prepare())
|
||||
|
@ -577,6 +577,23 @@ class SessionAdapterTests(base.TestCase):
|
||||
self.assertEqual(set(['/foo', '/bar']), set(resp.cookies.list_paths()))
|
||||
self.assertEqual(['.example.com'], resp.cookies.list_domains())
|
||||
|
||||
def test_base_params(self):
|
||||
data = 'testdata'
|
||||
self.adapter.register_uri('GET', self.url, text=data)
|
||||
resp = self.session.get(self.url)
|
||||
|
||||
self.assertEqual('GET', self.adapter.last_request.method)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertEqual(data, resp.text)
|
||||
|
||||
self.assertIs(None, self.adapter.last_request.allow_redirects)
|
||||
self.assertIs(None, self.adapter.last_request.timeout)
|
||||
self.assertIs(True, self.adapter.last_request.verify)
|
||||
self.assertIs(None, self.adapter.last_request.cert)
|
||||
|
||||
# actually it's an OrderedDict, but equality works fine
|
||||
self.assertEqual({}, self.adapter.last_request.proxies)
|
||||
|
||||
def test_allow_redirects(self):
|
||||
data = 'testdata'
|
||||
self.adapter.register_uri('GET', self.url, text=data, status_code=300)
|
||||
@ -598,3 +615,52 @@ class SessionAdapterTests(base.TestCase):
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertEqual(data, resp.text)
|
||||
self.assertEqual(timeout, self.adapter.last_request.timeout)
|
||||
|
||||
def test_verify_false(self):
|
||||
data = 'testdata'
|
||||
verify = False
|
||||
|
||||
self.adapter.register_uri('GET', self.url, text=data)
|
||||
resp = self.session.get(self.url, verify=verify)
|
||||
self.assertEqual('GET', self.adapter.last_request.method)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertEqual(data, resp.text)
|
||||
self.assertIs(verify, self.adapter.last_request.verify)
|
||||
|
||||
def test_verify_path(self):
|
||||
data = 'testdata'
|
||||
verify = '/path/to/cacerts.pem'
|
||||
|
||||
self.adapter.register_uri('GET', self.url, text=data)
|
||||
resp = self.session.get(self.url, verify=verify)
|
||||
self.assertEqual('GET', self.adapter.last_request.method)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertEqual(data, resp.text)
|
||||
self.assertEqual(verify, self.adapter.last_request.verify)
|
||||
|
||||
def test_certs(self):
|
||||
data = 'testdata'
|
||||
cert = ('/path/to/cert.pem', 'path/to/key.pem')
|
||||
|
||||
self.adapter.register_uri('GET', self.url, text=data)
|
||||
resp = self.session.get(self.url, cert=cert)
|
||||
|
||||
self.assertEqual('GET', self.adapter.last_request.method)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertEqual(data, resp.text)
|
||||
self.assertEqual(cert, self.adapter.last_request.cert)
|
||||
self.assertTrue(self.adapter.last_request.verify)
|
||||
|
||||
def test_proxies(self):
|
||||
data = 'testdata'
|
||||
proxies = {'http': 'foo.bar:3128',
|
||||
'http://host.name': 'foo.bar:4012'}
|
||||
|
||||
self.adapter.register_uri('GET', self.url, text=data)
|
||||
resp = self.session.get(self.url, proxies=proxies)
|
||||
|
||||
self.assertEqual('GET', self.adapter.last_request.method)
|
||||
self.assertEqual(200, resp.status_code)
|
||||
self.assertEqual(data, resp.text)
|
||||
self.assertEqual(proxies, self.adapter.last_request.proxies)
|
||||
self.assertIsNot(proxies, self.adapter.last_request.proxies)
|
||||
|
Loading…
x
Reference in New Issue
Block a user