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

Compare commits

..

7 Commits

Author SHA1 Message Date
Kevin McCarthy
34d5384318 bump version to v5.0.0 2023-06-26 12:54:39 -05:00
Sebastian Pipping
ad1010d0f8 Merge pull request #695 from kevin1024/drop37
Drop support for Python 3.7 (after 2023-06-27)
2023-06-26 18:32:42 +02:00
Amos Ng
d99593bcd3 Split persister errors into CassetteNotFoundError and CassetteDecodeError (#681) 2023-06-26 18:27:35 +02:00
Sebastian Pipping
8c03c37df4 Merge pull request #725 from kevin1024/make-assert-is-json-less-misleading
assertions.py: Fix mis-leading `assert_is_json`
2023-06-26 17:43:28 +02:00
Jair Henrique
b827cbe2da Drop support to python 3.7 2023-06-26 11:46:20 -03:00
Kevin McCarthy
92ca5a102c fix misspelled word 2023-06-26 09:22:16 -05:00
Sebastian Pipping
f21c8f0224 assertions.py: Fix mis-leading assert_is_json
Parameter name "a_string" was mistaken and function
name "assert_is_json" was less clear than ideal,
given that it explicitly needs bytes unlike json.loads .
2023-06-24 15:59:23 +02:00
39 changed files with 153 additions and 128 deletions

View File

@@ -13,7 +13,7 @@ jobs:
strategy:
fail-fast: false
matrix:
python-version: ["3.7", "3.8", "3.9", "3.10", "3.11", "pypy-3.8"]
python-version: ["3.8", "3.9", "3.10", "3.11", "pypy-3.8"]
steps:
- uses: actions/checkout@v3.5.2

View File

@@ -136,7 +136,8 @@ Create your own persistence class, see the example below:
Your custom persister must implement both ``load_cassette`` and ``save_cassette``
methods. The ``load_cassette`` method must return a deserialized cassette or raise
``ValueError`` if no cassette is found.
either ``CassetteNotFoundError`` if no cassette is found, or ``CassetteDecodeError``
if the cassette cannot be successfully deserialized.
Once the persister class is defined, register with VCR like so...

View File

@@ -7,11 +7,14 @@ For a full list of triaged issues, bugs and PRs and what release they are target
All help in providing PRs to close out bug issues is appreciated. Even if that is providing a repo that fully replicates issues. We have very generous contributors that have added these to bug issues which meant another contributor picked up the bug and closed it out.
- 5.0.0
- BREAKING CHANGE: Drop support for Python 3.7. 3.7 is EOL as of 6/27/23 Thanks @jairhenrique
- BREAKING CHANGE: Custom Cassette persisters no longer catch ValueError. If you have implemented a custom persister (has anyone implemented a custom persister? Let us know!) then you will need to throw a CassetteNotFoundError when unable to find a cassette. See #681 for discussion and reason for this change. Thanks @amosjyng for the PR and the review from @hartwork
- 4.4.0
- HUGE thanks to @hartwork for all the work done on this release!
- Bring vcr/unittest in to vcrpy as a full feature of vcr instead of a separate library. Big thanks to @hartwork for doing this and to @agriffis for originally creating the library
- Make decompression robust towards already decompressed input (thanks @hartwork)
- Bugfix: Add read1 method (fixes compatability with biopython), thanks @mghantous
- Bugfix: Add read1 method (fixes compatibility with biopython), thanks @mghantous
- Bugfix: Prevent filters from corrupting request (thanks @abramclark)
- Bugfix: Add support for `response.raw.stream()` to fix urllib v2 compat
- Bugfix: Replace `assert` with `raise AssertionError`: fixes support for `PYTHONOPTIMIZE=1`

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# vcrpy documentation build configuration file, created by
# sphinx-quickstart on Sun Sep 13 11:18:00 2015.

View File

@@ -96,11 +96,11 @@ The test suite is pretty big and slow, but you can tell tox to only run specific
tox -e {pyNN}-{HTTP_LIBRARY} -- <pytest flags passed through>
tox -e py37-requests -- -v -k "'test_status_code or test_gzip'"
tox -e py37-requests -- -v --last-failed
tox -e py38-requests -- -v -k "'test_status_code or test_gzip'"
tox -e py38-requests -- -v --last-failed
This will run only tests that look like ``test_status_code`` or
``test_gzip`` in the test suite, and only in the python 3.7 environment
``test_gzip`` in the test suite, and only in the python 3.8 environment
that has ``requests`` installed.
Also, in order for the boto tests to run, you will need an AWS key.
@@ -130,17 +130,17 @@ in this example::
pip3 install tox tox-pyenv
# Install supported versions (at time of writing), this does not activate them
pyenv install 3.7.5 3.8.0 pypy3.8
pyenv install 3.8.0 pypy3.8
# This activates them
pyenv local 3.7.5 3.8.0 pypy3.8
pyenv local 3.8.0 pypy3.8
# Run the whole test suite
tox
# Run the whole test suite or just part of it
tox -e lint
tox -e py37-requests
tox -e py38-requests
Troubleshooting on MacOSX

View File

@@ -9,7 +9,7 @@ with pip::
Compatibility
-------------
VCR.py supports Python 3.7+, and `pypy <http://pypy.org>`__.
VCR.py supports Python 3.8+, and `pypy <http://pypy.org>`__.
The following HTTP libraries are supported:

View File

@@ -8,7 +8,7 @@ import sys
from setuptools import find_packages, setup
from setuptools.command.test import test as TestCommand
long_description = open("README.rst", "r").read()
long_description = open("README.rst").read()
here = os.path.abspath(os.path.dirname(__file__))
@@ -85,7 +85,7 @@ setup(
author_email="me@kevinmccarthy.org",
url="https://github.com/kevin1024/vcrpy",
packages=find_packages(exclude=["tests*"]),
python_requires=">=3.7",
python_requires=">=3.8",
install_requires=install_requires,
license="MIT",
tests_require=tests_require,
@@ -95,7 +95,6 @@ setup(
"Intended Audience :: Developers",
"Programming Language :: Python",
"Programming Language :: Python :: 3",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",

View File

@@ -11,9 +11,10 @@ def assert_cassette_has_one_response(cass):
assert cass.play_count == 1
def assert_is_json(a_string):
def assert_is_json_bytes(b: bytes):
assert isinstance(b, bytes)
try:
json.loads(a_string.decode("utf-8"))
json.loads(b.decode("utf-8"))
except Exception:
assert False
assert True

View File

@@ -151,7 +151,7 @@ def test_post(tmpdir, body, caplog, mockbin_request_url):
(
log
for log in caplog.records
if log.getMessage() == "<Request (POST) {}> not in cassette, sending to real server".format(url)
if log.getMessage() == f"<Request (POST) {url}> not in cassette, sending to real server"
),
None,
), "Log message not found."

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
"""Basic tests for cassettes"""
# External imports

View File

@@ -20,12 +20,12 @@ except ImportError:
# https://github.com/boto/botocore/pull/1495
boto3_skip_vendored_requests = pytest.mark.skipif(
botocore_awsrequest,
reason="botocore version {ver} does not use vendored requests anymore.".format(ver=botocore.__version__),
reason=f"botocore version {botocore.__version__} does not use vendored requests anymore.",
)
boto3_skip_awsrequest = pytest.mark.skipif(
not botocore_awsrequest,
reason="botocore version {ver} still uses vendored requests.".format(ver=botocore.__version__),
reason=f"botocore version {botocore.__version__} still uses vendored requests.",
)
IAM_USER_NAME = "vcrpy"

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
"""Basic tests about save behavior"""
# External imports

View File

@@ -5,7 +5,7 @@ from urllib.parse import urlencode
from urllib.request import Request, urlopen
import pytest
from assertions import assert_cassette_has_one_response, assert_is_json
from assertions import assert_cassette_has_one_response, assert_is_json_bytes
import vcr
@@ -105,7 +105,7 @@ def test_decompress_gzip(tmpdir, httpbin):
with vcr.use_cassette(cass_file) as cass:
decoded_response = urlopen(url).read()
assert_cassette_has_one_response(cass)
assert_is_json(decoded_response)
assert_is_json_bytes(decoded_response)
def test_decomptess_empty_body(tmpdir, httpbin):
@@ -129,7 +129,7 @@ def test_decompress_deflate(tmpdir, httpbin):
with vcr.use_cassette(cass_file) as cass:
decoded_response = urlopen(url).read()
assert_cassette_has_one_response(cass)
assert_is_json(decoded_response)
assert_is_json_bytes(decoded_response)
def test_decompress_regular(tmpdir, httpbin):
@@ -141,7 +141,7 @@ def test_decompress_regular(tmpdir, httpbin):
with vcr.use_cassette(cass_file) as cass:
resp = urlopen(url).read()
assert_cassette_has_one_response(cass)
assert_is_json(resp)
assert_is_json_bytes(resp)
def test_before_record_request_corruption(tmpdir, httpbin):

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
"""Integration tests with httplib2"""
from urllib.parse import urlencode

View File

@@ -28,9 +28,9 @@ def test_ignore_localhost(tmpdir, httpbin):
with overridden_dns({"httpbin.org": "127.0.0.1"}):
cass_file = str(tmpdir.join("filter_qs.yaml"))
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
urlopen("http://localhost:{}/".format(httpbin.port))
urlopen(f"http://localhost:{httpbin.port}/")
assert len(cass) == 0
urlopen("http://httpbin.org:{}/".format(httpbin.port))
urlopen(f"http://httpbin.org:{httpbin.port}/")
assert len(cass) == 1
@@ -38,9 +38,9 @@ def test_ignore_httpbin(tmpdir, httpbin):
with overridden_dns({"httpbin.org": "127.0.0.1"}):
cass_file = str(tmpdir.join("filter_qs.yaml"))
with vcr.use_cassette(cass_file, ignore_hosts=["httpbin.org"]) as cass:
urlopen("http://httpbin.org:{}/".format(httpbin.port))
urlopen(f"http://httpbin.org:{httpbin.port}/")
assert len(cass) == 0
urlopen("http://localhost:{}/".format(httpbin.port))
urlopen(f"http://localhost:{httpbin.port}/")
assert len(cass) == 1
@@ -48,8 +48,8 @@ def test_ignore_localhost_and_httpbin(tmpdir, httpbin):
with overridden_dns({"httpbin.org": "127.0.0.1"}):
cass_file = str(tmpdir.join("filter_qs.yaml"))
with vcr.use_cassette(cass_file, ignore_hosts=["httpbin.org"], ignore_localhost=True) as cass:
urlopen("http://httpbin.org:{}".format(httpbin.port))
urlopen("http://localhost:{}".format(httpbin.port))
urlopen(f"http://httpbin.org:{httpbin.port}")
urlopen(f"http://localhost:{httpbin.port}")
assert len(cass) == 0
@@ -57,12 +57,12 @@ def test_ignore_localhost_twice(tmpdir, httpbin):
with overridden_dns({"httpbin.org": "127.0.0.1"}):
cass_file = str(tmpdir.join("filter_qs.yaml"))
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
urlopen("http://localhost:{}".format(httpbin.port))
urlopen(f"http://localhost:{httpbin.port}")
assert len(cass) == 0
urlopen("http://httpbin.org:{}".format(httpbin.port))
urlopen(f"http://httpbin.org:{httpbin.port}")
assert len(cass) == 1
with vcr.use_cassette(cass_file, ignore_localhost=True) as cass:
assert len(cass) == 1
urlopen("http://localhost:{}".format(httpbin.port))
urlopen("http://httpbin.org:{}".format(httpbin.port))
urlopen(f"http://localhost:{httpbin.port}")
urlopen(f"http://httpbin.org:{httpbin.port}")
assert len(cass) == 1

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
"""Test using a proxy."""
import http.server

View File

@@ -1,13 +1,14 @@
# -*- coding: utf-8 -*-
"""Tests for cassettes with custom persistence"""
# External imports
import os
from urllib.request import urlopen
import pytest
# Internal imports
import vcr
from vcr.persisters.filesystem import FilesystemPersister
from vcr.persisters.filesystem import CassetteDecodeError, CassetteNotFoundError, FilesystemPersister
class CustomFilesystemPersister:
@@ -25,6 +26,19 @@ class CustomFilesystemPersister:
FilesystemPersister.save_cassette(cassette_path, cassette_dict, serializer)
class BadPersister(FilesystemPersister):
"""A bad persister that raises different errors."""
@staticmethod
def load_cassette(cassette_path, serializer):
if "nonexistent" in cassette_path:
raise CassetteNotFoundError()
elif "encoding" in cassette_path:
raise CassetteDecodeError()
else:
raise ValueError("buggy persister")
def test_save_cassette_with_custom_persister(tmpdir, httpbin):
"""Ensure you can save a cassette using custom persister"""
my_vcr = vcr.VCR()
@@ -53,3 +67,22 @@ def test_load_cassette_with_custom_persister(tmpdir, httpbin):
with my_vcr.use_cassette(test_fixture, serializer="json"):
response = urlopen(httpbin.url).read()
assert b"difficult sometimes" in response
def test_load_cassette_persister_exception_handling(tmpdir, httpbin):
"""
Ensure expected errors from persister are swallowed while unexpected ones
are passed up the call stack.
"""
my_vcr = vcr.VCR()
my_vcr.register_persister(BadPersister)
with my_vcr.use_cassette("bad/nonexistent") as cass:
assert len(cass) == 0
with my_vcr.use_cassette("bad/encoding") as cass:
assert len(cass) == 0
with pytest.raises(ValueError):
with my_vcr.use_cassette("bad/buggy") as cass:
pass

View File

@@ -1,6 +1,6 @@
"""Test requests' interaction with vcr"""
import pytest
from assertions import assert_cassette_empty, assert_is_json
from assertions import assert_cassette_empty, assert_is_json_bytes
import vcr
@@ -114,22 +114,6 @@ def test_post_chunked_binary(tmpdir, httpbin):
assert req1 == req2
@pytest.mark.skipif("sys.version_info >= (3, 6)", strict=True, raises=ConnectionError)
def test_post_chunked_binary_secure(tmpdir, httpbin_secure):
"""Ensure that we can send chunked binary without breaking while trying to concatenate bytes with str."""
data1 = iter([b"data", b"to", b"send"])
data2 = iter([b"data", b"to", b"send"])
url = httpbin_secure.url + "/post"
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req1 = requests.post(url, data1).content
print(req1)
with vcr.use_cassette(str(tmpdir.join("requests.yaml"))):
req2 = requests.post(url, data2).content
assert req1 == req2
def test_redirects(tmpdir, httpbin_both):
"""Ensure that we can handle redirects"""
url = httpbin_both + "/redirect-to?url=bytes/1024"
@@ -176,7 +160,7 @@ def test_gzip__decode_compressed_response_false(tmpdir, httpbin_both):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
response = requests.get(httpbin_both + "/gzip")
assert response.headers["content-encoding"] == "gzip" # i.e. not removed
assert_is_json(response.content) # i.e. uncompressed bytes
assert_is_json_bytes(response.content) # i.e. uncompressed bytes
def test_gzip__decode_compressed_response_true(tmpdir, httpbin_both):

View File

@@ -2,7 +2,7 @@ import http.client as httplib
import json
import zlib
from assertions import assert_is_json
from assertions import assert_is_json_bytes
import vcr
@@ -84,7 +84,7 @@ def test_original_decoded_response_is_not_modified(tmpdir, httpbin):
inside = conn.getresponse()
assert "content-encoding" not in inside.headers
assert_is_json(inside.read())
assert_is_json_bytes(inside.read())
def _make_before_record_response(fields, replacement="[REDACTED]"):

View File

@@ -1,10 +1,9 @@
# -*- coding: utf-8 -*-
"""Test requests' interaction with vcr"""
import json
import pytest
from assertions import assert_cassette_empty, assert_is_json
from assertions import assert_cassette_empty, assert_is_json_bytes
import vcr
from vcr.errors import CannotOverwriteExistingCassetteException
@@ -195,11 +194,11 @@ def test_gzip(get_client, tmpdir, scheme):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
response = yield get(get_client(), url, **kwargs)
assert_is_json(response.body)
assert_is_json_bytes(response.body)
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))) as cass:
response = yield get(get_client(), url, **kwargs)
assert_is_json(response.body)
assert_is_json_bytes(response.body)
assert 1 == cass.play_count

