diff --git a/oslo/concurrency/lockutils.py b/oslo/concurrency/lockutils.py index 1b6cc1a..a7ef40f 100644 --- a/oslo/concurrency/lockutils.py +++ b/oslo/concurrency/lockutils.py @@ -197,12 +197,12 @@ def external_lock(name, lock_file_prefix=None, lock_path=None): return InterProcessLock(lock_file_path) -def remove_external_lock_file(name, lock_file_prefix=None): +def remove_external_lock_file(name, lock_file_prefix=None, lock_path=None): """Remove an external lock file when it's not used anymore This will be helpful when we have a lot of lock files """ with internal_lock(name): - lock_file_path = _get_lock_path(name, lock_file_prefix) + lock_file_path = _get_lock_path(name, lock_file_prefix, lock_path) try: os.remove(lock_file_path) except OSError: diff --git a/tests/unit/test_lockutils.py b/tests/unit/test_lockutils.py index 8724440..e764921 100644 --- a/tests/unit/test_lockutils.py +++ b/tests/unit/test_lockutils.py @@ -315,14 +315,16 @@ class LockTestCase(test_base.BaseTestCase): if os.path.exists(lock_dir): shutil.rmtree(lock_dir, ignore_errors=True) - def test_remove_lock_external_file(self): + def _test_remove_lock_external_file(self, lock_dir, use_external=False): lock_name = 'mylock' lock_pfix = 'mypfix-remove-lock-test-' - lock_dir = tempfile.mkdtemp() - self.config(lock_path=lock_dir) + if use_external: + lock_path = lock_dir + else: + lock_path = None - lockutils.remove_external_lock_file(lock_name, lock_pfix) + lockutils.remove_external_lock_file(lock_name, lock_pfix, lock_path) for ent in os.listdir(lock_dir): self.assertRaises(OSError, ent.startswith, lock_pfix) @@ -330,6 +332,16 @@ class LockTestCase(test_base.BaseTestCase): if os.path.exists(lock_dir): shutil.rmtree(lock_dir, ignore_errors=True) + def test_remove_lock_external_file(self): + lock_dir = tempfile.mkdtemp() + self.config(lock_path=lock_dir) + self._test_remove_lock_external_file(lock_dir) + + def test_remove_lock_external_file_lock_path(self): + lock_dir = tempfile.mkdtemp() + self._test_remove_lock_external_file(lock_dir, + use_external=True) + def test_no_slash_in_b64(self): # base64(sha1(foobar)) has a slash in it with lockutils.lock("foobar"):