diff options
Diffstat (limited to 'tests/test_transport.py')
| -rw-r--r-- | tests/test_transport.py | 45 |
1 files changed, 31 insertions, 14 deletions
diff --git a/tests/test_transport.py b/tests/test_transport.py index d6884e4b..80f5e611 100644 --- a/tests/test_transport.py +++ b/tests/test_transport.py @@ -36,7 +36,7 @@ from paramiko import Transport, SecurityOptions, ServerInterface, RSAKey, DSSKey from paramiko import AUTH_FAILED, AUTH_SUCCESSFUL from paramiko import OPEN_SUCCEEDED, OPEN_FAILED_ADMINISTRATIVELY_PROHIBITED from paramiko.common import MSG_KEXINIT, cMSG_CHANNEL_WINDOW_ADJUST, \ - MIN_PACKET_SIZE, MAX_WINDOW_SIZE, \ + MIN_PACKET_SIZE, MIN_WINDOW_SIZE, MAX_WINDOW_SIZE, \ DEFAULT_WINDOW_SIZE, DEFAULT_MAX_PACKET_SIZE from paramiko.py3compat import bytes from paramiko.message import Message @@ -137,12 +137,12 @@ class TransportTest(unittest.TestCase): event = threading.Event() self.server = NullServer() - self.assertTrue(not event.isSet()) + self.assertTrue(not event.is_set()) self.ts.start_server(event, self.server) self.tc.connect(hostkey=public_host_key, username='slowdive', password='pygmalion') event.wait(1.0) - self.assertTrue(event.isSet()) + self.assertTrue(event.is_set()) self.assertTrue(self.ts.is_active()) def test_1_security_options(self): @@ -181,7 +181,7 @@ class TransportTest(unittest.TestCase): self.ts.add_server_key(host_key) event = threading.Event() server = NullServer() - self.assertTrue(not event.isSet()) + self.assertTrue(not event.is_set()) self.assertEqual(None, self.tc.get_username()) self.assertEqual(None, self.ts.get_username()) self.assertEqual(False, self.tc.is_authenticated()) @@ -190,7 +190,7 @@ class TransportTest(unittest.TestCase): self.tc.connect(hostkey=public_host_key, username='slowdive', password='pygmalion') event.wait(1.0) - self.assertTrue(event.isSet()) + self.assertTrue(event.is_set()) self.assertTrue(self.ts.is_active()) self.assertEqual('slowdive', self.tc.get_username()) self.assertEqual('slowdive', self.ts.get_username()) @@ -206,13 +206,13 @@ class TransportTest(unittest.TestCase): self.ts.add_server_key(host_key) event = threading.Event() server = NullServer() - self.assertTrue(not event.isSet()) + self.assertTrue(not event.is_set()) self.socks.send(LONG_BANNER) self.ts.start_server(event, server) self.tc.connect(hostkey=public_host_key, username='slowdive', password='pygmalion') event.wait(1.0) - self.assertTrue(event.isSet()) + self.assertTrue(event.is_set()) self.assertTrue(self.ts.is_active()) def test_4_special(self): @@ -683,7 +683,7 @@ class TransportTest(unittest.TestCase): def run(self): try: for i in range(1, 1+self.iterations): - if self.done_event.isSet(): + if self.done_event.is_set(): break self.watchdog_event.set() #print i, "SEND" @@ -702,7 +702,7 @@ class TransportTest(unittest.TestCase): def run(self): try: - while not self.done_event.isSet(): + while not self.done_event.is_set(): if self.chan.recv_ready(): chan.recv(65536) self.watchdog_event.set() @@ -756,12 +756,12 @@ class TransportTest(unittest.TestCase): # Act as a watchdog timer, checking deadlocked = False - while not deadlocked and not done_event.isSet(): + while not deadlocked and not done_event.is_set(): for event in (st.watchdog_event, rt.watchdog_event): event.wait(timeout) - if done_event.isSet(): + if done_event.is_set(): break - if not event.isSet(): + if not event.is_set(): deadlocked = True break event.clear() @@ -782,7 +782,7 @@ class TransportTest(unittest.TestCase): """ verify that we conform to the rfc of packet and window sizes. """ - for val, correct in [(32767, MIN_PACKET_SIZE), + for val, correct in [(4095, MIN_PACKET_SIZE), (None, DEFAULT_MAX_PACKET_SIZE), (2**32, MAX_WINDOW_SIZE)]: self.assertEqual(self.tc._sanitize_packet_size(val), correct) @@ -791,7 +791,24 @@ class TransportTest(unittest.TestCase): """ verify that we conform to the rfc of packet and window sizes. """ - for val, correct in [(32767, MIN_PACKET_SIZE), + for val, correct in [(32767, MIN_WINDOW_SIZE), (None, DEFAULT_WINDOW_SIZE), (2**32, MAX_WINDOW_SIZE)]: self.assertEqual(self.tc._sanitize_window_size(val), correct) + + def test_L_handshake_timeout(self): + """ + verify that we can get a hanshake timeout. + """ + host_key = RSAKey.from_private_key_file(test_path('test_rsa.key')) + public_host_key = RSAKey(data=host_key.asbytes()) + self.ts.add_server_key(host_key) + event = threading.Event() + server = NullServer() + self.assertTrue(not event.is_set()) + self.tc.handshake_timeout = 0.000000000001 + self.ts.start_server(event, server) + self.assertRaises(EOFError, self.tc.connect, + hostkey=public_host_key, + username='slowdive', + password='pygmalion') |