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()