View File

@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
"""Integration tests with urllib2"""
import ssl

View File

@@ -4,7 +4,7 @@
import pytest
import pytest_httpbin
from assertions import assert_cassette_empty, assert_is_json
from assertions import assert_cassette_empty, assert_is_json_bytes
import vcr
from vcr.patch import force_reset
@@ -136,10 +136,10 @@ def test_gzip(tmpdir, httpbin_both, verify_pool_mgr):
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
response = verify_pool_mgr.request("GET", url)
assert_is_json(response.data)
assert_is_json_bytes(response.data)
with vcr.use_cassette(str(tmpdir.join("gzip.yaml"))):
assert_is_json(response.data)
assert_is_json_bytes(response.data)
def test_https_with_cert_validation_disabled(tmpdir, httpbin_secure, pool_mgr):

View File

@@ -29,6 +29,19 @@ def test_cassette_load(tmpdir):
assert len(a_cassette) == 1
def test_cassette_load_nonexistent():
a_cassette = Cassette.load(path="something/nonexistent.yml")
assert len(a_cassette) == 0
def test_cassette_load_invalid_encoding(tmpdir):
a_file = tmpdir.join("invalid_encoding.yml")
with open(a_file, "wb") as fd:
fd.write(b"\xda")
a_cassette = Cassette.load(path=str(a_file))
assert len(a_cassette) == 0
def test_cassette_not_played():
a = Cassette("test")
assert not a.play_count

