summaryrefslogtreecommitdiffhomepage
path: root/tests/test_buffered_pipe.py
diff options
context:
space:
mode:
authorJeff Forcier <jeff@bitprophet.org>2018-05-29 15:26:48 -0700
committerJeff Forcier <jeff@bitprophet.org>2018-05-29 15:26:48 -0700
commit3c90e2cd52f7b956671db944061dc5b6cc7872d1 (patch)
tree1f03c2065463d2c7c71181c3a766593b4ed04f9d /tests/test_buffered_pipe.py
parent2b40bf14493245ef4112128a66b3f347c6f7b746 (diff)
parentf2ab9241daa49be4a7293b903eac34188d3a74d8 (diff)
Merge branch '2.4'
Diffstat (limited to 'tests/test_buffered_pipe.py')
-rw-r--r--tests/test_buffered_pipe.py27
1 files changed, 14 insertions, 13 deletions
diff --git a/tests/test_buffered_pipe.py b/tests/test_buffered_pipe.py
index 03616c55..9f986a5e 100644
--- a/tests/test_buffered_pipe.py
+++ b/tests/test_buffered_pipe.py
@@ -30,9 +30,9 @@ from paramiko.py3compat import b
def delay_thread(p):
- p.feed('a')
+ p.feed("a")
time.sleep(0.5)
- p.feed('b')
+ p.feed("b")
p.close()
@@ -42,41 +42,42 @@ def close_thread(p):
class BufferedPipeTest(unittest.TestCase):
+
def test_1_buffered_pipe(self):
p = BufferedPipe()
self.assertTrue(not p.read_ready())
- p.feed('hello.')
+ p.feed("hello.")
self.assertTrue(p.read_ready())
data = p.read(6)
- self.assertEqual(b'hello.', data)
+ 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.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))
+ self.assertEqual(b"", p.read(1))
def test_2_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))
+ 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))
+ self.assertEqual(b"b", p.read(1, 1.0))
+ self.assertEqual(b"", p.read(1))
def test_3_close_while_reading(self):
p = BufferedPipe()
threading.Thread(target=close_thread, args=(p,)).start()
data = p.read(1, 1.0)
- self.assertEqual(b'', data)
+ self.assertEqual(b"", data)
def test_4_or_pipe(self):
p = pipe.make_pipe()