From a9eb79d290073267cafe5372c9d8830b0ac6a577 Mon Sep 17 00:00:00 2001 From: Dong-hee Na Date: Thu, 31 Oct 2019 15:47:58 +0900 Subject: [PATCH 1/2] bpo-22367: Add test_lockf --- Lib/test/test_fcntl.py | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py index a2b5997067539e..b30ba9594f5949 100644 --- a/Lib/test/test_fcntl.py +++ b/Lib/test/test_fcntl.py @@ -12,7 +12,6 @@ fcntl = import_module('fcntl') -# TODO - Write tests for flock() and lockf(). def get_lockdata(): try: @@ -138,6 +137,34 @@ def test_flock(self): self.assertRaises(ValueError, fcntl.flock, -1, fcntl.LOCK_SH) self.assertRaises(TypeError, fcntl.flock, 'spam', fcntl.LOCK_SH) + def test_lockf(self): + from multiprocessing import Process + self.f = open(TESTFN, 'wb+') + self.f.write(b'testpython') + ex_cmd = fcntl.LOCK_EX | fcntl.LOCK_NB + sh_cmd = fcntl.LOCK_SH | fcntl.LOCK_NB + def try_lock_f_on_other_process(cmd): + if cmd == ex_cmd: + self.assertRaises(BlockingIOError, fcntl.lockf, self.f, cmd, 1, 0) + if cmd == sh_cmd: + fcntl.lockf(self.f, cmd, 1, 0) + fcntl.lockf(self.f, cmd, 1, 1) + + fcntl.lockf(self.f, ex_cmd, 1, 0) + p = Process(target=try_lock_f_on_other_process, args=(ex_cmd,)) + p.start() + p.join() + fcntl.lockf(self.f, fcntl.LOCK_UN) + self.assertEqual(p.exitcode, 0) + + fcntl.lockf(self.f, sh_cmd, 1, 0) + p = Process(target=try_lock_f_on_other_process, args=(sh_cmd,)) + p.start() + p.join() + fcntl.lockf(self.f, fcntl.LOCK_UN) + self.assertEqual(p.exitcode, 0) + + @cpython_only def test_flock_overflow(self): import _testcapi From 4138cba7a6cb61d26ef308262ab64971880b8a0d Mon Sep 17 00:00:00 2001 From: Dong-hee Na Date: Thu, 31 Oct 2019 21:54:29 +0900 Subject: [PATCH 2/2] bpo-22367: Apply codereview --- Lib/test/test_fcntl.py | 34 +++++++++++++++++----------------- 1 file changed, 17 insertions(+), 17 deletions(-) diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py index b30ba9594f5949..9d1be28c6d3992 100644 --- a/Lib/test/test_fcntl.py +++ b/Lib/test/test_fcntl.py @@ -5,6 +5,7 @@ import struct import sys import unittest +from multiprocessing import Process from test.support import (verbose, TESTFN, unlink, run_unittest, import_module, cpython_only) @@ -137,34 +138,33 @@ def test_flock(self): self.assertRaises(ValueError, fcntl.flock, -1, fcntl.LOCK_SH) self.assertRaises(TypeError, fcntl.flock, 'spam', fcntl.LOCK_SH) - def test_lockf(self): - from multiprocessing import Process + def test_lockf_exclusive(self): self.f = open(TESTFN, 'wb+') - self.f.write(b'testpython') - ex_cmd = fcntl.LOCK_EX | fcntl.LOCK_NB - sh_cmd = fcntl.LOCK_SH | fcntl.LOCK_NB - def try_lock_f_on_other_process(cmd): - if cmd == ex_cmd: - self.assertRaises(BlockingIOError, fcntl.lockf, self.f, cmd, 1, 0) - if cmd == sh_cmd: - fcntl.lockf(self.f, cmd, 1, 0) - fcntl.lockf(self.f, cmd, 1, 1) - - fcntl.lockf(self.f, ex_cmd, 1, 0) - p = Process(target=try_lock_f_on_other_process, args=(ex_cmd,)) + cmd = fcntl.LOCK_EX | fcntl.LOCK_NB + def try_lockf_on_other_process(): + self.assertRaises(BlockingIOError, fcntl.lockf, self.f, cmd) + + fcntl.lockf(self.f, cmd) + p = Process(target=try_lockf_on_other_process) p.start() p.join() fcntl.lockf(self.f, fcntl.LOCK_UN) self.assertEqual(p.exitcode, 0) - fcntl.lockf(self.f, sh_cmd, 1, 0) - p = Process(target=try_lock_f_on_other_process, args=(sh_cmd,)) + def test_lockf_share(self): + self.f = open(TESTFN, 'wb+') + cmd = fcntl.LOCK_SH | fcntl.LOCK_NB + def try_lockf_on_other_process(): + fcntl.lockf(self.f, cmd) + fcntl.lockf(self.f, fcntl.LOCK_UN) + + fcntl.lockf(self.f, cmd) + p = Process(target=try_lockf_on_other_process) p.start() p.join() fcntl.lockf(self.f, fcntl.LOCK_UN) self.assertEqual(p.exitcode, 0) - @cpython_only def test_flock_overflow(self): import _testcapi