View File

@@ -17,9 +17,9 @@ def test_try_migrate_with_json(tmpdir):
cassette = tmpdir.join("cassette.json").strpath
shutil.copy("tests/fixtures/migration/old_cassette.json", cassette)
assert vcr.migration.try_migrate(cassette)
with open("tests/fixtures/migration/new_cassette.json", "r") as f:
with open("tests/fixtures/migration/new_cassette.json") as f:
expected_json = json.load(f)
with open(cassette, "r") as f:
with open(cassette) as f:
actual_json = json.load(f)
assert actual_json == expected_json
@@ -28,9 +28,9 @@ def test_try_migrate_with_yaml(tmpdir):
cassette = tmpdir.join("cassette.yaml").strpath
shutil.copy("tests/fixtures/migration/old_cassette.yaml", cassette)
assert vcr.migration.try_migrate(cassette)
with open("tests/fixtures/migration/new_cassette.yaml", "r") as f:
with open("tests/fixtures/migration/new_cassette.yaml") as f:
expected_yaml = yaml.load(f, Loader=Loader)
with open(cassette, "r") as f:
with open(cassette) as f:
actual_yaml = yaml.load(f, Loader=Loader)
assert actual_yaml == expected_yaml

View File

@@ -1,4 +1,3 @@
# coding: UTF-8
import io
from vcr.stubs import VCRHTTPResponse

