Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_cache_control_header_nocache(self):
"""Testing the cache with the no-cache control"""
request = Request('http://no_cache', method='GET')
first_resp = self.cache.make_request(request)
second_resp = self.cache.make_request(request)
self.assertEqual(self.urlopener.get_hit_count('http://no_cache'), 2)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertFalse(isinstance(second_resp, CachedHTTPResponse))
def test_vary_header(self):
"""Testing the cache with the Vary header"""
request = Request('http://vary', headers={'User-agent': 'foo'},
method='GET')
first_resp = self.cache.make_request(request)
second_resp = self.cache.make_request(request)
self.assertEqual(self.urlopener.get_hit_count('http://vary'), 1)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertTrue(isinstance(second_resp, CachedHTTPResponse))
def test_cache_control_header_nocache_with_etag(self):
"""Testing the cache with the no-cache control and a specified ETag"""
request = Request('http://no_cache_etag', method='GET')
first_resp = self.cache.make_request(request)
second_resp = self.cache.make_request(request)
self.assertEqual(self.urlopener.get_hit_count('http://no_cache_etag'),
2)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertTrue(isinstance(second_resp, CachedHTTPResponse))
def test_cache_control_header_must_revalidate(self):
"""Testing the cache with the must-revalidate control"""
request = Request('http://must_revalidate', method='GET')
first_resp = self.cache.make_request(request)
second_resp = self.cache.make_request(request)
self.assertEqual(
self.urlopener.get_hit_count('http://must_revalidate'),
2)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertTrue(isinstance(second_resp, CachedHTTPResponse))
def test_vary_header_different_requests(self):
"""Testing the cache with the Vary header and different requests"""
first_request = Request('http://vary', headers={'User-agent': 'foo'},
method='GET')
second_request = Request('http://vary', headers={'User-agent': 'bar'},
method='GET')
first_resp = self.cache.make_request(first_request)
second_resp = self.cache.make_request(second_request)
self.assertEqual(self.urlopener.get_hit_count('http://vary'), 2)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertFalse(isinstance(second_resp, CachedHTTPResponse))
def test_expires_header_expired(self):
"""Testing the cache with the Expires header in the past"""
request = Request('http://expired', method='GET')
first_resp = self.cache.make_request(request)
second_resp = self.cache.make_request(request)
self.assertEqual(self.urlopener.get_hit_count('http://expired'), 2)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertFalse(isinstance(second_resp, CachedHTTPResponse))
def test_cache_control_header_no_store(self):
"""Testing the cache with the no-store control"""
request = Request('http://no_store', method='GET')
first_resp = self.cache.make_request(request)
second_resp = self.cache.make_request(request)
self.assertEqual(self.urlopener.get_hit_count('http://no_store'), 2)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertFalse(isinstance(second_resp, CachedHTTPResponse))
def test_cache_control_header_nocache_with_last_modified_updated(self):
"""Testing the cache with the no-cache control and an updated
Last-Modified header
"""
endpoint = 'http://no_cache_lastmodified_updated'
future_date = datetime.datetime.utcnow() + datetime.timedelta(days=1)
self.urlopener.endpoints[endpoint] = {
'hit_count': 0,
'headers': {
'Cache-Control': 'no-cache',
'Last-Modified': '1999-12-31T00:00:00'
},
}
request = Request(endpoint, method='GET')
first_resp = self.cache.make_request(request)
self.urlopener.endpoints[endpoint]['headers']['Last-Modified'] = (
future_date.strftime(CacheEntry.DATE_FORMAT))
second_resp = self.cache.make_request(request)
third_resp = self.cache.make_request(request)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertFalse(isinstance(second_resp, CachedHTTPResponse))
self.assertTrue(isinstance(third_resp, CachedHTTPResponse))
self.assertEqual(self.urlopener.get_hit_count(endpoint), 3)
def test_cache_control_header_nocache_with_etag_updated(self):
"""Testing the cache with the no-cache control and an updated ETag"""
request = Request('http://no_cache_etag', method='GET')
first_resp = self.cache.make_request(request)
# Pretend the end point has been updated since the last request.
self.urlopener.endpoints['http://no_cache_etag']['headers']['ETag'] = (
'new-etag')
second_resp = self.cache.make_request(request)
third_resp = self.cache.make_request(request)
self.assertFalse(isinstance(first_resp, CachedHTTPResponse))
self.assertFalse(isinstance(second_resp, CachedHTTPResponse))
self.assertTrue(isinstance(third_resp, CachedHTTPResponse))
self.assertEqual(self.urlopener.get_hit_count('http://no_cache_etag'),
3)
The request argument should be an instance of
'rbtools.api.request.HttpRequest'.
"""
try:
content_type, body = request.encode_multipart_formdata()
headers = request.headers
if body:
headers.update({
'Content-Type': content_type,
'Content-Length': str(len(body)),
})
else:
headers['Content-Length'] = '0'
rsp = self._urlopen(Request(
request.url, body, headers, request.method))
except HTTPError as e:
self.process_error(e.code, e.read())
except URLError as e:
raise ServerInterfaceError('%s' % e.reason)
if self.save_cookies:
try:
self.cookie_jar.save()
except IOError:
pass
return rsp