Skip to content

Commit

Permalink
make sure start/stop of VCRSystem is used in the right way, by wrapping
Browse files Browse the repository at this point in the history 
one big try/finally around vcr usage with a stop in the finally block

this makes sure that in all situations that might raise an exception
VCRSystem is always kapt in the expected state at the end
  • Loading branch information
@megies
megies committed Apr 6, 2017
1 parent c861629 commit b602658
Showing 1 changed file with 88 additions and 86 deletions.
174 changes: 88 additions & 86 deletions vcr/core.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -245,7 +245,6 @@ def replay_next(cls, name_got, args_got, kwargs_got):
msg = '\nExpected: {} {} {}\nGot: {} {} {}'.format(
name_expected, args_expected, kwargs_expected,
name_got, args_got, kwargs_got)
cls.clear_playlist()
raise VCRPlaybackOutgoingTrafficMismatch(msg)
return value_

Expand Down Expand Up @@ -516,12 +515,6 @@ def _vcr_inner(*args, **kwargs):
# execute decorated function without VCR
return func(*args, **kwargs)

# enable VCR
if debug:
system_debug = VCRSystem.debug
VCRSystem.debug = True
VCRSystem.start()

# prepare VCR tape
if func.__module__ == 'doctest':
source_filename = func.__self__._dt_test.filename
Expand Down Expand Up @@ -563,89 +556,98 @@ def _vcr_inner(*args, **kwargs):
# auto-generated file name
tape = os.path.join(path, '%s.%s.vcr' % (file_name, func_name))

# check for tape file and determine mode
if not (playback_only or VCRSystem.playback_only) and (
not os.path.isfile(tape) or
overwrite or VCRSystem.overwrite):
# record mode
if PY2:
msg = 'VCR records only in PY3 to be backward ' + \
'compatible with PY2 - skipping VCR mechanics for %s'
warnings.warn(msg % (func.__name__))
# disable VCR
VCRSystem.stop()
# execute decorated function without VCR
return func(*args, **kwargs)
if VCRSystem.debug:
print('\nVCR RECORDING (%s) ...' % (func_name))
VCRSystem.status = VCR_RECORD
# execute decorated function
value = func(*args, **kwargs)
# write to file
if len(VCRSystem.playlist) == 0:
msg = 'no socket activity - @vcr decorator unneeded for %s'
msg = msg % func.__name__
if VCRSystem.raise_if_not_needed:
raise Exception(msg)
else:
# enable VCR
if debug:
system_debug = VCRSystem.debug
VCRSystem.debug = True
VCRSystem.start()

# one big try/finally to make sure VCRSystem gets stopped before
# any exception is raised
try:
# check for tape file and determine mode
if not (playback_only or VCRSystem.playback_only) and (
not os.path.isfile(tape) or
overwrite or VCRSystem.overwrite):

# record mode
if PY2:
msg = ('VCR records only in PY3 to be backward '
'compatible with PY2 - skipping VCR mechanics '
'for {}').format(func.__name__)
warnings.warn(msg)
# disable VCR
VCRSystem.stop()
# execute decorated function without VCR
return func(*args, **kwargs)

if VCRSystem.debug:
print('\nVCR RECORDING (%s) ...' % (func_name))
VCRSystem.status = VCR_RECORD

# execute decorated function
value = func(*args, **kwargs)

# write tape to file
if len(VCRSystem.playlist):
# add source code hash as first item in playlist
sha256 = get_source_code_sha256(func)
VCRSystem.playlist.insert(0, sha256)
# remove existing tape
try:
os.remove(tape)
except OSError:
pass
with gzip.open(tape, 'wb') as fh:
pickle.dump(VCRSystem.playlist, fh, protocol=2)
# no network traffic recorded.. warn or raise
else:
msg = ('no socket activity - @vcr decorator unneeded '
'for {}').format(func.__name__)
if VCRSystem.raise_if_not_needed:
raise Exception(msg)
else:
warnings.warn(msg)
else:
# add source code hash as first item in playlist
sha256 = get_source_code_sha256(func)
VCRSystem.playlist.insert(0, sha256)
# remove existing tape
try:
os.remove(tape)
except OSError:
pass
with gzip.open(tape, 'wb') as fh:
pickle.dump(VCRSystem.playlist, fh, protocol=2)
else:
# playback mode
if VCRSystem.debug:
print('\nVCR PLAYBACK (%s) ...' % (func_name))
VCRSystem.status = VCR_PLAYBACK
# if playback is requested and tape is missing: raise!
if not os.path.exists(tape):
msg = 'Missing VCR tape file for playback: {}'.format(tape)
# reset before raising
if debug:
VCRSystem.debug = system_debug
VCRSystem.stop()
raise IOError(msg)
# load playlist
with gzip.open(tape, 'rb') as fh:
VCRSystem.playlist = pickle.load(fh)
if VCRSystem.debug:
print('Loaded playlist:')
print('SHA256: {}'.format(VCRSystem.playlist[0]))
for i, item in enumerate(VCRSystem.playlist[1:]):
print('{:3d}: {} {} {}'.format(i, *item))
# check if source code has changed
sha256_playlist = VCRSystem.playlist.pop(0)
if VCRSystem.raise_if_source_code_changed:
sha256 = get_source_code_sha256(func)
if sha256 != sha256_playlist:
msg = ('Source code of test routine has changed since '
'time when VCR tape was recorded '
'(file: {}).').format(tape)
raise VCRPlaybackSourceCodeChangedError(msg)
# playback mode
if VCRSystem.debug:
print('SHA256 sum of source code matches playlist')
# execute decorated function
try:
print('\nVCR PLAYBACK (%s) ...' % (func_name))
VCRSystem.status = VCR_PLAYBACK

# if playback is requested and tape is missing: raise!
if not os.path.exists(tape):
msg = ('Missing VCR tape file for playback: ' +
str(tape))
raise IOError(msg)
# load playlist
with gzip.open(tape, 'rb') as fh:
VCRSystem.playlist = pickle.load(fh)

if VCRSystem.debug:
print('Loaded playlist:')
print('SHA256: {}'.format(VCRSystem.playlist[0]))
for i, item in enumerate(VCRSystem.playlist[1:]):
print('{:3d}: {} {} {}'.format(i, *item))

# optional check if source code has changed
sha256_playlist = VCRSystem.playlist.pop(0)
if VCRSystem.raise_if_source_code_changed:
sha256 = get_source_code_sha256(func)
if sha256 != sha256_playlist:
msg = ('Source code of test routine has changed '
'since time when VCR tape was recorded '
'(file: {}).').format(tape)
raise VCRPlaybackSourceCodeChangedError(msg)
if VCRSystem.debug:
print('SHA256 sum of source code matches playlist')

# execute decorated function
value = func(*args, **kwargs)
except Exception:
# reset before raising
if debug:
VCRSystem.debug = system_debug
VCRSystem.stop()
raise

# disable VCR
if debug:
VCRSystem.debug = system_debug
VCRSystem.stop()
finally:
# disable VCR
if debug:
VCRSystem.debug = system_debug
VCRSystem.stop()

return value

Expand Down

0 comments on commit b602658

Please sign in to comment.