View File

@@ -1,4 +1,3 @@
# -*- encoding: utf-8 -*-
from unittest import mock
import pytest
@@ -9,24 +8,24 @@ from vcr.serializers import compat, jsonserializer, yamlserializer
def test_deserialize_old_yaml_cassette():
with open("tests/fixtures/migration/old_cassette.yaml", "r") as f:
with open("tests/fixtures/migration/old_cassette.yaml") as f:
with pytest.raises(ValueError):
deserialize(f.read(), yamlserializer)
def test_deserialize_old_json_cassette():
with open("tests/fixtures/migration/old_cassette.json", "r") as f:
with open("tests/fixtures/migration/old_cassette.json") as f:
with pytest.raises(ValueError):
deserialize(f.read(), jsonserializer)
def test_deserialize_new_yaml_cassette():
with open("tests/fixtures/migration/new_cassette.yaml", "r") as f:
with open("tests/fixtures/migration/new_cassette.yaml") as f:
deserialize(f.read(), yamlserializer)
def test_deserialize_new_json_cassette():
with open("tests/fixtures/migration/new_cassette.json", "r") as f:
with open("tests/fixtures/migration/new_cassette.json") as f:
deserialize(f.read(), jsonserializer)

18
tox.ini
View File

@@ -3,7 +3,7 @@ skip_missing_interpreters=true
envlist =
cov-clean,
lint,
{py37,py38,py39,py310,py311}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3,aiohttp,httpx},
{py38,py39,py310,py311}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3,aiohttp,httpx},
{py310,py311}-{requests-urllib3-2,urllib3-2},
{pypy3}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3},
{py310}-httpx019,
@@ -12,7 +12,6 @@ envlist =
[gh-actions]
python =
3.7: py37
3.8: py38
3.9: py39
3.10: py310, lint
@@ -66,9 +65,9 @@ deps =
# In other circumstances, we might want to generate a PDF or an ebook
commands =
sphinx-build -W -b html -d {envtmpdir}/doctrees . {envtmpdir}/html
# We use Python 3.7. Tox sometimes tries to autodetect it based on the name of
# We use Python 3.8. Tox sometimes tries to autodetect it based on the name of
# the testenv, but "docs" does not give useful clues so we have to be explicit.
basepython = python3.7
basepython = python3.8
[testenv]
# Need to use develop install so that paths
@@ -94,15 +93,14 @@ deps =
aiohttp: pytest-asyncio
aiohttp: pytest-aiohttp
httpx: httpx
{py37,py38,py39,py310}-{httpx}: httpx
{py37,py38,py39,py310}-{httpx}: pytest-asyncio
{py38,py39,py310}-{httpx}: httpx
{py38,py39,py310}-{httpx}: pytest-asyncio
httpx: httpx>0.19
# httpx==0.19 is the latest version that supports allow_redirects, newer versions use follow_redirects
httpx019: httpx==0.19
{py37,py38,py39,py310}-{httpx}: pytest-asyncio
{py38,py39,py310}-{httpx}: pytest-asyncio
depends =
lint,{py37,py38,py39,py310,py311,pypy3}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3},{py310,py311}-{requests-urllib3-2,urllib3-2},{py37,py38,py39,py310,py311}-{aiohttp},{py37,py38,py39,py310,py311}-{httpx}: cov-clean
cov-report: lint,{py37,py38,py39,py310,py311,pypy3}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3},{py310,py311}-{requests-urllib3-2,urllib3-2},{py37,py38,py39,py310,py311}-{aiohttp}
lint,{py38,py39,py310,py311,pypy3}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3},{py310,py311}-{requests-urllib3-2,urllib3-2},{py38,py39,py310,py311}-{aiohttp},{py38,py39,py310,py311}-{httpx}: cov-clean
cov-report: lint,{py38,py39,py310,py311,pypy3}-{requests-urllib3-1,httplib2,urllib3-1,tornado4,boto3},{py310,py311}-{requests-urllib3-2,urllib3-2},{py38,py39,py310,py311}-{aiohttp}
passenv =
AWS_ACCESS_KEY_ID
AWS_DEFAULT_REGION

