From befa032d8869e0fab4732d910f3887642879d644 Mon Sep 17 00:00:00 2001 From: Dong-hee Na Date: Fri, 8 Nov 2019 05:31:41 +0900 Subject: bpo-22367: Add tests for fcntl.lockf(). (GH-17010) --- Lib/test/test_fcntl.py | 29 ++++++++++++++++++++++++++++- 1 file changed, 28 insertions(+), 1 deletion(-) 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 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 -- cgit v0.12