/usr/lib/python2.7/dist-packages/betamax/recorder.py is in python-betamax 0.5.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 | # -*- coding: utf-8 -*-
from . import matchers, serializers
from .adapter import BetamaxAdapter
from .cassette import Cassette
from .configure import Configuration
from .options import Options
class Betamax(object):
"""This object contains the main API of the request-vcr library.
This object is entirely a context manager so all you have to do is:
.. code::
s = requests.Session()
with Betamax(s) as vcr:
vcr.use_cassette('example')
r = s.get('https://httpbin.org/get')
Or more concisely, you can do:
.. code::
s = requests.Session()
with Betamax(s).use_cassette('example') as vcr:
r = s.get('https://httpbin.org/get')
This object allows for the user to specify the cassette library directory
and default cassette options.
.. code::
s = requests.Session()
with Betamax(s, cassette_library_dir='tests/cassettes') as vcr:
vcr.use_cassette('example')
r = s.get('https://httpbin.org/get')
with Betamax(s, default_cassette_options={
're_record_interval': 1000
}) as vcr:
vcr.use_cassette('example')
r = s.get('https://httpbin.org/get')
"""
def __init__(self, session, cassette_library_dir=None,
default_cassette_options={}):
#: Store the requests.Session object being wrapped.
self.session = session
#: Store the session's original adapters.
self.http_adapters = session.adapters.copy()
#: Create a new adapter to replace the existing ones
self.betamax_adapter = BetamaxAdapter(old_adapters=self.http_adapters)
# We need a configuration instance to make life easier
self.config = Configuration()
# Merge the new cassette options with the default ones
self.config.default_cassette_options.update(
default_cassette_options or {}
)
# If it was passed in, use that instead.
if cassette_library_dir:
self.config.cassette_library_dir = cassette_library_dir
def __enter__(self):
self.start()
return self
def __exit__(self, *ex_args):
self.stop()
# ex_args comes through as the exception type, exception value and
# exception traceback. If any of them are not None, we should probably
# try to raise the exception and not muffle anything.
if any(ex_args):
# If you return False, Python will re-raise the exception for you
return False
@staticmethod
def configure():
"""Help to configure the library as a whole.
.. code::
with Betamax.configure() as config:
config.cassette_library_dir = 'tests/cassettes/'
config.default_cassette_options['match_options'] = [
'method', 'uri', 'headers'
]
"""
return Configuration()
@property
def current_cassette(self):
"""Return the cassette that is currently in use.
:returns: :class:`Cassette <betamax.cassette.Cassette>`
"""
return self.betamax_adapter.cassette
@staticmethod
def register_request_matcher(matcher_class):
"""Register a new request matcher.
:param matcher_class: (required), this must sub-class
:class:`BaseMatcher <betamax.matchers.BaseMatcher>`
"""
matchers.matcher_registry[matcher_class.name] = matcher_class()
@staticmethod
def register_serializer(serializer_class):
"""Register a new serializer.
:param matcher_class: (required), this must sub-class
:class:`BaseSerializer <betamax.serializers.BaseSerializer>`
"""
name = serializer_class.name
serializers.serializer_registry[name] = serializer_class()
# ❙▸
def start(self):
"""Start recording or replaying interactions."""
for k in self.http_adapters:
self.session.mount(k, self.betamax_adapter)
# ■
def stop(self):
"""Stop recording or replaying interactions."""
# No need to keep the cassette in memory any longer.
self.betamax_adapter.eject_cassette()
# On exit, we no longer wish to use our adapter and we want the
# session to behave normally! Woooo!
self.betamax_adapter.close()
for (k, v) in self.http_adapters.items():
self.session.mount(k, v)
def use_cassette(self, cassette_name, **kwargs):
"""Tell Betamax which cassette you wish to use for the context.
:param str cassette_name: relative name, without the serialization
format, of the cassette you wish Betamax would use
:param str serialize_with: the format you want Betamax to serialize
the cassette with
:param str serialize: DEPRECATED the format you want Betamax to
serialize the request and response data to and from
"""
kwargs = Options(kwargs)
serialize = kwargs['serialize'] or kwargs['serialize_with']
kwargs['cassette_library_dir'] = self.config.cassette_library_dir
can_load = Cassette.can_be_loaded(
self.config.cassette_library_dir,
cassette_name,
serialize,
kwargs['record']
)
if can_load:
self.betamax_adapter.load_cassette(cassette_name, serialize,
kwargs)
else:
# If we're not recording or replaying an existing cassette, we
# should tell the user/developer that there is no cassette, only
# Zuul
raise ValueError('Cassette must have a valid name and may not be'
' None.')
return self
|