View File

@@ -4,7 +4,7 @@ from logging import NullHandler
from .config import VCR
from .record_mode import RecordMode as mode # noqa import is not used in this file
__version__ = "4.4.0"
__version__ = "5.0.0"
logging.getLogger(__name__).addHandler(NullHandler())

View File

@@ -11,7 +11,7 @@ from ._handle_coroutine import handle_coroutine
from .errors import UnhandledHTTPRequestError
from .matchers import get_matchers_results, method, requests_match, uri
from .patch import CassettePatcherBuilder
from .persisters.filesystem import FilesystemPersister
from .persisters.filesystem import CassetteDecodeError, CassetteNotFoundError, FilesystemPersister
from .record_mode import RecordMode
from .serializers import yamlserializer
from .util import partition_dict
@@ -280,7 +280,7 @@ class Cassette:
return response
# The cassette doesn't contain the request asked for.
raise UnhandledHTTPRequestError(
"The cassette (%r) doesn't contain the request (%r) asked for" % (self._path, request)
f"The cassette ({self._path!r}) doesn't contain the request ({request!r}) asked for"
)
def responses_of(self, request):
@@ -295,7 +295,7 @@ class Cassette:
return responses
# The cassette doesn't contain the request asked for.
raise UnhandledHTTPRequestError(
"The cassette (%r) doesn't contain the request (%r) asked for" % (self._path, request)
f"The cassette ({self._path!r}) doesn't contain the request ({request!r}) asked for"
)
def rewind(self):
@@ -352,11 +352,11 @@ class Cassette:
self.append(request, response)
self.dirty = False
self.rewound = True
except ValueError:
except (CassetteDecodeError, CassetteNotFoundError):
pass
def __str__(self):
return "<Cassette containing {} recorded response(s)>".format(len(self))
return f"<Cassette containing {len(self)} recorded response(s)>"
def __len__(self):
"""Return the number of request,response pairs stored in here"""

