<s> import gevent <EOL> import pytest <EOL> from collections import namedtuple <EOL> try : <EOL> from azure . storage . blob import BlobService <EOL> except ImportError : <EOL> from azure . storage import BlobService <EOL> from fast_wait import fast_wait <EOL> from gevent import coros <EOL> from wal_e import exception <EOL> from wal_e . worker . wabs import wabs_deleter <EOL> assert fast_wait <EOL> B = namedtuple ( '<STR_LIT>' , [ '<STR_LIT:name>' ] ) <EOL> class ContainerDeleteKeysCollector ( object ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self ) : <EOL> self . deleted_keys = [ ] <EOL> self . aborted_keys = [ ] <EOL> self . exc = None <EOL> self . _exc_protect = coros . RLock ( ) <EOL> def inject ( self , exc ) : <EOL> self . _exc_protect . acquire ( ) <EOL> self . exc = exc <EOL> self . _exc_protect . release ( ) <EOL> def __call__ ( self , container , key ) : <EOL> self . _exc_protect . acquire ( ) <EOL> try : <EOL> if self . exc : <EOL> self . aborted_keys . append ( key ) <EOL> gevent . sleep ( <NUM_LIT:0.1> ) <EOL> raise self . exc <EOL> finally : <EOL> self . _exc_protect . release ( ) <EOL> self . deleted_keys . append ( key ) <EOL> @ pytest . fixture <EOL> def collect ( monkeypatch ) : <EOL> """<STR_LIT>""" <EOL> collect = ContainerDeleteKeysCollector ( ) <EOL> monkeypatch . setattr ( BlobService , '<STR_LIT>' , collect ) <EOL> return collect <EOL> def test_fast_wait ( ) : <EOL> """<STR_LIT>""" <EOL> gevent . sleep ( <NUM_LIT> ) <EOL> def test_construction ( ) : <EOL> """<STR_LIT>""" <EOL> wabs_deleter . Deleter ( '<STR_LIT:test>' , '<STR_LIT>' ) <EOL> def test_close_error ( ) : <EOL> """<STR_LIT>""" <EOL> d = wabs_deleter . Deleter ( BlobService ( '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> d . close ( ) <EOL> with pytest . raises ( exception . UserCritical ) : <EOL> d . delete ( '<STR_LIT>' ) <EOL> def test_processes_one_deletion ( collect ) : <EOL> key_name = '<STR_LIT>' <EOL> b = B ( name = key_name ) <EOL> d = wabs_deleter . Deleter ( BlobService ( '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> d . delete ( b ) <EOL> d . close ( ) <EOL> assert collect . deleted_keys == [ key_name ] <EOL> def test_processes_many_deletions ( collect ) : <EOL> target = sorted ( [ '<STR_LIT>' + str ( x ) for x in range ( <NUM_LIT> ) ] ) <EOL> blobs = [ B ( name = key_name ) for key_name in target ] <EOL> d = wabs_deleter . Deleter ( BlobService ( '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> for b in blobs : <EOL> d . delete ( b ) <EOL> d . close ( ) <EOL> assert sorted ( collect . deleted_keys ) == target <EOL> def test_retry_on_normal_error ( collect ) : <EOL> """<STR_LIT>""" <EOL> key_name = '<STR_LIT>' <EOL> b = B ( name = key_name ) <EOL> collect . inject ( Exception ( '<STR_LIT>' ) ) <EOL> d = wabs_deleter . Deleter ( BlobService ( '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> d . delete ( b ) <EOL> while len ( collect . aborted_keys ) < <NUM_LIT:2> : <EOL> gevent . sleep ( <NUM_LIT:0.1> ) <EOL> assert not collect . deleted_keys <EOL> collect . inject ( None ) <EOL> d . close ( ) <EOL> assert collect . deleted_keys == [ key_name ] <EOL> def test_no_retry_on_keyboadinterrupt ( collect ) : <EOL> """<STR_LIT>""" <EOL> key_name = '<STR_LIT>' <EOL> b = B ( name = key_name ) <EOL> class MarkedKeyboardInterrupt ( KeyboardInterrupt ) : <EOL> pass <EOL> collect . inject ( MarkedKeyboardInterrupt ( '<STR_LIT>' ) ) <EOL> d = wabs_deleter . Deleter ( BlobService ( '<STR_LIT:test>' , '<STR_LIT>' ) , '<STR_LIT>' ) <EOL> with pytest . raises ( MarkedKeyboardInterrupt ) : <EOL> d . delete ( b ) <EOL> while True : <EOL> gevent . sleep ( <NUM_LIT:0.1> ) <EOL> assert collect . aborted_keys == [ key_name ] <EOL> collect . inject ( None ) <EOL> d . close ( ) <EOL> assert not collect . deleted_keys </s>
