bpo-22367: Add tests for fcntl.lockf(). (GH-17010)
diff --git a/Lib/test/test_fcntl.py b/Lib/test/test_fcntl.py
index a2b5997..9d1be28 100644
--- a/Lib/test/test_fcntl.py
+++ b/Lib/test/test_fcntl.py
@@ -5,6 +5,7 @@
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 @@
fcntl = import_module('fcntl')
-# TODO - Write tests for flock() and lockf().
def get_lockdata():
try:
@@ -138,6 +138,33 @@
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