View File

@@ -88,7 +88,7 @@ class VCR:
try:
serializer = self.serializers[serializer_name]
except KeyError:
raise KeyError("Serializer {} doesn't exist or isn't registered".format(serializer_name))
raise KeyError(f"Serializer {serializer_name} doesn't exist or isn't registered")
return serializer
def _get_matchers(self, matcher_names):
@@ -97,7 +97,7 @@ class VCR:
for m in matcher_names:
matchers.append(self.matchers[m])
except KeyError:
raise KeyError("Matcher {} doesn't exist or isn't registered".format(m))
raise KeyError(f"Matcher {m} doesn't exist or isn't registered")
return matchers
def use_cassette(self, path=None, **kwargs):

View File

@@ -10,37 +10,37 @@ log = logging.getLogger(__name__)
def method(r1, r2):
if r1.method != r2.method:
raise AssertionError("{} != {}".format(r1.method, r2.method))
raise AssertionError(f"{r1.method} != {r2.method}")
def uri(r1, r2):
if r1.uri != r2.uri:
raise AssertionError("{} != {}".format(r1.uri, r2.uri))
raise AssertionError(f"{r1.uri} != {r2.uri}")
def host(r1, r2):
if r1.host != r2.host:
raise AssertionError("{} != {}".format(r1.host, r2.host))
raise AssertionError(f"{r1.host} != {r2.host}")
def scheme(r1, r2):
if r1.scheme != r2.scheme:
raise AssertionError("{} != {}".format(r1.scheme, r2.scheme))
raise AssertionError(f"{r1.scheme} != {r2.scheme}")
def port(r1, r2):
if r1.port != r2.port:
raise AssertionError("{} != {}".format(r1.port, r2.port))
raise AssertionError(f"{r1.port} != {r2.port}")
def path(r1, r2):
if r1.path != r2.path:
raise AssertionError("{} != {}".format(r1.path, r2.path))
raise AssertionError(f"{r1.path} != {r2.path}")
def query(r1, r2):
if r1.query != r2.query:
raise AssertionError("{} != {}".format(r1.query, r2.query))
raise AssertionError(f"{r1.query} != {r2.query}")
def raw_body(r1, r2):
@@ -59,7 +59,7 @@ def body(r1, r2):
def headers(r1, r2):
if r1.headers != r2.headers:
raise AssertionError("{} != {}".format(r1.headers, r2.headers))
raise AssertionError(f"{r1.headers} != {r2.headers}")
def _header_checker(value, header="Content-Type"):
@@ -107,7 +107,7 @@ def _get_transformer(request):
def requests_match(r1, r2, matchers):
successes, failures = get_matchers_results(r1, r2, matchers)
if failures:
log.debug("Requests {} and {} differ.\n" "Failure details:\n" "{}".format(r1, r2, failures))
log.debug(f"Requests {r1} and {r2} differ.\nFailure details:\n{failures}")
return len(failures) == 0

