<s> import collections <EOL> import fcntl <EOL> import os <EOL> import shutil <EOL> import signal <EOL> import subprocess <EOL> import sys <EOL> import tempfile <EOL> import threading <EOL> import time <EOL> from oslo_config import cfg <EOL> from oslotest import base as test_base <EOL> import six <EOL> from oslo_concurrency . fixture import lockutils as fixtures <EOL> from oslo_concurrency import lockutils <EOL> from oslo_config import fixture as config <EOL> class LockTestCase ( test_base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( LockTestCase , self ) . setUp ( ) <EOL> self . config = self . useFixture ( config . Config ( lockutils . CONF ) ) . config <EOL> def test_synchronized_wrapped_function_metadata ( self ) : <EOL> @ lockutils . synchronized ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> def foo ( ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> self . assertEqual ( foo . __doc__ , '<STR_LIT>' , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> self . assertEqual ( foo . __name__ , '<STR_LIT:foo>' , "<STR_LIT>" <EOL> "<STR_LIT>" ) <EOL> def test_lock_internally_different_collections ( self ) : <EOL> s1 = lockutils . Semaphores ( ) <EOL> s2 = lockutils . Semaphores ( ) <EOL> trigger = threading . Event ( ) <EOL> who_ran = collections . deque ( ) <EOL> def f ( name , semaphores , pull_trigger ) : <EOL> with lockutils . internal_lock ( '<STR_LIT>' , semaphores = semaphores ) : <EOL> if pull_trigger : <EOL> trigger . set ( ) <EOL> else : <EOL> trigger . wait ( ) <EOL> who_ran . append ( name ) <EOL> threads = [ <EOL> threading . Thread ( target = f , args = ( <NUM_LIT:1> , s1 , True ) ) , <EOL> threading . Thread ( target = f , args = ( <NUM_LIT:2> , s2 , False ) ) , <EOL> ] <EOL> for thread in threads : <EOL> thread . start ( ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> self . assertEqual ( [ <NUM_LIT:1> , <NUM_LIT:2> ] , sorted ( who_ran ) ) <EOL> def test_lock_internally ( self ) : <EOL> """<STR_LIT>""" <EOL> saved_sem_num = len ( lockutils . _semaphores ) <EOL> seen_threads = list ( ) <EOL> def f ( _id ) : <EOL> with lockutils . lock ( '<STR_LIT>' , '<STR_LIT>' , external = False ) : <EOL> for x in range ( <NUM_LIT:10> ) : <EOL> seen_threads . append ( _id ) <EOL> threads = [ ] <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> thread = threading . Thread ( target = f , args = ( i , ) ) <EOL> threads . append ( thread ) <EOL> thread . start ( ) <EOL> for thread in threads : <EOL> thread . join ( ) <EOL> self . assertEqual ( len ( seen_threads ) , <NUM_LIT:100> ) <EOL> for i in range ( <NUM_LIT:10> ) : <EOL> for j in range ( <NUM_LIT:9> ) : <EOL> self . assertEqual ( seen_threads [ i * <NUM_LIT:10> ] , <EOL> seen_threads [ i * <NUM_LIT:10> + <NUM_LIT:1> + j ] ) <EOL> self . assertEqual ( saved_sem_num , len ( lockutils . _semaphores ) , <EOL> "<STR_LIT>" ) <EOL> def test_nested_synchronized_external_works ( self ) : <EOL> """<STR_LIT>""" <EOL> tempdir = tempfile . mkdtemp ( ) <EOL> try : <EOL> self . config ( lock_path = tempdir , group = '<STR_LIT>' ) <EOL> sentinel = object ( ) <EOL> @ lockutils . synchronized ( '<STR_LIT>' , '<STR_LIT>' , external = True ) <EOL> def outer_lock ( ) : <EOL> @ lockutils . synchronized ( '<STR_LIT>' , '<STR_LIT>' , external = True ) <EOL> def inner_lock ( ) : <EOL> return sentinel <EOL> return inner_lock ( ) <EOL> self . assertEqual ( sentinel , outer_lock ( ) ) <EOL> finally : <EOL> if os . path . exists ( tempdir ) : <EOL> shutil . rmtree ( tempdir ) <EOL> def _do_test_lock_externally ( self ) : <EOL> """<STR_LIT>""" <EOL> def lock_files ( handles_dir ) : <EOL> with lockutils . lock ( '<STR_LIT>' , '<STR_LIT>' , external = True ) : <EOL> handles = [ ] <EOL> for n in range ( <NUM_LIT:50> ) : <EOL> path = os . path . join ( handles_dir , ( '<STR_LIT>' % n ) ) <EOL> handles . append ( open ( path , '<STR_LIT:w>' ) ) <EOL> count = <NUM_LIT:0> <EOL> for handle in handles : <EOL> try : <EOL> fcntl . flock ( handle , fcntl . LOCK_EX | fcntl . LOCK_NB ) <EOL> count += <NUM_LIT:1> <EOL> fcntl . flock ( handle , fcntl . LOCK_UN ) <EOL> except IOError : <EOL> os . _exit ( <NUM_LIT:2> ) <EOL> finally : <EOL> handle . close ( ) <EOL> self . assertEqual ( <NUM_LIT:50> , count ) <EOL> handles_dir = tempfile . mkdtemp ( ) <EOL> try : <EOL> children = [ ] <EOL> for n in range ( <NUM_LIT:50> ) : <EOL> pid = os . fork ( ) <EOL> if pid : <EOL> children . append ( pid ) <EOL> else : <EOL> try : <EOL> lock_files ( handles_dir ) <EOL> finally : <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> for child in children : <EOL> ( pid , status ) = os . waitpid ( child , <NUM_LIT:0> ) <EOL> if pid : <EOL> self . assertEqual ( <NUM_LIT:0> , status ) <EOL> finally : <EOL> if os . path . exists ( handles_dir ) : <EOL> shutil . rmtree ( handles_dir , ignore_errors = True ) <EOL> def test_lock_externally ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> try : <EOL> self . _do_test_lock_externally ( ) <EOL> finally : <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def test_lock_externally_lock_dir_not_exist ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> os . rmdir ( lock_dir ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> try : <EOL> self . _do_test_lock_externally ( ) <EOL> finally : <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def test_synchronized_with_prefix ( self ) : <EOL> lock_name = '<STR_LIT>' <EOL> lock_pfix = '<STR_LIT>' <EOL> foo = lockutils . synchronized_with_prefix ( lock_pfix ) <EOL> @ foo ( lock_name , external = True ) <EOL> def bar ( dirpath , pfix , name ) : <EOL> return True <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> self . assertTrue ( bar ( lock_dir , lock_pfix , lock_name ) ) <EOL> def test_synchronized_without_prefix ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> @ lockutils . synchronized ( '<STR_LIT>' , external = True ) <EOL> def test_without_prefix ( ) : <EOL> pass <EOL> try : <EOL> test_without_prefix ( ) <EOL> finally : <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def test_synchronized_prefix_without_hypen ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> @ lockutils . synchronized ( '<STR_LIT>' , '<STR_LIT>' , True ) <EOL> def test_without_hypen ( ) : <EOL> pass <EOL> try : <EOL> test_without_hypen ( ) <EOL> finally : <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def test_contextlock ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> try : <EOL> with lockutils . lock ( "<STR_LIT:test>" ) as sem : <EOL> if six . PY2 : <EOL> self . assertTrue ( isinstance ( sem , threading . _Semaphore ) ) <EOL> else : <EOL> self . assertTrue ( isinstance ( sem , threading . Semaphore ) ) <EOL> with lockutils . lock ( "<STR_LIT>" , external = True ) as lock : <EOL> self . assertTrue ( lock . exists ( ) ) <EOL> with lockutils . lock ( "<STR_LIT>" , <EOL> external = True ) as lock1 : <EOL> self . assertTrue ( isinstance ( lock1 , <EOL> lockutils . InterProcessLock ) ) <EOL> finally : <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def test_contextlock_unlocks ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> sem = None <EOL> try : <EOL> with lockutils . lock ( "<STR_LIT:test>" ) as sem : <EOL> if six . PY2 : <EOL> self . assertTrue ( isinstance ( sem , threading . _Semaphore ) ) <EOL> else : <EOL> self . assertTrue ( isinstance ( sem , threading . Semaphore ) ) <EOL> with lockutils . lock ( "<STR_LIT>" , external = True ) as lock : <EOL> self . assertTrue ( lock . exists ( ) ) <EOL> with lockutils . lock ( "<STR_LIT>" , external = True ) as lock : <EOL> self . assertTrue ( lock . exists ( ) ) <EOL> with lockutils . lock ( "<STR_LIT:test>" ) as sem2 : <EOL> self . assertEqual ( sem , sem2 ) <EOL> finally : <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def _test_remove_lock_external_file ( self , lock_dir , use_external = False ) : <EOL> lock_name = '<STR_LIT>' <EOL> lock_pfix = '<STR_LIT>' <EOL> if use_external : <EOL> lock_path = lock_dir <EOL> else : <EOL> lock_path = None <EOL> lockutils . remove_external_lock_file ( lock_name , lock_pfix , lock_path ) <EOL> for ent in os . listdir ( lock_dir ) : <EOL> self . assertRaises ( OSError , ent . startswith , lock_pfix ) <EOL> if os . path . exists ( lock_dir ) : <EOL> shutil . rmtree ( lock_dir , ignore_errors = True ) <EOL> def test_remove_lock_external_file ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . config ( lock_path = lock_dir , group = '<STR_LIT>' ) <EOL> self . _test_remove_lock_external_file ( lock_dir ) <EOL> def test_remove_lock_external_file_lock_path ( self ) : <EOL> lock_dir = tempfile . mkdtemp ( ) <EOL> self . _test_remove_lock_external_file ( lock_dir , <EOL> use_external = True ) <EOL> def test_no_slash_in_b64 ( self ) : <EOL> with lockutils . lock ( "<STR_LIT>" ) : <EOL> pass <EOL> def test_deprecated_names ( self ) : <EOL> paths = self . create_tempfiles ( [ [ '<STR_LIT>' , '<STR_LIT:\n>' . join ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' ] ) <EOL> ] ] ) <EOL> conf = cfg . ConfigOpts ( ) <EOL> conf ( [ '<STR_LIT>' , paths [ <NUM_LIT:0> ] ] ) <EOL> conf . register_opts ( lockutils . _opts , '<STR_LIT>' ) <EOL> self . assertEqual ( conf . oslo_concurrency . lock_path , '<STR_LIT:foo>' ) <EOL> self . assertTrue ( conf . oslo_concurrency . disable_process_locking ) <EOL> class FileBasedLockingTestCase ( test_base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( FileBasedLockingTestCase , self ) . setUp ( ) <EOL> self . lock_dir = tempfile . mkdtemp ( ) <EOL> def test_lock_file_exists ( self ) : <EOL> lock_file = os . path . join ( self . lock_dir , '<STR_LIT>' ) <EOL> @ lockutils . synchronized ( '<STR_LIT>' , external = True , <EOL> lock_path = self . lock_dir ) <EOL> def foo ( ) : <EOL> self . assertTrue ( os . path . exists ( lock_file ) ) <EOL> foo ( ) <EOL> def test_interprocess_lock ( self ) : <EOL> lock_file = os . path . join ( self . lock_dir , '<STR_LIT>' ) <EOL> pid = os . fork ( ) <EOL> if pid : <EOL> start = time . time ( ) <EOL> while not os . path . exists ( lock_file ) : <EOL> if time . time ( ) - start > <NUM_LIT:5> : <EOL> self . fail ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:0> ) <EOL> lock1 = lockutils . InterProcessLock ( '<STR_LIT:foo>' ) <EOL> lock1 . lockfile = open ( lock_file , '<STR_LIT:w>' ) <EOL> while time . time ( ) - start < <NUM_LIT:5> : <EOL> try : <EOL> lock1 . trylock ( ) <EOL> lock1 . unlock ( ) <EOL> time . sleep ( <NUM_LIT:0> ) <EOL> except IOError : <EOL> break <EOL> else : <EOL> self . fail ( '<STR_LIT>' ) <EOL> os . kill ( pid , signal . SIGKILL ) <EOL> else : <EOL> try : <EOL> lock2 = lockutils . InterProcessLock ( '<STR_LIT:foo>' ) <EOL> lock2 . lockfile = open ( lock_file , '<STR_LIT:w>' ) <EOL> have_lock = False <EOL> while not have_lock : <EOL> try : <EOL> lock2 . trylock ( ) <EOL> have_lock = True <EOL> except IOError : <EOL> pass <EOL> finally : <EOL> time . sleep ( <NUM_LIT> ) <EOL> os . _exit ( <NUM_LIT:0> ) <EOL> def test_interthread_external_lock ( self ) : <EOL> call_list = [ ] <EOL> @ lockutils . synchronized ( '<STR_LIT:foo>' , external = True , lock_path = self . lock_dir ) <EOL> def foo ( param ) : <EOL> """<STR_LIT>""" <EOL> call_list . append ( param ) <EOL> time . sleep ( <NUM_LIT> ) <EOL> call_list . append ( param ) <EOL> def other ( param ) : <EOL> foo ( param ) <EOL> thread = threading . Thread ( target = other , args = ( '<STR_LIT>' , ) ) <EOL> thread . start ( ) <EOL> start = time . time ( ) <EOL> while not os . path . exists ( os . path . join ( self . lock_dir , '<STR_LIT:foo>' ) ) : <EOL> if time . time ( ) - start > <NUM_LIT:5> : <EOL> self . fail ( '<STR_LIT>' ) <EOL> time . sleep ( <NUM_LIT:0> ) <EOL> thread1 = threading . Thread ( target = other , args = ( '<STR_LIT>' , ) ) <EOL> thread1 . start ( ) <EOL> thread1 . join ( ) <EOL> thread . join ( ) <EOL> self . assertEqual ( call_list , [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> def test_non_destructive ( self ) : <EOL> lock_file = os . path . join ( self . lock_dir , '<STR_LIT>' ) <EOL> with open ( lock_file , '<STR_LIT:w>' ) as f : <EOL> f . write ( '<STR_LIT:test>' ) <EOL> with lockutils . lock ( '<STR_LIT>' , external = True , <EOL> lock_path = self . lock_dir ) : <EOL> with open ( lock_file ) as f : <EOL> self . assertEqual ( f . read ( ) , '<STR_LIT:test>' ) <EOL> class LockutilsModuleTestCase ( test_base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( LockutilsModuleTestCase , self ) . setUp ( ) <EOL> self . old_env = os . environ . get ( '<STR_LIT>' ) <EOL> if self . old_env is not None : <EOL> del os . environ [ '<STR_LIT>' ] <EOL> def tearDown ( self ) : <EOL> if self . old_env is not None : <EOL> os . environ [ '<STR_LIT>' ] = self . old_env <EOL> super ( LockutilsModuleTestCase , self ) . tearDown ( ) <EOL> def test_main ( self ) : <EOL> script = '<STR_LIT:\n>' . join ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> argv = [ '<STR_LIT>' , sys . executable , '<STR_LIT:-c>' , script ] <EOL> retval = lockutils . _lock_wrapper ( argv ) <EOL> self . assertEqual ( retval , <NUM_LIT:0> , "<STR_LIT>" ) <EOL> def test_return_value_maintained ( self ) : <EOL> script = '<STR_LIT:\n>' . join ( [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] ) <EOL> argv = [ '<STR_LIT>' , sys . executable , '<STR_LIT:-c>' , script ] <EOL> retval = lockutils . _lock_wrapper ( argv ) <EOL> self . assertEqual ( retval , <NUM_LIT:1> ) <EOL> def test_direct_call_explodes ( self ) : <EOL> cmd = [ sys . executable , '<STR_LIT>' , '<STR_LIT>' ] <EOL> with open ( os . devnull , '<STR_LIT:w>' ) as devnull : <EOL> retval = subprocess . call ( cmd , stderr = devnull ) <EOL> self . assertEqual ( <NUM_LIT:1> , retval ) <EOL> class TestLockFixture ( test_base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestLockFixture , self ) . setUp ( ) <EOL> self . config = self . useFixture ( config . Config ( lockutils . CONF ) ) . config <EOL> self . tempdir = tempfile . mkdtemp ( ) <EOL> def _check_in_lock ( self ) : <EOL> self . assertTrue ( self . lock . exists ( ) ) <EOL> def tearDown ( self ) : <EOL> self . _check_in_lock ( ) <EOL> super ( TestLockFixture , self ) . tearDown ( ) <EOL> def test_lock_fixture ( self ) : <EOL> self . config ( lock_path = self . tempdir , group = '<STR_LIT>' ) <EOL> fixture = fixtures . LockFixture ( '<STR_LIT>' ) <EOL> self . useFixture ( fixture ) <EOL> self . lock = fixture . lock <EOL> class TestGetLockPath ( test_base . BaseTestCase ) : <EOL> def setUp ( self ) : <EOL> super ( TestGetLockPath , self ) . setUp ( ) <EOL> self . conf = self . useFixture ( config . Config ( lockutils . CONF ) ) . conf <EOL> def test_get_default ( self ) : <EOL> lockutils . set_defaults ( lock_path = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , lockutils . get_lock_path ( self . conf ) ) <EOL> def test_get_override ( self ) : <EOL> lockutils . _register_opts ( self . conf ) <EOL> self . conf . set_override ( '<STR_LIT>' , '<STR_LIT>' , <EOL> group = '<STR_LIT>' ) <EOL> self . assertEqual ( '<STR_LIT>' , lockutils . get_lock_path ( self . conf ) ) </s>
