diff options
author | Rémi Cardona <remi.cardona@logilab.fr> | 2015-09-10 16:08:32 +0200 |
---|---|---|
committer | Rémi Cardona <remi.cardona@logilab.fr> | 2015-09-10 16:08:32 +0200 |
commit | 9c3aeaeac0e8adc5d7d2b8360de72d9b4721acb4 (patch) | |
tree | bf07531f6e88ee8133910bb3c0499843694eb567 /test | |
parent | d41ea6b6eb567799c6a534cf004da520d6ae4f6e (diff) | |
download | logilab-common-9c3aeaeac0e8adc5d7d2b8360de72d9b4721acb4.tar.gz |
[shellutils] Remove racy file lock
acquire_lock is racy on Unix (TOCTOU) and broken on Windows (lock file
is removed instantly since ProcInfo always raises NoSuchProcess on this
OS).
Diffstat (limited to 'test')
-rw-r--r-- | test/unittest_shellutils.py | 44 |
1 files changed, 3 insertions, 41 deletions
diff --git a/test/unittest_shellutils.py b/test/unittest_shellutils.py index 16ee63e..9342ae9 100644 --- a/test/unittest_shellutils.py +++ b/test/unittest_shellutils.py @@ -26,13 +26,13 @@ from six.moves import range from logilab.common.testlib import TestCase, unittest_main from logilab.common.shellutils import (globfind, find, ProgressBar, - acquire_lock, release_lock, RawInput) -from logilab.common.compat import str_to_bytes, StringIO -from logilab.common.proc import NoSuchProcess +from logilab.common.compat import StringIO + DATA_DIR = join(dirname(abspath(__file__)), 'data', 'find_test') + class FindTC(TestCase): def test_include(self): files = set(find(DATA_DIR, '.py')) @@ -170,44 +170,6 @@ class ProgressBarTC(TestCase): self.assertEqual(pgb_stream.getvalue(), expected_stream.getvalue()) -class AcquireLockTC(TestCase): - - def setUp(self): - self.tmpdir = tempfile.mkdtemp() - self.lock = join(self.tmpdir, 'LOCK') - - def tearDown(self): - shutil.rmtree(self.tmpdir) - - def test_acquire_normal(self): - self.assertTrue(acquire_lock(self.lock, 1, 1)) - self.assertTrue(os.path.exists(self.lock)) - release_lock(self.lock) - self.assertFalse(os.path.exists(self.lock)) - - def test_no_possible_acquire(self): - self.assertRaises(Exception, acquire_lock, self.lock, 0) - - def test_wrong_process(self): - fd = os.open(self.lock, os.O_EXCL | os.O_RDWR | os.O_CREAT) - os.write(fd, str_to_bytes('1111111111')) - os.close(fd) - self.assertTrue(os.path.exists(self.lock)) - self.assertRaises(Exception, acquire_lock, self.lock, 1, 1) - - def test_wrong_process_and_continue(self): - fd = os.open(self.lock, os.O_EXCL | os.O_RDWR | os.O_CREAT) - os.write(fd, str_to_bytes('1111111111')) - os.close(fd) - self.assertTrue(os.path.exists(self.lock)) - self.assertTrue(acquire_lock(self.lock)) - - def test_locked_for_one_hour(self): - self.assertTrue(acquire_lock(self.lock)) - touch = datetime.datetime.fromtimestamp(time.time() - 3601).strftime("%m%d%H%M") - os.system("touch -t %s %s" % (touch, self.lock)) - self.assertRaises(UserWarning, acquire_lock, self.lock, max_try=2, delay=1) - class RawInputTC(TestCase): def auto_input(self, *args): |