summaryrefslogtreecommitdiffhomepage
path: root/tests
diff options
context:
space:
mode:
Diffstat (limited to 'tests')
-rw-r--r--tests/stub_sftp.py1
-rw-r--r--tests/test_auth.py6
-rw-r--r--tests/test_client.py14
-rw-r--r--tests/test_kex.py49
-rw-r--r--tests/test_pkey.py10
-rw-r--r--tests/test_sftp.py25
6 files changed, 61 insertions, 44 deletions
diff --git a/tests/stub_sftp.py b/tests/stub_sftp.py
index 13e7f3b4..100076d6 100644
--- a/tests/stub_sftp.py
+++ b/tests/stub_sftp.py
@@ -168,7 +168,6 @@ class StubSFTPServer(SFTPServerInterface):
return SFTPServer.convert_errno(e.errno)
return SFTP_OK
-
def mkdir(self, path, attr):
path = self._realpath(path)
try:
diff --git a/tests/test_auth.py b/tests/test_auth.py
index 14d465ca..6358a053 100644
--- a/tests/test_auth.py
+++ b/tests/test_auth.py
@@ -85,7 +85,7 @@ class NullServer(ServerInterface):
return AUTH_SUCCESSFUL
if username == "bad-server":
raise Exception("Ack!")
- if username == 'unresponsive-server':
+ if username == "unresponsive-server":
sleep(5)
return AUTH_SUCCESSFUL
return AUTH_FAILED
@@ -264,8 +264,8 @@ class AuthTest(unittest.TestCase):
self.start_server()
self.tc.connect()
try:
- remain = self.tc.auth_password('unresponsive-server', 'hello')
+ remain = self.tc.auth_password("unresponsive-server", "hello")
except:
etype, evalue, etb = sys.exc_info()
self.assertTrue(issubclass(etype, AuthenticationException))
- self.assertTrue('Authentication timeout' in str(evalue))
+ self.assertTrue("Authentication timeout" in str(evalue))
diff --git a/tests/test_client.py b/tests/test_client.py
index 53ba53e4..87f7bcb2 100644
--- a/tests/test_client.py
+++ b/tests/test_client.py
@@ -41,10 +41,10 @@ from .util import _support, slow
FINGERPRINTS = {
- 'ssh-dss': b'\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c',
- 'ssh-rsa': b'\x60\x73\x38\x44\xcb\x51\x86\x65\x7f\xde\xda\xa2\x2b\x5a\x57\xd5',
- 'ecdsa-sha2-nistp256': b'\x25\x19\xeb\x55\xe6\xa1\x47\xff\x4f\x38\xd2\x75\x6f\xa5\xd5\x60',
- 'ssh-ed25519': b'\xb3\xd5"\xaa\xf9u^\xe8\xcd\x0e\xea\x02\xb9)\xa2\x80',
+ "ssh-dss": b"\x44\x78\xf0\xb9\xa2\x3c\xc5\x18\x20\x09\xff\x75\x5b\xc1\xd2\x6c",
+ "ssh-rsa": b"\x60\x73\x38\x44\xcb\x51\x86\x65\x7f\xde\xda\xa2\x2b\x5a\x57\xd5",
+ "ecdsa-sha2-nistp256": b"\x25\x19\xeb\x55\xe6\xa1\x47\xff\x4f\x38\xd2\x75\x6f\xa5\xd5\x60",
+ "ssh-ed25519": b'\xb3\xd5"\xaa\xf9u^\xe8\xcd\x0e\xea\x02\xb9)\xa2\x80',
}
@@ -62,7 +62,7 @@ class NullServer(paramiko.ServerInterface):
def check_auth_password(self, username, password):
if (username == "slowdive") and (password == "pygmalion"):
return paramiko.AUTH_SUCCESSFUL
- if (username == 'slowdive') and (password == 'unresponsive-server'):
+ if (username == "slowdive") and (password == "unresponsive-server"):
time.sleep(5)
return paramiko.AUTH_SUCCESSFUL
return paramiko.AUTH_FAILED
@@ -208,7 +208,7 @@ class SSHClientTest(unittest.TestCase):
self._test_connection(key_filename=_support("test_ecdsa_256.key"))
def test_client_ed25519(self):
- self._test_connection(key_filename=_support('test_ed25519.key'))
+ self._test_connection(key_filename=_support("test_ed25519.key"))
def test_3_multiple_key_files(self):
"""
@@ -408,7 +408,7 @@ class SSHClientTest(unittest.TestCase):
self.assertRaises(
AuthenticationException,
self._test_connection,
- password='unresponsive-server',
+ password="unresponsive-server",
auth_timeout=0.5,
)
diff --git a/tests/test_kex.py b/tests/test_kex.py
index b9b59db1..65eb9a17 100644
--- a/tests/test_kex.py
+++ b/tests/test_kex.py
@@ -40,15 +40,24 @@ from cryptography.hazmat.primitives.asymmetric import ec
def dummy_urandom(n):
return byte_chr(0xcc) * n
+
def dummy_generate_key_pair(obj):
private_key_value = 94761803665136558137557783047955027733968423115106677159790289642479432803037
public_key_numbers = "042bdab212fa8ba1b7c843301682a4db424d307246c7e1e6083c41d9ca7b098bf30b3d63e2ec6278488c135360456cc054b3444ecc45998c08894cbc1370f5f989"
- public_key_numbers_obj = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), unhexlify(public_key_numbers))
- obj.P = ec.EllipticCurvePrivateNumbers(private_value=private_key_value, public_numbers=public_key_numbers_obj).private_key(default_backend())
+ public_key_numbers_obj = ec.EllipticCurvePublicNumbers.from_encoded_point(
+ ec.SECP256R1(), unhexlify(public_key_numbers)
+ )
+ obj.P = ec.EllipticCurvePrivateNumbers(
+ private_value=private_key_value, public_numbers=public_key_numbers_obj
+ ).private_key(default_backend())
if obj.transport.server_mode:
- obj.Q_S = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), unhexlify(public_key_numbers)).public_key(default_backend())
+ obj.Q_S = ec.EllipticCurvePublicNumbers.from_encoded_point(
+ ec.SECP256R1(), unhexlify(public_key_numbers)
+ ).public_key(default_backend())
return
- obj.Q_C = ec.EllipticCurvePublicNumbers.from_encoded_point(ec.SECP256R1(), unhexlify(public_key_numbers)).public_key(default_backend())
+ obj.Q_C = ec.EllipticCurvePublicNumbers.from_encoded_point(
+ ec.SECP256R1(), unhexlify(public_key_numbers)
+ ).public_key(default_backend())
class FakeKey(object):
@@ -446,20 +455,24 @@ class KexTest(unittest.TestCase):
transport.server_mode = False
kex = KexNistp256(transport)
kex.start_kex()
- self.assertEqual((paramiko.kex_ecdh_nist._MSG_KEXECDH_REPLY,), transport._expect)
+ self.assertEqual(
+ (paramiko.kex_ecdh_nist._MSG_KEXECDH_REPLY,), transport._expect
+ )
- #fake reply
+ # fake reply
msg = Message()
- msg.add_string('fake-host-key')
- Q_S = unhexlify("043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210")
+ msg.add_string("fake-host-key")
+ Q_S = unhexlify(
+ "043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210"
+ )
msg.add_string(Q_S)
- msg.add_string('fake-sig')
+ msg.add_string("fake-sig")
msg.rewind()
kex.parse_next(paramiko.kex_ecdh_nist._MSG_KEXECDH_REPLY, msg)
- H = b'BAF7CE243A836037EB5D2221420F35C02B9AB6C957FE3BDE3369307B9612570A'
+ H = b"BAF7CE243A836037EB5D2221420F35C02B9AB6C957FE3BDE3369307B9612570A"
self.assertEqual(K, kex.transport._K)
self.assertEqual(H, hexlify(transport._H).upper())
- self.assertEqual((b'fake-host-key', b'fake-sig'), transport._verify)
+ self.assertEqual((b"fake-host-key", b"fake-sig"), transport._verify)
self.assertTrue(transport._activated)
def test_12_kex_nistp256_server(self):
@@ -468,12 +481,16 @@ class KexTest(unittest.TestCase):
transport.server_mode = True
kex = KexNistp256(transport)
kex.start_kex()
- self.assertEqual((paramiko.kex_ecdh_nist._MSG_KEXECDH_INIT,), transport._expect)
+ self.assertEqual(
+ (paramiko.kex_ecdh_nist._MSG_KEXECDH_INIT,), transport._expect
+ )
- #fake init
- msg=Message()
- Q_C = unhexlify("043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210")
- H = b'2EF4957AFD530DD3F05DBEABF68D724FACC060974DA9704F2AEE4C3DE861E7CA'
+ # fake init
+ msg = Message()
+ Q_C = unhexlify(
+ "043ae159594ba062efa121480e9ef136203fa9ec6b6e1f8723a321c16e62b945f573f3b822258cbcd094b9fa1c125cbfe5f043280893e66863cc0cb4dccbe70210"
+ )
+ H = b"2EF4957AFD530DD3F05DBEABF68D724FACC060974DA9704F2AEE4C3DE861E7CA"
msg.add_string(Q_C)
msg.rewind()
kex.parse_next(paramiko.kex_ecdh_nist._MSG_KEXECDH_INIT, msg)
diff --git a/tests/test_pkey.py b/tests/test_pkey.py
index a70671b5..3a1279b6 100644
--- a/tests/test_pkey.py
+++ b/tests/test_pkey.py
@@ -462,15 +462,15 @@ class KeyTest(unittest.TestCase):
self.assertEqual(str(key), comparable)
def test_ed25519(self):
- key1 = Ed25519Key.from_private_key_file(_support('test_ed25519.key'))
+ key1 = Ed25519Key.from_private_key_file(_support("test_ed25519.key"))
key2 = Ed25519Key.from_private_key_file(
- _support('test_ed25519_password.key'), b'abc123'
+ _support("test_ed25519_password.key"), b"abc123"
)
self.assertNotEqual(key1.asbytes(), key2.asbytes())
def test_ed25519_compare(self):
# verify that the private & public keys compare equal
- key = Ed25519Key.from_private_key_file(_support('test_ed25519.key'))
+ key = Ed25519Key.from_private_key_file(_support("test_ed25519.key"))
self.assertEqual(key, key)
pub = Ed25519Key(data=key.asbytes())
self.assertTrue(key.can_sign())
@@ -480,10 +480,10 @@ class KeyTest(unittest.TestCase):
def test_ed25519_nonbytes_password(self):
# https://github.com/paramiko/paramiko/issues/1039
key = Ed25519Key.from_private_key_file(
- _support('test_ed25519_password.key'),
+ _support("test_ed25519_password.key"),
# NOTE: not a bytes. Amusingly, the test above for same key DOES
# explicitly cast to bytes...code smell!
- 'abc123',
+ "abc123",
)
# No exception -> it's good. Meh.
diff --git a/tests/test_sftp.py b/tests/test_sftp.py
index a86fca5d..87c57340 100644
--- a/tests/test_sftp.py
+++ b/tests/test_sftp.py
@@ -189,35 +189,36 @@ class TestSFTP(object):
"""Test posix-rename@openssh.com protocol extension."""
try:
# first check that the normal rename works as specified
- with sftp.open(sftp.FOLDER + '/a', 'w') as f:
- f.write('one')
- sftp.rename(sftp.FOLDER + '/a', sftp.FOLDER + '/b')
- with sftp.open(sftp.FOLDER + '/a', 'w') as f:
- f.write('two')
+ with sftp.open(sftp.FOLDER + "/a", "w") as f:
+ f.write("one")
+ sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b")
+ with sftp.open(sftp.FOLDER + "/a", "w") as f:
+ f.write("two")
try:
- sftp.rename(sftp.FOLDER + '/a', sftp.FOLDER + '/b')
- self.assertTrue(False, 'no exception when rename-ing onto existing file')
+ sftp.rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b")
+ self.assertTrue(
+ False, "no exception when rename-ing onto existing file"
+ )
except (OSError, IOError):
pass
# now check with the posix_rename
- sftp.posix_rename(sftp.FOLDER + '/a', sftp.FOLDER + '/b')
- with sftp.open(sftp.FOLDER + '/b', 'r') as f:
+ sftp.posix_rename(sftp.FOLDER + "/a", sftp.FOLDER + "/b")
+ with sftp.open(sftp.FOLDER + "/b", "r") as f:
data = u(f.read())
err = "Contents of renamed file not the same as original file"
assert data == "two", err
finally:
try:
- sftp.remove(sftp.FOLDER + '/a')
+ sftp.remove(sftp.FOLDER + "/a")
except:
pass
try:
- sftp.remove(sftp.FOLDER + '/b')
+ sftp.remove(sftp.FOLDER + "/b")
except:
pass
-
def test_6_folder(self, sftp):
"""
create a temporary folder, verify that we can create a file in it, then