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

Author: corona10

Lib/test/test_fcntl.py

@@ -5,14 +5,14 @@
 import struct
 import sys
 import unittest
+from multiprocessing import Process
 from test.support import (verbose, TESTFN, unlink, run_unittest, import_module,
                           cpython_only)
 
 # Skip test if no fcntl module.
 fcntl = import_module('fcntl')
 
 
-# TODO - Write tests for flock() and lockf().
 
 def get_lockdata():
     try:
@@ -138,6 +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_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