diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_client.py | 25 | ||||
| -rwxr-xr-x | tests/test_sftp.py | 20 | ||||
| -rw-r--r-- | tests/test_transport.py | 18 |
3 files changed, 61 insertions, 2 deletions
diff --git a/tests/test_client.py b/tests/test_client.py index 49f2a64a..12022172 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -20,7 +20,10 @@ Some unit tests for SSHClient. """ +from __future__ import with_statement + import gc + import socket from tempfile import mkstemp import threading @@ -298,6 +301,28 @@ class SSHClientTest (unittest.TestCase): self.assertTrue(p() is None) + def test_client_can_be_used_as_context_manager(self): + """ + verify that an SSHClient can be used a context manager + """ + threading.Thread(target=self._run).start() + host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = paramiko.RSAKey(data=host_key.asbytes()) + + with paramiko.SSHClient() as tc: + self.tc = tc + self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + self.assertEquals(0, len(self.tc.get_host_keys())) + self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion') + + self.event.wait(1.0) + self.assertTrue(self.event.isSet()) + self.assertTrue(self.ts.is_active()) + + self.assertTrue(self.tc._transport is not None) + + self.assertTrue(self.tc._transport is None) + def test_7_banner_timeout(self): """ verify that the SSHClient has a configurable banner timeout. diff --git a/tests/test_sftp.py b/tests/test_sftp.py index 1ae9781d..72c7ba03 100755 --- a/tests/test_sftp.py +++ b/tests/test_sftp.py @@ -23,12 +23,13 @@ a real actual sftp server is contacted, and a new folder is created there to do test file operations in (so no existing files will be harmed). """ -from binascii import hexlify import os +import socket import sys -import warnings import threading import unittest +import warnings +from binascii import hexlify from tempfile import mkstemp import paramiko @@ -195,6 +196,21 @@ class SFTPTest (unittest.TestCase): pass sftp = paramiko.SFTP.from_transport(tc) + def test_2_sftp_can_be_used_as_context_manager(self): + """ + verify that the sftp session is closed when exiting the context manager + """ + global sftp + with sftp: + pass + try: + sftp.open(FOLDER + '/test2', 'w') + self.fail('expected exception') + except (EOFError, socket.error): + pass + finally: + sftp = paramiko.SFTP.from_transport(tc) + def test_3_write(self): """ verify that a file can be created and written, and the size is correct. diff --git a/tests/test_transport.py b/tests/test_transport.py index 344d64b8..50b1d86b 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -20,6 +20,8 @@ Some unit tests for the ssh2 protocol in Transport. """ +from __future__ import with_statement + from binascii import hexlify import select import socket @@ -281,6 +283,22 @@ class TransportTest(unittest.TestCase): self.assertEqual('Hello there.\n', f.readline()) self.assertEqual('This is on stderr.\n', f.readline()) self.assertEqual('', f.readline()) + + def test_6a_channel_can_be_used_as_context_manager(self): + """ + verify that exec_command() does something reasonable. + """ + self.setup_test_server() + + with self.tc.open_session() as chan: + with self.ts.accept(1.0) as schan: + chan.exec_command('yes') + schan.send('Hello there.\n') + schan.close() + + f = chan.makefile() + self.assertEqual('Hello there.\n', f.readline()) + self.assertEqual('', f.readline()) def test_7_invoke_shell(self): """ |