bpo-22367: Add tests for fcntl.lockf(). (GH-17010)

(cherry picked from commit befa032d8869e0fab4732d910f3887642879d644)

Co-authored-by: Dong-hee Na <donghee.na92@gmail.com>

diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py
index acd5c7cc586472d255d07255eb1a5d8b75791db8..30eeb6dc4d02a780cea2c92d98e0ebc07b89f7da 100644 (file)
--- a/Lib/test/test_fcntl.py
+++ b/Lib/test/test_fcntl.py
@@ -5,6 +5,7 @@ import os
 import struct
 import sys
 import unittest
+from multiprocessing import Process
 from test.support import (verbose, TESTFN, unlink, run_unittest, import_module,
                           cpython_only)
 
@@ -12,7 +13,6 @@ from test.support import (verbose, TESTFN, unlink, run_unittest, import_module,
 fcntl = import_module('fcntl')
 
 
-# TODO - Write tests for flock() and lockf().
 
 def get_lockdata():
     try:
@@ -138,6 +138,33 @@ class TestFcntl(unittest.TestCase):
         self.assertRaises(ValueError, fcntl.flock, -1, fcntl.LOCK_SH)
         self.assertRaises(TypeError, fcntl.flock, 'spam', fcntl.LOCK_SH)
 
+    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.start()
+        p.join()
+        fcntl.lockf(self.f, fcntl.LOCK_UN)
+        self.assertEqual(p.exitcode, 0)
+
+    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