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

replace fixture with constant

Browse Source
This commit is contained in:
Thomas Grainger
2024-09-18 22:34:48 +01:00
parent e2815fbc88
commit d123a5e8d0
1 changed files with 14 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
tests/integration/test_aiohttp.py
View File

@@ -13,17 +13,12 @@ aiohttp = pytest.importorskip("aiohttp")
from .aiohttp_utils import aiohttp_app, aiohttp_request # noqa: E402 from .aiohttp_utils import aiohttp_app, aiohttp_request # noqa: E402
ssl_ca_location = os.environ["REQUESTS_CA_BUNDLE"]
ssl_cert_location = os.environ["REQUESTS_CA_BUNDLE"].replace("cacert.pem", "cert.pem")
ssl_key_location = os.environ["REQUESTS_CA_BUNDLE"].replace("cacert.pem", "key.pem")
@pytest.fixture HTTPBIN_SSL_CONTEXT = ssl.create_default_context(cafile=ssl_ca_location)
def httpbin_ssl_context(): HTTPBIN_SSL_CONTEXT.load_cert_chain(ssl_cert_location, ssl_key_location)
ssl_ca_location = os.environ["REQUESTS_CA_BUNDLE"]
ssl_cert_location = os.environ["REQUESTS_CA_BUNDLE"].replace("cacert.pem", "cert.pem")
ssl_key_location = os.environ["REQUESTS_CA_BUNDLE"].replace("cacert.pem", "key.pem")
ssl_context = ssl.create_default_context(cafile=ssl_ca_location)
ssl_context.load_cert_chain(ssl_cert_location, ssl_key_location)
return ssl_context
def run_in_loop(fn): def run_in_loop(fn):
@@ -352,7 +347,7 @@ def test_double_requests(tmpdir, httpbin):
assert cassette.play_count == 2 assert cassette.play_count == 2
def test_cookies(httpbin_both, httpbin_ssl_context, tmpdir): def test_cookies(httpbin_both, tmpdir):
async def run(loop): async def run(loop):
cookies_url = httpbin_both.url + ( cookies_url = httpbin_both.url + (
"/response-headers?" "/response-headers?"
@@ -367,12 +362,12 @@ def test_cookies(httpbin_both, httpbin_ssl_context, tmpdir):
# ------------------------- Record -------------------------- # # ------------------------- Record -------------------------- #
with vcr.use_cassette(tmp) as cassette: with vcr.use_cassette(tmp) as cassette:
async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session: async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session:
cookies_resp = await session.get(cookies_url, ssl=httpbin_ssl_context) cookies_resp = await session.get(cookies_url, ssl=HTTPBIN_SSL_CONTEXT)
home_resp = await session.get( home_resp = await session.get(
home_url, home_url,
cookies=req_cookies, cookies=req_cookies,
headers=req_headers, headers=req_headers,
ssl=httpbin_ssl_context, ssl=HTTPBIN_SSL_CONTEXT,
) )
assert cassette.play_count == 0 assert cassette.play_count == 0
assert_responses(cookies_resp, home_resp) assert_responses(cookies_resp, home_resp)
@@ -380,12 +375,12 @@ def test_cookies(httpbin_both, httpbin_ssl_context, tmpdir):
# -------------------------- Play --------------------------- # # -------------------------- Play --------------------------- #
with vcr.use_cassette(tmp, record_mode=vcr.mode.NONE) as cassette: with vcr.use_cassette(tmp, record_mode=vcr.mode.NONE) as cassette:
async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session: async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session:
cookies_resp = await session.get(cookies_url, ssl=httpbin_ssl_context) cookies_resp = await session.get(cookies_url, ssl=HTTPBIN_SSL_CONTEXT)
home_resp = await session.get( home_resp = await session.get(
home_url, home_url,
cookies=req_cookies, cookies=req_cookies,
headers=req_headers, headers=req_headers,
ssl=httpbin_ssl_context, ssl=HTTPBIN_SSL_CONTEXT,
) )
assert cassette.play_count == 2 assert cassette.play_count == 2
assert_responses(cookies_resp, home_resp) assert_responses(cookies_resp, home_resp)
@@ -402,7 +397,7 @@ def test_cookies(httpbin_both, httpbin_ssl_context, tmpdir):
run_in_loop(run) run_in_loop(run)
def test_cookies_redirect(httpbin_both, httpbin_ssl_context, tmpdir): def test_cookies_redirect(httpbin_both, tmpdir):
async def run(loop): async def run(loop):
# Sets cookie as provided by the query string and redirects # Sets cookie as provided by the query string and redirects
cookies_url = httpbin_both.url + "/cookies/set?Cookie_1=Val_1" cookies_url = httpbin_both.url + "/cookies/set?Cookie_1=Val_1"
@@ -411,7 +406,7 @@ def test_cookies_redirect(httpbin_both, httpbin_ssl_context, tmpdir):
# ------------------------- Record -------------------------- # # ------------------------- Record -------------------------- #
with vcr.use_cassette(tmp) as cassette: with vcr.use_cassette(tmp) as cassette:
async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session: async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session:
cookies_resp = await session.get(cookies_url, ssl=httpbin_ssl_context) cookies_resp = await session.get(cookies_url, ssl=HTTPBIN_SSL_CONTEXT)
assert not cookies_resp.cookies assert not cookies_resp.cookies
cookies = session.cookie_jar.filter_cookies(cookies_url) cookies = session.cookie_jar.filter_cookies(cookies_url)
assert cookies["Cookie_1"].value == "Val_1" assert cookies["Cookie_1"].value == "Val_1"
@@ -422,7 +417,7 @@ def test_cookies_redirect(httpbin_both, httpbin_ssl_context, tmpdir):
# -------------------------- Play --------------------------- # # -------------------------- Play --------------------------- #
with vcr.use_cassette(tmp, record_mode=vcr.mode.NONE) as cassette: with vcr.use_cassette(tmp, record_mode=vcr.mode.NONE) as cassette:
async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session: async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session:
cookies_resp = await session.get(cookies_url, ssl=httpbin_ssl_context) cookies_resp = await session.get(cookies_url, ssl=HTTPBIN_SSL_CONTEXT)
assert not cookies_resp.cookies assert not cookies_resp.cookies
cookies = session.cookie_jar.filter_cookies(cookies_url) cookies = session.cookie_jar.filter_cookies(cookies_url)
assert cookies["Cookie_1"].value == "Val_1" assert cookies["Cookie_1"].value == "Val_1"
@@ -436,7 +431,7 @@ def test_cookies_redirect(httpbin_both, httpbin_ssl_context, tmpdir):
"Cookie_1=Val_1; Expires=Wed, 21 Oct 2015 07:28:00 GMT", "Cookie_1=Val_1; Expires=Wed, 21 Oct 2015 07:28:00 GMT",
] ]
async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session: async with aiohttp.ClientSession(loop=loop, cookie_jar=aiohttp.CookieJar(unsafe=True)) as session:
cookies_resp = await session.get(cookies_url, ssl=httpbin_ssl_context) cookies_resp = await session.get(cookies_url, ssl=HTTPBIN_SSL_CONTEXT)
assert not cookies_resp.cookies assert not cookies_resp.cookies
cookies = session.cookie_jar.filter_cookies(cookies_url) cookies = session.cookie_jar.filter_cookies(cookies_url)
assert cookies["Cookie_1"].value == "Val_1" assert cookies["Cookie_1"].value == "Val_1"