diff options
Diffstat (limited to 'Lib/test')
| -rw-r--r-- | Lib/test/test_fcntl.py | 29 |
1 files changed, 28 insertions, 1 deletions
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 |