diff options
Diffstat (limited to 'tests/test_buffered_pipe.py')
-rw-r--r-- | tests/test_buffered_pipe.py | 31 |
1 files changed, 16 insertions, 15 deletions
diff --git a/tests/test_buffered_pipe.py b/tests/test_buffered_pipe.py index 47ece936..b9d2bef4 100644 --- a/tests/test_buffered_pipe.py +++ b/tests/test_buffered_pipe.py @@ -25,8 +25,9 @@ import time import unittest from paramiko.buffered_pipe import BufferedPipe, PipeTimeout from paramiko import pipe +from paramiko.py3compat import b -from util import ParamikoTest +from tests.util import ParamikoTest def delay_thread(pipe): @@ -44,39 +45,39 @@ def close_thread(pipe): class BufferedPipeTest(ParamikoTest): def test_1_buffered_pipe(self): p = BufferedPipe() - self.assert_(not p.read_ready()) + self.assertTrue(not p.read_ready()) p.feed('hello.') - self.assert_(p.read_ready()) + self.assertTrue(p.read_ready()) data = p.read(6) - self.assertEquals('hello.', data) + self.assertEqual(b'hello.', data) p.feed('plus/minus') - self.assertEquals('plu', p.read(3)) - self.assertEquals('s/m', p.read(3)) - self.assertEquals('inus', p.read(4)) + 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.assert_(not p.read_ready()) - self.assertEquals('', p.read(1)) + self.assertTrue(not p.read_ready()) + self.assertEqual(b'', p.read(1)) def test_2_delay(self): p = BufferedPipe() - self.assert_(not p.read_ready()) + self.assertTrue(not p.read_ready()) threading.Thread(target=delay_thread, args=(p,)).start() - self.assertEquals('a', p.read(1, 0.1)) + self.assertEqual(b'a', p.read(1, 0.1)) try: p.read(1, 0.1) - self.assert_(False) + self.assertTrue(False) except PipeTimeout: pass - self.assertEquals('b', p.read(1, 1.0)) - self.assertEquals('', 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.assertEquals('', data) + self.assertEqual(b'', data) def test_4_or_pipe(self): p = pipe.make_pipe() |