diff options
author | Jeff Forcier <jeff@bitprophet.org> | 2018-05-29 15:26:48 -0700 |
---|---|---|
committer | Jeff Forcier <jeff@bitprophet.org> | 2018-05-29 15:26:48 -0700 |
commit | 3c90e2cd52f7b956671db944061dc5b6cc7872d1 (patch) | |
tree | 1f03c2065463d2c7c71181c3a766593b4ed04f9d /tests/test_buffered_pipe.py | |
parent | 2b40bf14493245ef4112128a66b3f347c6f7b746 (diff) | |
parent | f2ab9241daa49be4a7293b903eac34188d3a74d8 (diff) |
Merge branch '2.4'
Diffstat (limited to 'tests/test_buffered_pipe.py')
-rw-r--r-- | tests/test_buffered_pipe.py | 27 |
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() |