mirror of
https://github.com/kevin1024/vcrpy.git
synced 2025-12-10 09:35:34 +00:00
Substantial refactoring
This refactoring includes some PEP-8 compliance changes, as well as some more complete testing to ensure that we're in fact serving everything out of cassettes when we thing we are. Incidentally, it also includes fixes for #3 and #4
This commit is contained in:
Dan Lecocq
committed by
Kevin McCarthy
Kevin McCarthy
parent
3742e2fdc0
commit
b488ca67fe
@@ -1,110 +1,163 @@
|
||||
# coding=utf-8
|
||||
import os
|
||||
import unittest
|
||||
import vcr
|
||||
import pytest
|
||||
from utils import assert_httpbin_responses_equal
|
||||
'''Test requests' interaction with vcr'''
|
||||
|
||||
# coding=utf-8
|
||||
|
||||
# Internal imports
|
||||
import vcr
|
||||
from .common import TestVCR
|
||||
|
||||
import os
|
||||
import pytest
|
||||
requests = pytest.importorskip("requests")
|
||||
|
||||
TEST_CASSETTE_FILE = 'cassettes/test_req.yaml'
|
||||
|
||||
class TestRequestsGet(unittest.TestCase):
|
||||
class TestRequestsBase(TestVCR):
|
||||
'''Some utility for running Requests tests'''
|
||||
fixtures = os.path.join('tests', 'fixtures', 'requests')
|
||||
|
||||
def setUp(self):
|
||||
self.unmolested_response = requests.get('http://httpbin.org/get')
|
||||
with vcr.use_cassette(TEST_CASSETTE_FILE):
|
||||
self.initial_response = requests.get('http://httpbin.org/get')
|
||||
self.cached_response = requests.get('http://httpbin.org/get')
|
||||
|
||||
class TestHTTPRequests(TestRequestsBase):
|
||||
'''Some tests using requests and http'''
|
||||
scheme = 'http'
|
||||
|
||||
def test_status_code(self):
|
||||
'''Ensure that we can read the status code'''
|
||||
url = self.scheme + '://httpbin.org/'
|
||||
with vcr.use_cassette(self.fixture('atts.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
self.assertEqual(
|
||||
requests.get(url).status_code,
|
||||
requests.get(url).status_code)
|
||||
# Ensure that we've now cached a single response
|
||||
self.assertEqual(len(cass), 1)
|
||||
self.assertEqual(len(cass.cached()), 1)
|
||||
|
||||
def test_headers(self):
|
||||
'''Ensure that we can read the headers back'''
|
||||
url = self.scheme + '://httpbin.org/'
|
||||
with vcr.use_cassette(self.fixture('headers.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
self.assertEqual(
|
||||
requests.get(url).headers,
|
||||
requests.get(url).headers)
|
||||
# Ensure that we've now cached a single response
|
||||
self.assertEqual(len(cass), 1)
|
||||
self.assertEqual(len(cass.cached()), 1)
|
||||
|
||||
def test_body(self):
|
||||
'''Ensure the responses are all identical enough'''
|
||||
url = self.scheme + '://httpbin.org/bytes/1024'
|
||||
with vcr.use_cassette(self.fixture('body.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
self.assertEqual(
|
||||
requests.get(url).content,
|
||||
requests.get(url).content)
|
||||
# Ensure that we've now cached a single response
|
||||
self.assertEqual(len(cass), 1)
|
||||
self.assertEqual(len(cass.cached()), 1)
|
||||
|
||||
def test_auth(self):
|
||||
'''Ensure that we can handle basic auth'''
|
||||
auth = ('user', 'passwd')
|
||||
url = self.scheme + '://httpbin.org/basic-auth/user/passwd'
|
||||
with vcr.use_cassette(self.fixture('auth.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
one = requests.get(url, auth=auth)
|
||||
two = requests.get(url, auth=auth)
|
||||
self.assertEqual(one.content, two.content)
|
||||
self.assertEqual(one.status_code, two.status_code)
|
||||
# Ensure that we've now cached a single response
|
||||
self.assertEqual(len(cass), 1)
|
||||
self.assertEqual(len(cass.cached()), 1)
|
||||
|
||||
def test_auth_failed(self):
|
||||
'''Ensure that we can save failed auth statuses'''
|
||||
auth = ('user', 'wrongwrongwrong')
|
||||
url = self.scheme + '://httpbin.org/basic-auth/user/passwd'
|
||||
with vcr.use_cassette(self.fixture('auth-failed.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
one = requests.get(url, auth=auth)
|
||||
two = requests.get(url, auth=auth)
|
||||
self.assertEqual(one.content, two.content)
|
||||
self.assertEqual(one.status_code, two.status_code)
|
||||
self.assertNotEqual(one.status_code, 200)
|
||||
# Ensure that we've now cached a single response
|
||||
self.assertEqual(len(cass), 1)
|
||||
self.assertEqual(len(cass.cached()), 1)
|
||||
|
||||
def test_post(self):
|
||||
'''Ensure that we can post and cache the results'''
|
||||
data = {'key1': 'value1', 'key2': 'value2'}
|
||||
url = self.scheme + '://httpbin.org/post'
|
||||
with vcr.use_cassette(self.fixture('redirect.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
self.assertEqual(
|
||||
requests.post(url, data).content,
|
||||
requests.post(url, data).content)
|
||||
# Ensure that we've now cached a single response
|
||||
self.assertEqual(len(cass), 1)
|
||||
self.assertEqual(len(cass.cached()), 1)
|
||||
|
||||
def test_redirects(self):
|
||||
'''Ensure that we can handle redirects'''
|
||||
url = self.scheme + '://httpbin.org/redirect-to?url=bytes/1024'
|
||||
with vcr.use_cassette(self.fixture('redirect.yaml')) as cass:
|
||||
# Ensure that this is empty to begin with
|
||||
self.assertEqual(len(cass), 0)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
self.assertEqual(
|
||||
requests.get(url).content,
|
||||
requests.get(url).content)
|
||||
# Ensure that we've now cached /two/ responses. One for the redirect
|
||||
# and one for the final fetch
|
||||
self.assertEqual(len(cass), 2)
|
||||
self.assertEqual(len(cass.cached()), 2)
|
||||
|
||||
|
||||
class TestHTTPSRequests(TestHTTPRequests):
|
||||
'''Same as above, now in https'''
|
||||
scheme = 'https'
|
||||
|
||||
def test_cross_scheme(self):
|
||||
'''Ensure that requests between schemes are treated separately'''
|
||||
# First fetch a url under https, and then again under https and then
|
||||
# ensure that we haven't served anything out of cache, and we have two
|
||||
# requests / response pairs in the cassette
|
||||
with vcr.use_cassette(self.fixture('cross_scheme.yaml')) as cass:
|
||||
requests.get('https://httpbin.org/')
|
||||
requests.get('http://httpbin.org/')
|
||||
self.assertEqual(len(cass), 2)
|
||||
self.assertEqual(len(cass.cached()), 0)
|
||||
|
||||
|
||||
class TestWild(TestRequestsBase):
|
||||
'''Test some examples from the wild'''
|
||||
fixtures = os.path.join('tests', 'fixtures', 'wild')
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
os.remove(TEST_CASSETTE_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
# No deleting our directory, and ensure that it exists
|
||||
self.assertTrue(os.path.exists(self.fixture()))
|
||||
|
||||
def test_initial_response_code(self):
|
||||
self.assertEqual(self.unmolested_response.status_code, self.initial_response.status_code)
|
||||
|
||||
def test_cached_response_code(self):
|
||||
self.assertEqual(self.unmolested_response.status_code, self.cached_response.status_code)
|
||||
|
||||
def test_initial_response_headers(self):
|
||||
self.assertEqual(self.unmolested_response.headers['content-type'], self.initial_response.headers['content-type'])
|
||||
|
||||
def test_cached_response_headers(self):
|
||||
self.assertEqual(self.unmolested_response.headers['content-type'], self.cached_response.headers['content-type'])
|
||||
|
||||
def test_initial_response_text(self):
|
||||
assert_httpbin_responses_equal(self.unmolested_response.text, self.initial_response.text)
|
||||
|
||||
def test_cached_response_text(self):
|
||||
assert_httpbin_responses_equal(self.unmolested_response.text, self.cached_response.text)
|
||||
|
||||
|
||||
class TestRequestsAuth(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.unmolested_response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd'))
|
||||
with vcr.use_cassette(TEST_CASSETTE_FILE):
|
||||
self.initial_response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd'))
|
||||
self.cached_response = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwd'))
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
os.remove(TEST_CASSETTE_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def test_initial_response_code(self):
|
||||
self.assertEqual(self.unmolested_response.status_code, self.initial_response.status_code)
|
||||
|
||||
def test_cached_response_code(self):
|
||||
self.assertEqual(self.unmolested_response.status_code, self.cached_response.status_code)
|
||||
|
||||
def test_cached_response_auth_can_fail(self):
|
||||
auth_fail_cached = requests.get('https://httpbin.org/basic-auth/user/passwd', auth=('user', 'passwdzzz'))
|
||||
self.assertNotEqual(self.unmolested_response.status_code, auth_fail_cached.status_code)
|
||||
|
||||
class TestRequestsPost(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
payload = {'key1': 'value1', 'key2': 'value2'}
|
||||
self.unmolested_response = requests.post('http://httpbin.org/post', payload)
|
||||
with vcr.use_cassette(TEST_CASSETTE_FILE):
|
||||
self.initial_response = requests.post('http://httpbin.org/post', payload)
|
||||
self.cached_response = requests.post('http://httpbin.org/post', payload)
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
os.remove(TEST_CASSETTE_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def test_initial_post_response_text(self):
|
||||
assert_httpbin_responses_equal(self.unmolested_response.text, self.initial_response.text)
|
||||
|
||||
def test_cached_post_response_text(self):
|
||||
assert_httpbin_responses_equal(self.unmolested_response.text, self.cached_response.text)
|
||||
|
||||
|
||||
class TestRequestsHTTPS(unittest.TestCase):
|
||||
maxDiff = None
|
||||
|
||||
def setUp(self):
|
||||
self.unmolested_response = requests.get('https://httpbin.org/get')
|
||||
with vcr.use_cassette(TEST_CASSETTE_FILE):
|
||||
self.initial_response = requests.get('https://httpbin.org/get')
|
||||
self.cached_response = requests.get('https://httpbin.org/get')
|
||||
|
||||
def tearDown(self):
|
||||
try:
|
||||
os.remove(TEST_CASSETTE_FILE)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
def test_initial_https_response_text(self):
|
||||
assert_httpbin_responses_equal(self.unmolested_response.text, self.initial_response.text)
|
||||
|
||||
def test_cached_https_response_text(self):
|
||||
assert_httpbin_responses_equal(self.unmolested_response.text, self.cached_response.text)
|
||||
def test_domain_redirect(self):
|
||||
'''Ensure that redirects across domains are considered unique'''
|
||||
# In this example, seomoz.org redirects to moz.com, and if those
|
||||
# requests are considered identical, then we'll be stuck in a redirect
|
||||
# loop.
|
||||
url = 'http://seomoz.org/'
|
||||
with vcr.use_cassette(self.fixture('domain_redirect.yaml')) as cass:
|
||||
requests.get(url, headers={'User-Agent': 'vcrpy-test'})
|
||||
# Ensure that we've now served two responses. One for the original
|
||||
# redirect, and a second for the actual fetch
|
||||
self.assertEqual(len(cass.cached()), 2)
|
||||
|
||||
Reference in New Issue
Block a user