提交 66d973bd authored 作者: Brandon T. Willard's avatar Brandon T. Willard 提交者: Brandon T. Willard

Make sure the first thread obtains the lock first in test_locking_thread

上级 7579a8cd
...@@ -81,13 +81,22 @@ def run_locking_test(ctx): ...@@ -81,13 +81,22 @@ def run_locking_test(ctx):
def test_locking_thread(): def test_locking_thread():
import traceback
with tempfile.TemporaryDirectory() as dir_name: with tempfile.TemporaryDirectory() as dir_name:
def test_fn_1(): def test_fn_1(arg):
try:
with lock_ctx(dir_name): with lock_ctx(dir_name):
# Sleep "indefinitely" # Notify the outside that we've obtained the lock
time.sleep(100) arg.append(False)
while True not in arg:
time.sleep(0.5)
except Exception:
# Notify the outside that we done
arg.append(False)
# If something unexpected happened, we want to know what it was
traceback.print_exc()
def test_fn_2(arg): def test_fn_2(arg):
try: try:
...@@ -98,18 +107,30 @@ def test_locking_thread(): ...@@ -98,18 +107,30 @@ def test_locking_thread():
# It timed out, which means that the lock was still held by the # It timed out, which means that the lock was still held by the
# first thread # first thread
arg.append(True) arg.append(True)
except Exception:
# If something unexpected happened, we want to know what it was
traceback.print_exc()
thread_1 = threading.Thread(target=test_fn_1)
res = [] res = []
thread_1 = threading.Thread(target=test_fn_1, args=(res,))
thread_2 = threading.Thread(target=test_fn_2, args=(res,)) thread_2 = threading.Thread(target=test_fn_2, args=(res,))
thread_1.start() thread_1.start()
# Make sure the first thread has obtained the lock
while False not in res:
time.sleep(0.5)
thread_2.start() thread_2.start()
# The second thread should raise `filelock.Timeout` # The second thread should raise `filelock.Timeout`
thread_2.join() thread_2.join()
assert True in res assert True in res
thread_1.join()
assert not thread_1.is_alive()
assert not thread_2.is_alive()
@pytest.mark.skipif(sys.platform != "linux", reason="Fork is only available on linux") @pytest.mark.skipif(sys.platform != "linux", reason="Fork is only available on linux")
def test_locking_multiprocess_fork(): def test_locking_multiprocess_fork():
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论