From: Miss Islington (bot) <31488909+miss-islington@users.noreply.github.com>
Date: Sat, 9 Nov 2019 11:12:35 +0000 (-0800)
Subject: bpo-22367: Add tests for fcntl.lockf(). (GH-17010)
X-Git-Tag: v3.7.6rc1~63
X-Git-Url: http://git.ipfire.org/gitweb.cgi?a=commitdiff_plain;h=917dbe350a762ed6d75b7d074f3fb87975ba717b;p=thirdparty%2FPython%2Fcpython.git

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 acd5c7cc5864..30eeb6dc4d02 100644
--- 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