diff options
author | Dong-hee Na <donghee.na92@gmail.com> | 2019-11-19 08:12:42 (GMT) |
---|---|---|
committer | Ćukasz Langa <lukasz@langa.pl> | 2019-11-19 08:12:42 (GMT) |
commit | 9960230f76eb555d6dfbe8a324efed35610c85f9 (patch) | |
tree | 478d8c216c7bcea8723002e27dec7d1620a09ddd /Lib/test | |
parent | 79e18ec75d28fbb2f5a9d1d030a10d66ced38742 (diff) | |
download | cpython-9960230f76eb555d6dfbe8a324efed35610c85f9.zip cpython-9960230f76eb555d6dfbe8a324efed35610c85f9.tar.gz cpython-9960230f76eb555d6dfbe8a324efed35610c85f9.tar.bz2 |
bpo-22367: Update test_fcntl.py for spawn process mode (#17154)
Diffstat (limited to 'Lib/test')
-rw-r--r-- | Lib/test/test_fcntl.py | 28 |
1 files changed, 19 insertions, 9 deletions
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py index 9d1be28..9ab68c6 100644 --- a/Lib/test/test_fcntl.py +++ b/Lib/test/test_fcntl.py @@ -51,6 +51,21 @@ class BadFile: def fileno(self): return self.fn +def try_lockf_on_other_process_fail(fname, cmd): + f = open(fname, 'wb+') + try: + fcntl.lockf(f, cmd) + except BlockingIOError: + pass + finally: + f.close() + +def try_lockf_on_other_process(fname, cmd): + f = open(fname, 'wb+') + fcntl.lockf(f, cmd) + fcntl.lockf(f, fcntl.LOCK_UN) + f.close() + class TestFcntl(unittest.TestCase): def setUp(self): @@ -138,28 +153,23 @@ class TestFcntl(unittest.TestCase): self.assertRaises(ValueError, fcntl.flock, -1, fcntl.LOCK_SH) self.assertRaises(TypeError, fcntl.flock, 'spam', fcntl.LOCK_SH) + @unittest.skipIf(platform.system() == "AIX", "AIX returns PermissionError") def test_lockf_exclusive(self): self.f = open(TESTFN, 'wb+') 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 = Process(target=try_lockf_on_other_process_fail, args=(TESTFN, cmd)) p.start() p.join() fcntl.lockf(self.f, fcntl.LOCK_UN) self.assertEqual(p.exitcode, 0) + @unittest.skipIf(platform.system() == "AIX", "AIX returns PermissionError") 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 = Process(target=try_lockf_on_other_process, args=(TESTFN, cmd)) p.start() p.join() fcntl.lockf(self.f, fcntl.LOCK_UN) |