diff options
| -rw-r--r-- | tests/test_auth.py | 25 |
1 files changed, 25 insertions, 0 deletions
diff --git a/tests/test_auth.py b/tests/test_auth.py index 23517790..5c0bd264 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -23,6 +23,7 @@ Some unit tests for authenticating over a Transport. import sys import threading import unittest +from time import sleep from paramiko import Transport, ServerInterface, RSAKey, DSSKey, \ BadAuthenticationType, InteractiveQuery, \ @@ -73,6 +74,9 @@ class NullServer (ServerInterface): return AUTH_SUCCESSFUL if username == 'bad-server': raise Exception("Ack!") + if username == 'unresponsive-server': + sleep(5) + return AUTH_SUCCESSFUL return AUTH_FAILED def check_auth_publickey(self, username, key): @@ -232,3 +236,24 @@ class AuthTest (unittest.TestCase): except: etype, evalue, etb = sys.exc_info() self.assertTrue(issubclass(etype, AuthenticationException)) + + def test_9_auth_non_responsive(self): + """ + verify that authentication times out if server takes to long to + respond (or never responds). + """ + auth_timeout = self.tc.auth_timeout + self.tc.auth_timeout = 2 # Reduce to 2 seconds to speed up test + + try: + self.start_server() + self.tc.connect() + try: + remain = self.tc.auth_password('unresponsive-server', 'hello') + except: + etype, evalue, etb = sys.exc_info() + self.assertTrue(issubclass(etype, AuthenticationException)) + self.assertIn('Authentication timeout', str(evalue)) + finally: + # Restore value + self.tc.auth_timeout = auth_timeout |