Adding upstream version 3.4.0.upstream/3.4.0upstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 91 insertions, 0 deletions

diff --git a/tests/test_buffered_pipe.py b/tests/test_buffered_pipe.py
new file mode 100644
index 0000000..35e2cde
--- /dev/null
+++ b/tests/test_buffered_pipe.py
@@ -0,0 +1,91 @@
+# Copyright (C) 2006-2007  Robey Pointer <robeypointer@gmail.com>
+#
+# This file is part of paramiko.
+#
+# Paramiko is free software; you can redistribute it and/or modify it under the
+# terms of the GNU Lesser General Public License as published by the Free
+# Software Foundation; either version 2.1 of the License, or (at your option)
+# any later version.
+#
+# Paramiko is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Lesser General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with Paramiko; if not, write to the Free Software Foundation, Inc.,
+# 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301 USA.
+
+"""
+Some unit tests for BufferedPipe.
+"""
+
+import threading
+import time
+import unittest
+
+from paramiko.buffered_pipe import BufferedPipe, PipeTimeout
+from paramiko import pipe
+
+
+def delay_thread(p):
+    p.feed("a")
+    time.sleep(0.5)
+    p.feed("b")
+    p.close()
+
+
+def close_thread(p):
+    time.sleep(0.2)
+    p.close()
+
+
+class BufferedPipeTest(unittest.TestCase):
+    def test_buffered_pipe(self):
+        p = BufferedPipe()
+        self.assertTrue(not p.read_ready())
+        p.feed("hello.")
+        self.assertTrue(p.read_ready())
+        data = p.read(6)
+        self.assertEqual(b"hello.", data)
+
+        p.feed("plus/minus")
+        self.assertEqual(b"plu", p.read(3))
+        self.assertEqual(b"s/m", p.read(3))
+        self.assertEqual(b"inus", p.read(4))
+
+        p.close()
+        self.assertTrue(not p.read_ready())
+        self.assertEqual(b"", p.read(1))
+
+    def test_delay(self):
+        p = BufferedPipe()
+        self.assertTrue(not p.read_ready())
+        threading.Thread(target=delay_thread, args=(p,)).start()
+        self.assertEqual(b"a", p.read(1, 0.1))
+        try:
+            p.read(1, 0.1)
+            self.assertTrue(False)
+        except PipeTimeout:
+            pass
+        self.assertEqual(b"b", p.read(1, 1.0))
+        self.assertEqual(b"", p.read(1))
+
+    def test_close_while_reading(self):
+        p = BufferedPipe()
+        threading.Thread(target=close_thread, args=(p,)).start()
+        data = p.read(1, 1.0)
+        self.assertEqual(b"", data)
+
+    def test_or_pipe(self):
+        p = pipe.make_pipe()
+        p1, p2 = pipe.make_or_pipe(p)
+        self.assertFalse(p._set)
+        p1.set()
+        self.assertTrue(p._set)
+        p2.set()
+        self.assertTrue(p._set)
+        p1.clear()
+        self.assertTrue(p._set)
+        p2.clear()
+        self.assertFalse(p._set)