Celebrating 11 years old bug! Congratulations!

The bug so ancient it has even its own answer on the StackOverflow
(http://stackoverflow.com/questions/9280550/)!

1 files changed, 40 insertions, 1 deletions

diff --git a/tests/test_bio_membuf.py b/tests/test_bio_membuf.py
index 1e74873..4abdc73 100644
--- a/tests/test_bio_membuf.py
+++ b/tests/test_bio_membuf.py
@@ -4,14 +4,35 @@
 
 Copyright (c) 2000 Ng Pheng Siong. All rights reserved."""
 
+import multiprocessing
 try:
     import unittest2 as unittest
 except ImportError:
     import unittest
 
-import M2Crypto
 from M2Crypto.BIO import MemoryBuffer
 
+class TimeLimitExpired(Exception):
+    pass
+
+def time_limit(timeout, func, exc_msg, *args, **kwargs):
+    class FuncProc(multiprocessing.Process):
+        def __init__(self):
+            multiprocessing.Process.__init__(self)
+            self.result = None
+
+        def run(self):
+            self.result = func(args, kwargs)
+
+    it = FuncProc()
+    it.start()
+    it.join(timeout)
+    if it.is_alive():
+        it.terminate()
+        raise TimeLimitExpired(exc_msg)
+    else:
+        return it.result
+
 class MemoryBufferTestCase(unittest.TestCase):
 
     def setUp(self):
@@ -74,6 +95,24 @@ class MemoryBufferTestCase(unittest.TestCase):
             mb.write(self.data)
         assert mb.readable() and not mb.writeable()
 
+    def test_readline(self):
+        # test against possible endless loop
+        # http://stackoverflow.com/questions/9280550/
+        timeout_secs = 10
+
+        def run_test():
+            with MemoryBuffer('hello\nworld\n') as mb:
+                self.assertTrue(mb.readable())
+                self.assertEqual(mb.readline().rstrip(), 'hello')
+                self.assertEqual(mb.readline().rstrip(), 'world')
+
+            with MemoryBuffer('hello\nworld\n') as mb:
+                self.assertEqual(mb.readlines(),
+                                 ['hello\n', 'world\n'])
+
+        time_limit(timeout_secs, run_test,
+                   'The readline() should not timeout!')
+
 
 def suite():
     return unittest.makeSuite(MemoryBufferTestCase)