View File

@@ -55,7 +55,7 @@ def build_uri(**parts):
port = parts["port"]
scheme = parts["protocol"]
default_port = {"https": 443, "http": 80}[scheme]
parts["port"] = ":{}".format(port) if port != default_port else ""
parts["port"] = f":{port}" if port != default_port else ""
return "{protocol}://{host}{port}{path}".format(**parts)
@@ -118,7 +118,7 @@ def migrate(file_path, migration_fn):
# because we assume that original files can be reverted
# we will try to copy the content. (os.rename not needed)
with tempfile.TemporaryFile(mode="w+") as out_fp:
with open(file_path, "r") as in_fp:
with open(file_path) as in_fp:
if not migration_fn(in_fp, out_fp):
return False
with open(file_path, "w") as in_fp:
@@ -150,7 +150,7 @@ def main():
for file_path in files:
migrated = try_migrate(file_path)
status = "OK" if migrated else "FAIL"
sys.stderr.write("[{}] {}\n".format(status, file_path))
sys.stderr.write(f"[{status}] {file_path}\n")
sys.stderr.write("Done.\n")

View File

@@ -186,9 +186,7 @@ class CassettePatcherBuilder:
bases = (base_class,)
if not issubclass(base_class, object): # Check for old style class
bases += (object,)
return type(
"{}{}".format(base_class.__name__, self._cassette._path), bases, dict(cassette=self._cassette)
)
return type(f"{base_class.__name__}{self._cassette._path}", bases, dict(cassette=self._cassette))
@_build_patchers_from_mock_triples_decorator
def _httplib(self):

View File

@@ -5,17 +5,25 @@ from pathlib import Path
from ..serialize import deserialize, serialize
class CassetteNotFoundError(FileNotFoundError):
pass
class CassetteDecodeError(ValueError):
pass
class FilesystemPersister:
@classmethod
def load_cassette(cls, cassette_path, serializer):
cassette_path = Path(cassette_path) # if cassette path is already Path this is no operation
if not cassette_path.is_file():
raise ValueError("Cassette not found.")
raise CassetteNotFoundError()
try:
with cassette_path.open() as f:
data = f.read()
except UnicodeEncodeError as err:
raise ValueError("Can't read Cassette, Encoding is broken") from err
except UnicodeDecodeError as err:
raise CassetteDecodeError("Can't read Cassette, Encoding is broken") from err
return deserialize(data, serializer)

