summaryrefslogtreecommitdiff
path: root/test
diff options
context:
space:
mode:
authorRémi Cardona <remi.cardona@logilab.fr>2015-09-10 16:08:32 +0200
committerRémi Cardona <remi.cardona@logilab.fr>2015-09-10 16:08:32 +0200
commit9c3aeaeac0e8adc5d7d2b8360de72d9b4721acb4 (patch)
treebf07531f6e88ee8133910bb3c0499843694eb567 /test
parentd41ea6b6eb567799c6a534cf004da520d6ae4f6e (diff)
downloadlogilab-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.py44
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):