diff options
Diffstat (limited to 'tests')
| -rw-r--r-- | tests/test_client.py | 24 | ||||
| -rw-r--r-- | tests/test_gssapi.py | 40 | ||||
| -rwxr-xr-x | tests/test_sftp.py | 20 | ||||
| -rw-r--r-- | tests/test_transport.py | 18 |
4 files changed, 70 insertions, 32 deletions
diff --git a/tests/test_client.py b/tests/test_client.py index 6fda7f5e..28d1cb46 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -20,6 +20,8 @@ Some unit tests for SSHClient. """ +from __future__ import with_statement + import socket from tempfile import mkstemp import threading @@ -299,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_gssapi.py b/tests/test_gssapi.py index 0d3df72c..a328dd65 100644 --- a/tests/test_gssapi.py +++ b/tests/test_gssapi.py @@ -72,9 +72,7 @@ class GSSAPITest(unittest.TestCase): gss_flags = (gssapi.C_PROT_READY_FLAG, gssapi.C_INTEG_FLAG, gssapi.C_DELEG_FLAG) - """ - Initialize a GSS-API context. - """ + # Initialize a GSS-API context. ctx = gssapi.Context() ctx.flags = gss_flags krb5_oid = gssapi.OID.mech_from_string(krb5_mech) @@ -87,41 +85,31 @@ class GSSAPITest(unittest.TestCase): c_token = gss_ctxt.step(c_token) gss_ctxt_status = gss_ctxt.established self.assertEquals(False, gss_ctxt_status) - """ - Accept a GSS-API context. - """ + # Accept a GSS-API context. gss_srv_ctxt = gssapi.AcceptContext() s_token = gss_srv_ctxt.step(c_token) gss_ctxt_status = gss_srv_ctxt.established self.assertNotEquals(None, s_token) self.assertEquals(True, gss_ctxt_status) - """ - Establish the client context - """ + # Establish the client context c_token = gss_ctxt.step(s_token) self.assertEquals(None, c_token) else: while not gss_ctxt.established: c_token = gss_ctxt.step(c_token) self.assertNotEquals(None, c_token) - """ - Build MIC - """ + # Build MIC mic_token = gss_ctxt.get_mic(mic_msg) if server_mode: - """ - Check MIC - """ + # Check MIC status = gss_srv_ctxt.verify_mic(mic_msg, mic_token) self.assertEquals(0, status) else: gss_flags = sspicon.ISC_REQ_INTEGRITY |\ sspicon.ISC_REQ_MUTUAL_AUTH |\ sspicon.ISC_REQ_DELEGATE - """ - Initialize a GSS-API context. - """ + # Initialize a GSS-API context. target_name = "host/" + socket.getfqdn(targ_name) gss_ctxt = sspi.ClientAuth("Kerberos", scflags=gss_flags, @@ -130,26 +118,18 @@ class GSSAPITest(unittest.TestCase): error, token = gss_ctxt.authorize(c_token) c_token = token[0].Buffer self.assertEquals(0, error) - """ - Accept a GSS-API context. - """ + # Accept a GSS-API context. gss_srv_ctxt = sspi.ServerAuth("Kerberos", spn=target_name) error, token = gss_srv_ctxt.authorize(c_token) s_token = token[0].Buffer - """ - Establish the context. - """ + # Establish the context. error, token = gss_ctxt.authorize(s_token) c_token = token[0].Buffer self.assertEquals(None, c_token) self.assertEquals(0, error) - """ - Build MIC - """ + # Build MIC mic_token = gss_ctxt.sign(mic_msg) - """ - Check MIC - """ + # Check MIC gss_srv_ctxt.verify(mic_msg, mic_token) else: error, token = gss_ctxt.authorize(c_token) 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): """ |