Mirroring/vcrpy
1
0
Fork 0
mirror of https://github.com/kevin1024/vcrpy.git synced 2025-12-10 01:25:34 +00:00
Code Releases Activity

convert requests tests to pytest

Browse Source
This commit is contained in:
Kevin McCarthy
2013-08-10 17:03:20 -10:00
parent c5294a09f7
commit a1229190f5
1 changed files with 104 additions and 142 deletions
Download Patch File Download Diff File Expand all files Collapse all files

246
tests/integration/test_requests.py
View File

@@ -3,161 +3,123 @@
# coding=utf-8 # coding=utf-8
# Internal imports # Internal imports
import vcr
from .common import TestVCR
import os import os
import pytest import pytest
requests = pytest.importorskip("requests") requests = pytest.importorskip("requests")
import vcr
class TestRequestsBase(TestVCR): @pytest.fixture(params=["https","http"])
'''Some utility for running Requests tests''' def scheme(request):
fixtures = os.path.join('tests', 'fixtures', 'requests') """
Fixture that returns both http and https
"""
return request.param
class TestHTTPRequests(TestRequestsBase): def test_status_code(scheme, tmpdir):
'''Some tests using requests and http''' '''Ensure that we can read the status code'''
scheme = 'http' url = scheme + '://httpbin.org/'
with vcr.use_cassette(str(tmpdir.join('atts.yaml'))) as cass:
# Ensure that this is empty to begin with
assert len(cass) == 0
assert cass.play_count == 0
assert requests.get(url).status_code == requests.get(url).status_code
# Ensure that we've now cached a single response
assert len(cass) == 1
assert cass.play_count == 1
def test_status_code(self): def test_headers(scheme, tmpdir):
'''Ensure that we can read the status code''' '''Ensure that we can read the headers back'''
url = self.scheme + '://httpbin.org/' url = scheme + '://httpbin.org/'
with vcr.use_cassette(self.fixture('atts.yaml')) as cass: with vcr.use_cassette(str(tmpdir.join('headers.yaml'))) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) assert len(cass) == 0
self.assertEqual(cass.play_count, 0) assert cass.play_count == 0
self.assertEqual( assert requests.get(url).headers == requests.get(url).headers
requests.get(url).status_code, # Ensure that we've now cached a single response
requests.get(url).status_code) assert len(cass) == 1
# Ensure that we've now cached a single response assert cass.play_count == 1
self.assertEqual(len(cass), 1)
self.assertEqual(cass.play_count, 1)
def test_headers(self): def test_body(tmpdir, scheme):
'''Ensure that we can read the headers back''' '''Ensure the responses are all identical enough'''
url = self.scheme + '://httpbin.org/' url = scheme + '://httpbin.org/bytes/1024'
with vcr.use_cassette(self.fixture('headers.yaml')) as cass: with vcr.use_cassette(str(tmpdir.join('body.yaml'))) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) assert len(cass) == 0
self.assertEqual(cass.play_count, 0) assert cass.play_count == 0
self.assertEqual( assert requests.get(url).content == requests.get(url).content
requests.get(url).headers, # Ensure that we've now cached a single response
requests.get(url).headers) assert len(cass) == 1
# Ensure that we've now cached a single response assert cass.play_count == 1
self.assertEqual(len(cass), 1)
self.assertEqual(cass.play_count, 1)
def test_body(self): def test_auth(tmpdir, scheme):
'''Ensure the responses are all identical enough''' '''Ensure that we can handle basic auth'''
url = self.scheme + '://httpbin.org/bytes/1024' auth = ('user', 'passwd')
with vcr.use_cassette(self.fixture('body.yaml')) as cass: url = scheme + '://httpbin.org/basic-auth/user/passwd'
# Ensure that this is empty to begin with with vcr.use_cassette(str(tmpdir.join('auth.yaml'))) as cass:
self.assertEqual(len(cass), 0) # Ensure that this is empty to begin with
self.assertEqual(cass.play_count, 0) assert len(cass) == 0
self.assertEqual( assert cass.play_count == 0
requests.get(url).content, one = requests.get(url, auth=auth)
requests.get(url).content) two = requests.get(url, auth=auth)
# Ensure that we've now cached a single response assert one.content == two.content
self.assertEqual(len(cass), 1) assert one.status_code == two.status_code
self.assertEqual(cass.play_count, 1) # Ensure that we've now cached a single response
assert len(cass) == 1
assert cass.play_count == 1
def test_auth(self): def test_auth_failed(tmpdir, scheme):
'''Ensure that we can handle basic auth''' '''Ensure that we can save failed auth statuses'''
auth = ('user', 'passwd') auth = ('user', 'wrongwrongwrong')
url = self.scheme + '://httpbin.org/basic-auth/user/passwd' url = scheme + '://httpbin.org/basic-auth/user/passwd'
with vcr.use_cassette(self.fixture('auth.yaml')) as cass: with vcr.use_cassette(str(tmpdir.join('auth-failed.yaml'))) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) assert len(cass) == 0
self.assertEqual(cass.play_count, 0) assert cass.play_count == 0
one = requests.get(url, auth=auth) one = requests.get(url, auth=auth)
two = requests.get(url, auth=auth) two = requests.get(url, auth=auth)
self.assertEqual(one.content, two.content) assert one.content == two.content
self.assertEqual(one.status_code, two.status_code) assert one.status_code == two.status_code == 401
# Ensure that we've now cached a single response # Ensure that we've now cached a single response
self.assertEqual(len(cass), 1) assert len(cass) == 1
self.assertEqual(cass.play_count, 1) assert cass.play_count == 1
def test_auth_failed(self): def test_post(tmpdir, scheme):
'''Ensure that we can save failed auth statuses''' '''Ensure that we can post and cache the results'''
auth = ('user', 'wrongwrongwrong') data = {'key1': 'value1', 'key2': 'value2'}
url = self.scheme + '://httpbin.org/basic-auth/user/passwd' url = scheme + '://httpbin.org/post'
with vcr.use_cassette(self.fixture('auth-failed.yaml')) as cass: with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cass:
# Ensure that this is empty to begin with # Ensure that this is empty to begin with
self.assertEqual(len(cass), 0) assert len(cass) == 0
self.assertEqual(cass.play_count, 0) assert cass.play_count == 0
one = requests.get(url, auth=auth) assert requests.post(url, data).content == requests.post(url, data).content
two = requests.get(url, auth=auth) # Ensure that we've now cached a single response
self.assertEqual(one.content, two.content) assert len(cass) == 1
self.assertEqual(one.status_code, two.status_code) assert cass.play_count == 1
self.assertNotEqual(one.status_code, 200)
# Ensure that we've now cached a single response
self.assertEqual(len(cass), 1)
self.assertEqual(cass.play_count, 1)
def test_post(self): def test_redirects(tmpdir, scheme):
'''Ensure that we can post and cache the results''' '''Ensure that we can handle redirects'''
data = {'key1': 'value1', 'key2': 'value2'} url = scheme + '://httpbin.org/redirect-to?url=bytes/1024'
url = self.scheme + '://httpbin.org/post' with vcr.use_cassette(str(tmpdir.join('redirect.yaml'))) as cass:
with vcr.use_cassette(self.fixture('redirect.yaml')) as cass: # Ensure that this is empty to begin with
# Ensure that this is empty to begin with assert len(cass) == 0
self.assertEqual(len(cass), 0) assert cass.play_count == 0
self.assertEqual(cass.play_count, 0) assert requests.get(url).content == requests.get(url).content
self.assertEqual( # Ensure that we've now cached /two/ responses. One for the redirect
requests.post(url, data).content, # and one for the final fetch
requests.post(url, data).content) assert len(cass) == 2
# Ensure that we've now cached a single response assert cass.play_count == 2
self.assertEqual(len(cass), 1)
self.assertEqual(cass.play_count, 1)
def test_redirects(self): def test_cross_scheme(tmpdir, scheme):
'''Ensure that we can handle redirects''' '''Ensure that requests between schemes are treated separately'''
url = self.scheme + '://httpbin.org/redirect-to?url=bytes/1024' # First fetch a url under http, and then again under https and then
with vcr.use_cassette(self.fixture('redirect.yaml')) as cass: # ensure that we haven't served anything out of cache, and we have two
# Ensure that this is empty to begin with # requests / response pairs in the cassette
self.assertEqual(len(cass), 0) with vcr.use_cassette(str(tmpdir.join('cross_scheme.yaml'))) as cass:
self.assertEqual(cass.play_count, 0) requests.get('https://httpbin.org/')
self.assertEqual( requests.get('http://httpbin.org/')
requests.get(url).content, assert len(cass) == 2
requests.get(url).content) assert cass.play_count == 0
# 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(cass.play_count, 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 http, 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(cass.play_count, 0)
class TestWild(TestRequestsBase):
'''Test some examples from the wild'''
fixtures = os.path.join('tests', 'fixtures', 'wild')
def tearDown(self):
# No deleting our directory, and ensure that it exists
self.assertTrue(os.path.exists(self.fixture()))
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(cass.play_count, 2)