Author: Brian Kearns <[email protected]>
Branch:
Changeset: r59441:aedad230f8b8
Date: 2012-12-13 22:49 -0500
http://bitbucket.org/pypy/pypy/changeset/aedad230f8b8/
Log: improve fcntl.test_flock/test_lockf so they actually test locking
diff --git a/pypy/module/fcntl/test/test_fcntl.py
b/pypy/module/fcntl/test/test_fcntl.py
--- a/pypy/module/fcntl/test/test_fcntl.py
+++ b/pypy/module/fcntl/test/test_fcntl.py
@@ -100,22 +100,43 @@
def test_flock(self):
import fcntl
- import sys
+ import os
+ import errno
f = open(self.tmp + "c", "w+")
raises(TypeError, fcntl.flock, "foo")
raises(TypeError, fcntl.flock, f, "foo")
- fcntl.flock(f, fcntl.LOCK_SH)
- # this is an error EWOULDBLOCK, man: The file is locked and the
- # LOCK_NB flag was selected.
- raises(IOError, fcntl.flock, f, fcntl.LOCK_NB)
+
+ fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+ pid = os.fork()
+ if pid == 0:
+ rval = 2
+ try:
+ fcntl.flock(open(f.name, f.mode), fcntl.LOCK_EX |
fcntl.LOCK_NB)
+ except IOError, e:
+ if e.errno not in (errno.EACCES, errno.EAGAIN):
+ raise
+ rval = 0
+ else:
+ rval = 1
+ finally:
+ os._exit(rval)
+
+ assert pid > 0
+ (pid, status) = os.waitpid(pid, 0)
+ assert os.WIFEXITED(status) == True
+ assert os.WEXITSTATUS(status) == 0
+
fcntl.flock(f, fcntl.LOCK_UN)
f.close()
def test_lockf(self):
import fcntl
+ import os
+ import errno
f = open(self.tmp + "d", "w+")
@@ -124,7 +145,27 @@
raises(ValueError, fcntl.lockf, f, -256)
raises(ValueError, fcntl.lockf, f, 256)
- fcntl.lockf(f, fcntl.LOCK_SH)
+ fcntl.lockf(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+ pid = os.fork()
+ if pid == 0:
+ rval = 2
+ try:
+ fcntl.lockf(open(f.name, f.mode), fcntl.LOCK_EX |
fcntl.LOCK_NB)
+ except IOError, e:
+ if e.errno not in (errno.EACCES, errno.EAGAIN):
+ raise
+ rval = 0
+ else:
+ rval = 1
+ finally:
+ os._exit(rval)
+
+ assert pid > 0
+ (pid, status) = os.waitpid(pid, 0)
+ assert os.WIFEXITED(status) == True
+ assert os.WEXITSTATUS(status) == 0
+
fcntl.lockf(f, fcntl.LOCK_UN)
f.close()
@@ -188,11 +229,6 @@
os.close(mfd)
os.close(sfd)
- def test_lockf_with_ex(self):
- import fcntl
- f = open(self.tmp, "w")
- fcntl.lockf(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
-
def test_large_flag(self):
import sys
if any(plat in sys.platform
_______________________________________________
pypy-commit mailing list
[email protected]
http://mail.python.org/mailman/listinfo/pypy-commit
