summaryrefslogtreecommitdiffhomepage
path: root/tests/test_kex_gss.py
diff options
context:
space:
mode:
authorChris Rose <offline@offby1.net>2018-05-17 10:13:38 -0400
committerChris Rose <offline@offby1.net>2018-05-17 10:13:38 -0400
commit7f2c35052183b400827d9949a68b41c90f90a32d (patch)
treefea4a1ec04b7ee3ced14d61e8b6cf3f479e22704 /tests/test_kex_gss.py
parent52551321a2297bdb966869fa719e584c868dd857 (diff)
Blacken Paramiko on 2.4
Diffstat (limited to 'tests/test_kex_gss.py')
-rw-r--r--tests/test_kex_gss.py48
1 files changed, 28 insertions, 20 deletions
diff --git a/tests/test_kex_gss.py b/tests/test_kex_gss.py
index 025d1faa..afddee08 100644
--- a/tests/test_kex_gss.py
+++ b/tests/test_kex_gss.py
@@ -34,14 +34,14 @@ import paramiko
from .util import needs_gssapi
-class NullServer (paramiko.ServerInterface):
+class NullServer(paramiko.ServerInterface):
def get_allowed_auths(self, username):
- return 'gssapi-keyex'
+ return "gssapi-keyex"
- def check_auth_gssapi_keyex(self, username,
- gss_authenticated=paramiko.AUTH_FAILED,
- cc_file=None):
+ def check_auth_gssapi_keyex(
+ self, username, gss_authenticated=paramiko.AUTH_FAILED, cc_file=None
+ ):
if gss_authenticated == paramiko.AUTH_SUCCESSFUL:
return paramiko.AUTH_SUCCESSFUL
return paramiko.AUTH_FAILED
@@ -54,13 +54,14 @@ class NullServer (paramiko.ServerInterface):
return paramiko.OPEN_SUCCEEDED
def check_channel_exec_request(self, channel, command):
- if command != 'yes':
+ if command != "yes":
return False
return True
@needs_gssapi
class GSSKexTest(unittest.TestCase):
+
@staticmethod
def init(username, hostname):
global krb5_principal, targ_name
@@ -86,13 +87,13 @@ class GSSKexTest(unittest.TestCase):
def _run(self):
self.socks, addr = self.sockl.accept()
self.ts = paramiko.Transport(self.socks, gss_kex=True)
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+ host_key = paramiko.RSAKey.from_private_key_file("tests/test_rsa.key")
self.ts.add_server_key(host_key)
self.ts.set_gss_host(targ_name)
try:
self.ts.load_server_moduli()
except:
- print ('(Failed to load moduli -- gex will be unsupported.)')
+ print("(Failed to load moduli -- gex will be unsupported.)")
server = NullServer()
self.ts.start_server(self.event, server)
@@ -102,14 +103,21 @@ class GSSKexTest(unittest.TestCase):
Diffie-Hellman Key Exchange and user authentication with the GSS-API
context created during key exchange.
"""
- host_key = paramiko.RSAKey.from_private_key_file('tests/test_rsa.key')
+ host_key = paramiko.RSAKey.from_private_key_file("tests/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.hostname, self.port),
- 'ssh-rsa', public_host_key)
- self.tc.connect(self.hostname, self.port, username=self.username,
- gss_auth=True, gss_kex=True, gss_host=gss_host)
+ self.tc.get_host_keys().add(
+ "[%s]:%d" % (self.hostname, self.port), "ssh-rsa", public_host_key
+ )
+ self.tc.connect(
+ self.hostname,
+ self.port,
+ username=self.username,
+ gss_auth=True,
+ gss_kex=True,
+ gss_host=gss_host,
+ )
self.event.wait(1.0)
self.assert_(self.event.is_set())
@@ -118,19 +126,19 @@ class GSSKexTest(unittest.TestCase):
self.assertEquals(True, self.ts.is_authenticated())
self.assertEquals(True, self.tc.get_transport().gss_kex_used)
- stdin, stdout, stderr = self.tc.exec_command('yes')
+ stdin, stdout, stderr = self.tc.exec_command("yes")
schan = self.ts.accept(1.0)
if rekey:
self.tc.get_transport().renegotiate_keys()
- schan.send('Hello there.\n')
- schan.send_stderr('This is on stderr.\n')
+ schan.send("Hello there.\n")
+ 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.assertEquals("Hello there.\n", stdout.readline())
+ self.assertEquals("", stdout.readline())
+ self.assertEquals("This is on stderr.\n", stderr.readline())
+ self.assertEquals("", stderr.readline())
stdin.close()
stdout.close()