]> git.ipfire.org Git - thirdparty/Python/cpython.git/commitdiff
bpo-22367: Update test_fcntl.py for spawn process mode (GH-17154) (GH-17253)
authorMiss Islington (bot) <31488909+miss-islington@users.noreply.github.com>
Fri, 22 Nov 2019 14:15:42 +0000 (06:15 -0800)
committerVictor Stinner <vstinner@python.org>
Fri, 22 Nov 2019 14:15:42 +0000 (15:15 +0100)
(cherry picked from commit 9960230f76eb555d6dfbe8a324efed35610c85f9)

Co-authored-by: Dong-hee Na <donghee.na92@gmail.com>
Lib/test/test_fcntl.py

index 30eeb6dc4d02a780cea2c92d98e0ebc07b89f7da..036bd64b40511850e6e0e1e3957dc73e506673b3 100644 (file)
@@ -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)