summaryrefslogtreecommitdiffhomepage
path: root/tests/test_client.py
diff options
context:
space:
mode:
Diffstat (limited to 'tests/test_client.py')
-rw-r--r--tests/test_client.py109
1 files changed, 58 insertions, 51 deletions
diff --git a/tests/test_client.py b/tests/test_client.py
index fae1d329..b4724843 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -20,7 +20,6 @@
Some unit tests for SSHClient.
"""
-from __future__ import with_statement # Python 2.5 support
import socket
import threading
import time
@@ -29,8 +28,9 @@ import weakref
import warnings
import os
from binascii import hexlify
-
+from tests.util import test_path
import paramiko
+from paramiko.py3compat import b
class NullServer (paramiko.ServerInterface):
@@ -46,7 +46,7 @@ class NullServer (paramiko.ServerInterface):
return paramiko.AUTH_FAILED
def check_auth_publickey(self, username, key):
- if (key.get_name() == 'ssh-dss') and (hexlify(key.get_fingerprint()) == '4478f0b9a23cc5182009ff755bc1d26c'):
+ if (key.get_name() == 'ssh-dss') and key.get_fingerprint() == b'\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c':
return paramiko.AUTH_SUCCESSFUL
return paramiko.AUTH_FAILED
@@ -78,7 +78,7 @@ class SSHClientTest (unittest.TestCase):
def _run(self):
self.socks, addr = self.sockl.accept()
self.ts = paramiko.Transport(self.socks)
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+ host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key'))
self.ts.add_server_key(host_key)
server = NullServer()
self.ts.start_server(self.event, server)
@@ -88,18 +88,18 @@ class SSHClientTest (unittest.TestCase):
"""
verify that the SSHClient stuff works too.
"""
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
- public_host_key = paramiko.RSAKey(data=str(host_key))
+ host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key'))
+ public_host_key = paramiko.RSAKey(data=host_key.asbytes())
self.tc = paramiko.SSHClient()
self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port), 'ssh-rsa', public_host_key)
self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion')
self.event.wait(1.0)
- self.assert_(self.event.isSet())
- self.assert_(self.ts.is_active())
- self.assertEquals('slowdive', self.ts.get_username())
- self.assertEquals(True, self.ts.is_authenticated())
+ self.assertTrue(self.event.isSet())
+ self.assertTrue(self.ts.is_active())
+ self.assertEqual('slowdive', self.ts.get_username())
+ self.assertEqual(True, self.ts.is_authenticated())
stdin, stdout, stderr = self.tc.exec_command('yes')
schan = self.ts.accept(1.0)
@@ -108,10 +108,10 @@ class SSHClientTest (unittest.TestCase):
schan.send_stderr('This is on stderr.\n')
schan.close()
- self.assertEquals('Hello there.\n', stdout.readline())
- self.assertEquals('', stdout.readline())
- self.assertEquals('This is on stderr.\n', stderr.readline())
- self.assertEquals('', stderr.readline())
+ self.assertEqual('Hello there.\n', stdout.readline())
+ self.assertEqual('', stdout.readline())
+ self.assertEqual('This is on stderr.\n', stderr.readline())
+ self.assertEqual('', stderr.readline())
stdin.close()
stdout.close()
@@ -121,18 +121,18 @@ class SSHClientTest (unittest.TestCase):
"""
verify that SSHClient works with a DSA key.
"""
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
- public_host_key = paramiko.RSAKey(data=str(host_key))
+ host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key'))
+ public_host_key = paramiko.RSAKey(data=host_key.asbytes())
self.tc = paramiko.SSHClient()
self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port), 'ssh-rsa', public_host_key)
- self.tc.connect(self.addr, self.port, username='slowdive', key_filename='tests/test_dss.key')
+ self.tc.connect(self.addr, self.port, username='slowdive', key_filename=test_path('test_dss.key'))
self.event.wait(1.0)
- self.assert_(self.event.isSet())
- self.assert_(self.ts.is_active())
- self.assertEquals('slowdive', self.ts.get_username())
- self.assertEquals(True, self.ts.is_authenticated())
+ self.assertTrue(self.event.isSet())
+ self.assertTrue(self.ts.is_active())
+ self.assertEqual('slowdive', self.ts.get_username())
+ self.assertEqual(True, self.ts.is_authenticated())
stdin, stdout, stderr = self.tc.exec_command('yes')
schan = self.ts.accept(1.0)
@@ -141,10 +141,10 @@ class SSHClientTest (unittest.TestCase):
schan.send_stderr('This is on stderr.\n')
schan.close()
- self.assertEquals('Hello there.\n', stdout.readline())
- self.assertEquals('', stdout.readline())
- self.assertEquals('This is on stderr.\n', stderr.readline())
- self.assertEquals('', stderr.readline())
+ self.assertEqual('Hello there.\n', stdout.readline())
+ self.assertEqual('', stdout.readline())
+ self.assertEqual('This is on stderr.\n', stderr.readline())
+ self.assertEqual('', stderr.readline())
stdin.close()
stdout.close()
@@ -154,38 +154,38 @@ class SSHClientTest (unittest.TestCase):
"""
verify that SSHClient accepts and tries multiple key files.
"""
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
- public_host_key = paramiko.RSAKey(data=str(host_key))
+ host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key'))
+ public_host_key = paramiko.RSAKey(data=host_key.asbytes())
self.tc = paramiko.SSHClient()
self.tc.get_host_keys().add('[%s]:%d' % (self.addr, self.port), 'ssh-rsa', public_host_key)
- self.tc.connect(self.addr, self.port, username='slowdive', key_filename=[ 'tests/test_rsa.key', 'tests/test_dss.key' ])
+ self.tc.connect(self.addr, self.port, username='slowdive', key_filename=[ test_path('test_rsa.key'), test_path('test_dss.key') ])
self.event.wait(1.0)
- self.assert_(self.event.isSet())
- self.assert_(self.ts.is_active())
- self.assertEquals('slowdive', self.ts.get_username())
- self.assertEquals(True, self.ts.is_authenticated())
+ self.assertTrue(self.event.isSet())
+ self.assertTrue(self.ts.is_active())
+ self.assertEqual('slowdive', self.ts.get_username())
+ self.assertEqual(True, self.ts.is_authenticated())
def test_4_auto_add_policy(self):
"""
verify that SSHClient's AutoAddPolicy works.
"""
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
- public_host_key = paramiko.RSAKey(data=str(host_key))
+ host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key'))
+ public_host_key = paramiko.RSAKey(data=host_key.asbytes())
self.tc = paramiko.SSHClient()
self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy())
- self.assertEquals(0, len(self.tc.get_host_keys()))
+ self.assertEqual(0, len(self.tc.get_host_keys()))
self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion')
self.event.wait(1.0)
- self.assert_(self.event.isSet())
- self.assert_(self.ts.is_active())
- self.assertEquals('slowdive', self.ts.get_username())
- self.assertEquals(True, self.ts.is_authenticated())
- self.assertEquals(1, len(self.tc.get_host_keys()))
- self.assertEquals(public_host_key, self.tc.get_host_keys()['[%s]:%d' % (self.addr, self.port)]['ssh-rsa'])
+ self.assertTrue(self.event.isSet())
+ self.assertTrue(self.ts.is_active())
+ self.assertEqual('slowdive', self.ts.get_username())
+ self.assertEqual(True, self.ts.is_authenticated())
+ self.assertEqual(1, len(self.tc.get_host_keys()))
+ self.assertEqual(public_host_key, self.tc.get_host_keys()['[%s]:%d' % (self.addr, self.port)]['ssh-rsa'])
def test_5_save_host_keys(self):
"""
@@ -218,24 +218,31 @@ class SSHClientTest (unittest.TestCase):
verify that when an SSHClient is collected, its transport (and the
transport's packetizer) is closed.
"""
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
- public_host_key = paramiko.RSAKey(data=str(host_key))
+ host_key = paramiko.RSAKey.from_private_key_file(test_path('test_rsa.key'))
+ public_host_key = paramiko.RSAKey(data=host_key.asbytes())
self.tc = paramiko.SSHClient()
self.tc.set_missing_host_key_policy(paramiko.AutoAddPolicy())
- self.assertEquals(0, len(self.tc.get_host_keys()))
+ self.assertEqual(0, len(self.tc.get_host_keys()))
self.tc.connect(self.addr, self.port, username='slowdive', password='pygmalion')
self.event.wait(1.0)
- self.assert_(self.event.isSet())
- self.assert_(self.ts.is_active())
+ self.assertTrue(self.event.isSet())
+ self.assertTrue(self.ts.is_active())
p = weakref.ref(self.tc._transport.packetizer)
- self.assert_(p() is not None)
+ self.assertTrue(p() is not None)
+ self.tc.close()
del self.tc
+
# hrm, sometimes p isn't cleared right away. why is that?
- st = time.time()
- while (time.time() - st < 5.0) and (p() is not None):
- time.sleep(0.1)
- self.assert_(p() is None)
+ #st = time.time()
+ #while (time.time() - st < 5.0) and (p() is not None):
+ # time.sleep(0.1)
+
+ # instead of dumbly waiting for the GC to collect, force a collection
+ # to see whether the SSHClient object is deallocated correctly
+ import gc
+ gc.collect()
+ self.assertTrue(p() is None)
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-12 12:30:31 +0000