Hi,
Le Saturday 19 July 2008 15:14:44, vous avez écrit :
> Thank you Victor - I didn't want to change any underlying
> multiprocessing code until we had the test suite in a better state
> (which we do now) (...)
>
> One suggestion would be to include tests to prove the bugs is fixed if
> possible (to the patch), so we can add them to the suite.
Ok, here is a patch for test_multiprocessing.py.
- TestClosedFile.test_open() verify that Connection() rejects closed file
descriptor
- TestClosedFile.test_operations() verify that Connection() raises IOError for
operations on closed file
I don't know if Connection() is a socket or a pipe. Both should be tested.
Victor
Index: Lib/test/test_multiprocessing.py
===================================================================
--- Lib/test/test_multiprocessing.py (révision 65132)
+++ Lib/test/test_multiprocessing.py (copie de travail)
@@ -54,6 +54,9 @@
HAVE_GETVALUE = not getattr(_multiprocessing,
'HAVE_BROKEN_SEM_GETVALUE', False)
+# Any existing and readable file name
+EXISTING_FILE = "/etc/issue"
+
#
# Creates a wrapper for a function which records the time it takes to finish
#
@@ -1737,6 +1740,29 @@
testcases_threads = create_test_cases(ThreadsMixin, type='threads')
globals().update(testcases_threads)
+class TestClosedFile(unittest.TestCase):
+ def test_open(self):
+ # Create a file descriptor and directly close it
+ tmpfile = open(EXISTING_FILE)
+ fd = tmpfile.fileno()
+ tmpfile.close()
+ self.assertRaises(IOError, _multiprocessing.Connection, fd)
+
+ def test_operations(self):
+ # Create a dummy file descriptor
+ fd, unused_fd = os.pipe()
+ os.close(unused_fd)
+ try:
+ conn = _multiprocessing.Connection(fd)
+ conn.close()
+ fd = None
+ self.assertRaises(IOError, conn.send, "test")
+ self.assertRaises(IOError, conn.recv)
+ self.assertRaises(IOError, conn.poll)
+ finally:
+ if fd is not None:
+ os.close(fd)
+
#
#
#
@@ -1765,7 +1791,8 @@
testcases = (
sorted(testcases_processes.values(), key=lambda tc:tc.__name__) +
sorted(testcases_threads.values(), key=lambda tc:tc.__name__) +
- sorted(testcases_manager.values(), key=lambda tc:tc.__name__)
+ sorted(testcases_manager.values(), key=lambda tc:tc.__name__) +
+ [TestClosedFile]
)
loadTestsFromTestCase = unittest.defaultTestLoader.loadTestsFromTestCase
_______________________________________________
Python-Dev mailing list
[email protected]
http://mail.python.org/mailman/listinfo/python-dev
Unsubscribe:
http://mail.python.org/mailman/options/python-dev/archive%40mail-archive.com