View File

@@ -90,7 +90,7 @@ class Request:
return self.scheme
def __str__(self):
return "<Request ({}) {}>".format(self.method, self.uri)
return f"<Request ({self.method}) {self.uri}>"
def __repr__(self):
return self.__str__()

View File

@@ -188,26 +188,26 @@ class VCRConnection:
"""
port = self.real_connection.port
default_port = {"https": 443, "http": 80}[self._protocol]
return ":{}".format(port) if port != default_port else ""
return f":{port}" if port != default_port else ""
def _uri(self, url):
"""Returns request absolute URI"""
if url and not url.startswith("/"):
# Then this must be a proxy request.
return url
uri = "{}://{}{}{}".format(self._protocol, self.real_connection.host, self._port_postfix(), url)
uri = f"{self._protocol}://{self.real_connection.host}{self._port_postfix()}{url}"
log.debug("Absolute URI: %s", uri)
return uri
def _url(self, uri):
"""Returns request selector url from absolute URI"""
prefix = "{}://{}{}".format(self._protocol, self.real_connection.host, self._port_postfix())
prefix = f"{self._protocol}://{self.real_connection.host}{self._port_postfix()}"
return uri.replace(prefix, "", 1)
def request(self, method, url, body=None, headers=None, *args, **kwargs):
"""Persist the request metadata in self._vcr_request"""
self._vcr_request = Request(method=method, uri=self._uri(url), body=body, headers=headers or {})
log.debug("Got {}".format(self._vcr_request))
log.debug(f"Got {self._vcr_request}")
# Note: The request may not actually be finished at this point, so
# I'm not sending the actual request until getresponse(). This
@@ -223,7 +223,7 @@ class VCRConnection:
of putheader() calls.
"""
self._vcr_request = Request(method=method, uri=self._uri(url), body="", headers={})
log.debug("Got {}".format(self._vcr_request))
log.debug(f"Got {self._vcr_request}")
def putheader(self, header, *values):
self._vcr_request.headers[header] = values
@@ -255,7 +255,7 @@ class VCRConnection:
# Check to see if the cassette has a response for this request. If so,
# then return it
if self.cassette.can_play_response_for(self._vcr_request):
log.info("Playing response for {} from cassette".format(self._vcr_request))
log.info(f"Playing response for {self._vcr_request} from cassette")
response = self.cassette.play_response(self._vcr_request)
return VCRHTTPResponse(response)
else:
@@ -267,7 +267,7 @@ class VCRConnection:
# Otherwise, we should send the request, then get the response
# and return it.
log.info("{} not in cassette, sending to real server".format(self._vcr_request))
log.info(f"{self._vcr_request} not in cassette, sending to real server")
# This is imported here to avoid circular import.
# TODO(@IvanMalison): Refactor to allow normal import.
from vcr.patch import force_reset

View File

@@ -261,7 +261,7 @@ def vcr_request(cassette, real_request):
vcr_request = Request(method, str(request_url), data, _serialize_headers(headers))
if cassette.can_play_response_for(vcr_request):
log.info("Playing response for {} from cassette".format(vcr_request))
log.info(f"Playing response for {vcr_request} from cassette")
response = play_responses(cassette, vcr_request, kwargs)
for redirect in response.history:
self._cookie_jar.update_cookies(redirect.cookies, redirect.url)

View File

@@ -32,7 +32,7 @@ class VCRMixin:
return os.path.join(testdir, "cassettes")
def _get_cassette_name(self):
return "{0}.{1}.yaml".format(self.__class__.__name__, self._testMethodName)
return f"{self.__class__.__name__}.{self._testMethodName}.yaml"
class VCRTestCase(VCRMixin, unittest.TestCase):

View File

@@ -1,9 +1,5 @@
import types
try:
from collections.abc import Mapping, MutableMapping
except ImportError:
from collections import Mapping, MutableMapping
from collections.abc import Mapping, MutableMapping
# Shamelessly stolen from https://github.com/kennethreitz/requests/blob/master/requests/structures.py