diff options
author | Jeremy T. Bouse <jbouse@debian.org> | 2014-09-22 09:21:39 -0400 |
---|---|---|
committer | Jeremy T. Bouse <jbouse@debian.org> | 2014-09-22 09:21:39 -0400 |
commit | ccd643cdbd7ba752727d62051058a4454451414d (patch) | |
tree | 9bc89a6629132f474b3297735069d025145a965c /tests/test_packetizer.py | |
parent | 588a4823436454e8968ee36ae95ff92e1cddc3f7 (diff) | |
download | python-paramiko-ccd643cdbd7ba752727d62051058a4454451414d.tar python-paramiko-ccd643cdbd7ba752727d62051058a4454451414d.tar.gz |
Imported Upstream version 1.15.0upstream/1.15.0
Diffstat (limited to 'tests/test_packetizer.py')
-rw-r--r-- | tests/test_packetizer.py | 46 |
1 files changed, 46 insertions, 0 deletions
diff --git a/tests/test_packetizer.py b/tests/test_packetizer.py index a8c0f97..8faec03 100644 --- a/tests/test_packetizer.py +++ b/tests/test_packetizer.py @@ -74,3 +74,49 @@ class PacketizerTest (unittest.TestCase): self.assertEqual(100, m.get_int()) self.assertEqual(1, m.get_int()) self.assertEqual(900, m.get_int()) + + def test_3_closed(self): + rsock = LoopSocket() + wsock = LoopSocket() + rsock.link(wsock) + p = Packetizer(wsock) + p.set_log(util.get_logger('paramiko.transport')) + p.set_hexdump(True) + cipher = AES.new(zero_byte * 16, AES.MODE_CBC, x55 * 16) + p.set_outbound_cipher(cipher, 16, sha1, 12, x1f * 20) + + # message has to be at least 16 bytes long, so we'll have at least one + # block of data encrypted that contains zero random padding bytes + m = Message() + m.add_byte(byte_chr(100)) + m.add_int(100) + m.add_int(1) + m.add_int(900) + wsock.send = lambda x: 0 + from functools import wraps + import errno + import os + import signal + + class TimeoutError(Exception): + pass + + def timeout(seconds=1, error_message=os.strerror(errno.ETIME)): + def decorator(func): + def _handle_timeout(signum, frame): + raise TimeoutError(error_message) + + def wrapper(*args, **kwargs): + signal.signal(signal.SIGALRM, _handle_timeout) + signal.alarm(seconds) + try: + result = func(*args, **kwargs) + finally: + signal.alarm(0) + return result + + return wraps(func)(wrapper) + + return decorator + send = timeout()(p.send_message) + self.assertRaises(EOFError, send, m) |