diff options
Diffstat (limited to 'lib/python2.7/test/test_telnetlib.py')
| -rw-r--r-- | lib/python2.7/test/test_telnetlib.py | 454 |
1 files changed, 454 insertions, 0 deletions
diff --git a/lib/python2.7/test/test_telnetlib.py b/lib/python2.7/test/test_telnetlib.py new file mode 100644 index 0000000..7fdb49e --- /dev/null +++ b/lib/python2.7/test/test_telnetlib.py @@ -0,0 +1,454 @@ +import socket +import telnetlib +import time +import Queue + +import unittest +from unittest import TestCase +from test import test_support +threading = test_support.import_module('threading') + +HOST = test_support.HOST +EOF_sigil = object() + +def server(evt, serv, dataq=None): + """ Open a tcp server in three steps + 1) set evt to true to let the parent know we are ready + 2) [optional] if is not False, write the list of data from dataq.get() + to the socket. + """ + serv.listen(5) + evt.set() + try: + conn, addr = serv.accept() + if dataq: + data = '' + new_data = dataq.get(True, 0.5) + dataq.task_done() + for item in new_data: + if item == EOF_sigil: + break + if type(item) in [int, float]: + time.sleep(item) + else: + data += item + written = conn.send(data) + data = data[written:] + conn.close() + except socket.timeout: + pass + finally: + serv.close() + +class GeneralTests(TestCase): + + def setUp(self): + self.evt = threading.Event() + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(60) # Safety net. Look issue 11812 + self.port = test_support.bind_port(self.sock) + self.thread = threading.Thread(target=server, args=(self.evt,self.sock)) + self.thread.setDaemon(True) + self.thread.start() + self.evt.wait() + + def tearDown(self): + self.thread.join() + + def testBasic(self): + # connects + telnet = telnetlib.Telnet(HOST, self.port) + telnet.sock.close() + + def testTimeoutDefault(self): + self.assertTrue(socket.getdefaulttimeout() is None) + socket.setdefaulttimeout(30) + try: + telnet = telnetlib.Telnet(HOST, self.port) + finally: + socket.setdefaulttimeout(None) + self.assertEqual(telnet.sock.gettimeout(), 30) + telnet.sock.close() + + def testTimeoutNone(self): + # None, having other default + self.assertTrue(socket.getdefaulttimeout() is None) + socket.setdefaulttimeout(30) + try: + telnet = telnetlib.Telnet(HOST, self.port, timeout=None) + finally: + socket.setdefaulttimeout(None) + self.assertTrue(telnet.sock.gettimeout() is None) + telnet.sock.close() + + def testTimeoutValue(self): + telnet = telnetlib.Telnet(HOST, self.port, timeout=30) + self.assertEqual(telnet.sock.gettimeout(), 30) + telnet.sock.close() + + def testTimeoutOpen(self): + telnet = telnetlib.Telnet() + telnet.open(HOST, self.port, timeout=30) + self.assertEqual(telnet.sock.gettimeout(), 30) + telnet.sock.close() + +def _read_setUp(self): + self.evt = threading.Event() + self.dataq = Queue.Queue() + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(10) + self.port = test_support.bind_port(self.sock) + self.thread = threading.Thread(target=server, args=(self.evt,self.sock, self.dataq)) + self.thread.start() + self.evt.wait() + +def _read_tearDown(self): + self.thread.join() + +class ReadTests(TestCase): + setUp = _read_setUp + tearDown = _read_tearDown + + # use a similar approach to testing timeouts as test_timeout.py + # these will never pass 100% but make the fuzz big enough that it is rare + block_long = 0.6 + block_short = 0.3 + def test_read_until_A(self): + """ + read_until(expected, [timeout]) + Read until the expected string has been seen, or a timeout is + hit (default is no timeout); may block. + """ + want = ['x' * 10, 'match', 'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_until('match') + self.assertEqual(data, ''.join(want[:-2])) + + def test_read_until_B(self): + # test the timeout - it does NOT raise socket.timeout + want = ['hello', self.block_long, 'not seen', EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_until('not seen', self.block_short) + self.assertEqual(data, want[0]) + self.assertEqual(telnet.read_all(), 'not seen') + + def test_read_until_with_poll(self): + """Use select.poll() to implement telnet.read_until().""" + want = ['x' * 10, 'match', 'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + if not telnet._has_poll: + raise unittest.SkipTest('select.poll() is required') + telnet._has_poll = True + self.dataq.join() + data = telnet.read_until('match') + self.assertEqual(data, ''.join(want[:-2])) + + def test_read_until_with_select(self): + """Use select.select() to implement telnet.read_until().""" + want = ['x' * 10, 'match', 'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + telnet._has_poll = False + self.dataq.join() + data = telnet.read_until('match') + self.assertEqual(data, ''.join(want[:-2])) + + def test_read_all_A(self): + """ + read_all() + Read all data until EOF; may block. + """ + want = ['x' * 500, 'y' * 500, 'z' * 500, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_all() + self.assertEqual(data, ''.join(want[:-1])) + return + + def _test_blocking(self, func): + self.dataq.put([self.block_long, EOF_sigil]) + self.dataq.join() + start = time.time() + data = func() + self.assertTrue(self.block_short <= time.time() - start) + + def test_read_all_B(self): + self._test_blocking(telnetlib.Telnet(HOST, self.port).read_all) + + def test_read_all_C(self): + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + telnet.read_all() + telnet.read_all() # shouldn't raise + + def test_read_some_A(self): + """ + read_some() + Read at least one byte or EOF; may block. + """ + # test 'at least one byte' + want = ['x' * 500, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + data = telnet.read_all() + self.assertTrue(len(data) >= 1) + + def test_read_some_B(self): + # test EOF + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + self.assertEqual('', telnet.read_some()) + + def test_read_some_C(self): + self._test_blocking(telnetlib.Telnet(HOST, self.port).read_some) + + def _test_read_any_eager_A(self, func_name): + """ + read_very_eager() + Read all data available already queued or on the socket, + without blocking. + """ + want = [self.block_long, 'x' * 100, 'y' * 100, EOF_sigil] + expects = want[1] + want[2] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + func = getattr(telnet, func_name) + data = '' + while True: + try: + data += func() + self.assertTrue(expects.startswith(data)) + except EOFError: + break + self.assertEqual(expects, data) + + def _test_read_any_eager_B(self, func_name): + # test EOF + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + time.sleep(self.block_short) + func = getattr(telnet, func_name) + self.assertRaises(EOFError, func) + + # read_eager and read_very_eager make the same gaurantees + # (they behave differently but we only test the gaurantees) + def test_read_very_eager_A(self): + self._test_read_any_eager_A('read_very_eager') + def test_read_very_eager_B(self): + self._test_read_any_eager_B('read_very_eager') + def test_read_eager_A(self): + self._test_read_any_eager_A('read_eager') + def test_read_eager_B(self): + self._test_read_any_eager_B('read_eager') + # NB -- we need to test the IAC block which is mentioned in the docstring + # but not in the module docs + + def _test_read_any_lazy_B(self, func_name): + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + func = getattr(telnet, func_name) + telnet.fill_rawq() + self.assertRaises(EOFError, func) + + def test_read_lazy_A(self): + want = ['x' * 100, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + time.sleep(self.block_short) + self.assertEqual('', telnet.read_lazy()) + data = '' + while True: + try: + read_data = telnet.read_lazy() + data += read_data + if not read_data: + telnet.fill_rawq() + except EOFError: + break + self.assertTrue(want[0].startswith(data)) + self.assertEqual(data, want[0]) + + def test_read_lazy_B(self): + self._test_read_any_lazy_B('read_lazy') + + def test_read_very_lazy_A(self): + want = ['x' * 100, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + time.sleep(self.block_short) + self.assertEqual('', telnet.read_very_lazy()) + data = '' + while True: + try: + read_data = telnet.read_very_lazy() + except EOFError: + break + data += read_data + if not read_data: + telnet.fill_rawq() + self.assertEqual('', telnet.cookedq) + telnet.process_rawq() + self.assertTrue(want[0].startswith(data)) + self.assertEqual(data, want[0]) + + def test_read_very_lazy_B(self): + self._test_read_any_lazy_B('read_very_lazy') + +class nego_collector(object): + def __init__(self, sb_getter=None): + self.seen = '' + self.sb_getter = sb_getter + self.sb_seen = '' + + def do_nego(self, sock, cmd, opt): + self.seen += cmd + opt + if cmd == tl.SE and self.sb_getter: + sb_data = self.sb_getter() + self.sb_seen += sb_data + +tl = telnetlib +class OptionTests(TestCase): + setUp = _read_setUp + tearDown = _read_tearDown + # RFC 854 commands + cmds = [tl.AO, tl.AYT, tl.BRK, tl.EC, tl.EL, tl.GA, tl.IP, tl.NOP] + + def _test_command(self, data): + """ helper for testing IAC + cmd """ + self.setUp() + self.dataq.put(data) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + nego = nego_collector() + telnet.set_option_negotiation_callback(nego.do_nego) + txt = telnet.read_all() + cmd = nego.seen + self.assertTrue(len(cmd) > 0) # we expect at least one command + self.assertIn(cmd[0], self.cmds) + self.assertEqual(cmd[1], tl.NOOPT) + self.assertEqual(len(''.join(data[:-1])), len(txt + cmd)) + nego.sb_getter = None # break the nego => telnet cycle + self.tearDown() + + def test_IAC_commands(self): + # reset our setup + self.dataq.put([EOF_sigil]) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + self.tearDown() + + for cmd in self.cmds: + self._test_command(['x' * 100, tl.IAC + cmd, 'y'*100, EOF_sigil]) + self._test_command(['x' * 10, tl.IAC + cmd, 'y'*10, EOF_sigil]) + self._test_command([tl.IAC + cmd, EOF_sigil]) + # all at once + self._test_command([tl.IAC + cmd for (cmd) in self.cmds] + [EOF_sigil]) + self.assertEqual('', telnet.read_sb_data()) + + def test_SB_commands(self): + # RFC 855, subnegotiations portion + send = [tl.IAC + tl.SB + tl.IAC + tl.SE, + tl.IAC + tl.SB + tl.IAC + tl.IAC + tl.IAC + tl.SE, + tl.IAC + tl.SB + tl.IAC + tl.IAC + 'aa' + tl.IAC + tl.SE, + tl.IAC + tl.SB + 'bb' + tl.IAC + tl.IAC + tl.IAC + tl.SE, + tl.IAC + tl.SB + 'cc' + tl.IAC + tl.IAC + 'dd' + tl.IAC + tl.SE, + EOF_sigil, + ] + self.dataq.put(send) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + nego = nego_collector(telnet.read_sb_data) + telnet.set_option_negotiation_callback(nego.do_nego) + txt = telnet.read_all() + self.assertEqual(txt, '') + want_sb_data = tl.IAC + tl.IAC + 'aabb' + tl.IAC + 'cc' + tl.IAC + 'dd' + self.assertEqual(nego.sb_seen, want_sb_data) + self.assertEqual('', telnet.read_sb_data()) + nego.sb_getter = None # break the nego => telnet cycle + + +class ExpectTests(TestCase): + def setUp(self): + self.evt = threading.Event() + self.dataq = Queue.Queue() + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.settimeout(10) + self.port = test_support.bind_port(self.sock) + self.thread = threading.Thread(target=server, args=(self.evt,self.sock, + self.dataq)) + self.thread.start() + self.evt.wait() + + def tearDown(self): + self.thread.join() + + # use a similar approach to testing timeouts as test_timeout.py + # these will never pass 100% but make the fuzz big enough that it is rare + block_long = 0.6 + block_short = 0.3 + def test_expect_A(self): + """ + expect(expected, [timeout]) + Read until the expected string has been seen, or a timeout is + hit (default is no timeout); may block. + """ + want = ['x' * 10, 'match', 'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + (_,_,data) = telnet.expect(['match']) + self.assertEqual(data, ''.join(want[:-2])) + + def test_expect_B(self): + # test the timeout - it does NOT raise socket.timeout + want = ['hello', self.block_long, 'not seen', EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + self.dataq.join() + (_,_,data) = telnet.expect(['not seen'], self.block_short) + self.assertEqual(data, want[0]) + self.assertEqual(telnet.read_all(), 'not seen') + + def test_expect_with_poll(self): + """Use select.poll() to implement telnet.expect().""" + want = ['x' * 10, 'match', 'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + if not telnet._has_poll: + raise unittest.SkipTest('select.poll() is required') + telnet._has_poll = True + self.dataq.join() + (_,_,data) = telnet.expect(['match']) + self.assertEqual(data, ''.join(want[:-2])) + + def test_expect_with_select(self): + """Use select.select() to implement telnet.expect().""" + want = ['x' * 10, 'match', 'y' * 10, EOF_sigil] + self.dataq.put(want) + telnet = telnetlib.Telnet(HOST, self.port) + telnet._has_poll = False + self.dataq.join() + (_,_,data) = telnet.expect(['match']) + self.assertEqual(data, ''.join(want[:-2])) + + +def test_main(verbose=None): + test_support.run_unittest(GeneralTests, ReadTests, OptionTests, + ExpectTests) + +if __name__ == '__main__': + test_main() |