summaryrefslogtreecommitdiffhomepage
path: root/tests/test_buffered_pipe.py
diff options
context:
space:
mode:
authorJeff Forcier <jeff@bitprophet.org>2014-03-05 17:03:37 -0800
committerJeff Forcier <jeff@bitprophet.org>2014-03-05 17:03:37 -0800
commitb2be63ec623b5944f9b84cac8b8f41aeb2b42fb7 (patch)
tree389e17b0c08cd34872a2e3afbc34860ab44fb3ed /tests/test_buffered_pipe.py
parentbd61c7c0a9a4a2020d0acfb6a01e9ec85bb43b8e (diff)
parentae078f51d622931954e47e78029a889c4e721a05 (diff)
Merge remote-tracking branch 'scottkmaxwell/py3-support-without-py25' into python3
Conflicts: dev-requirements.txt paramiko/__init__.py paramiko/file.py paramiko/hostkeys.py paramiko/message.py paramiko/proxy.py paramiko/server.py paramiko/transport.py paramiko/util.py paramiko/win_pageant.py setup.py
Diffstat (limited to 'tests/test_buffered_pipe.py')
-rw-r--r--tests/test_buffered_pipe.py31
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()