From 17893908ee823aada5aa4769cfcddcf88e042246 Mon Sep 17 00:00:00 2001 From: Dong-hee Na Date: Fri, 8 Nov 2019 05:31:41 +0900 Subject: [PATCH] bpo-22367: Add tests for fcntl.lockf(). (GH-17010) (cherry picked from commit befa032d8869e0fab4732d910f3887642879d644) Co-authored-by: Dong-hee Na --- 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 5d4abe388f7828..88d989d5e358d1 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 @@ 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