Changeset 391 for python/trunk/Lib/test/test_telnetlib.py

Timestamp:

Mar 19, 2014, 11:31:01 PM (11 years ago)

Author:

dmik

Message:

python: Merge vendor 2.7.6 to trunk.

Location:

python/trunk

Files:

• . (modified) (1 prop)
• Lib/test/test_telnetlib.py (modified) (5 diffs)

python/trunk/Lib/test/test_telnetlib.py

@@ -1 +1 @@
 import socket
-import threading
 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
-
-def server(evt, serv):
+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()
-        evt.set()
 
 class GeneralTests(TestCase):
@@ -25 +46 @@
         self.evt = threading.Event()
         self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
-        self.sock.settimeout(3)
+        self.sock.settimeout(60)  # Safety net. Look issue 11812
         self.port = test_support.bind_port(self.sock)
-        threading.Thread(target=server, args=(self.evt,self.sock)).start()
+        self.thread = threading.Thread(target=server, args=(self.evt,self.sock))
+        self.thread.setDaemon(True)
+        self.thread.start()
         self.evt.wait()
-        self.evt.clear()
-        time.sleep(.1)
 
     def tearDown(self):
-        self.evt.wait()
+        self.thread.join()
 
     def testBasic(self):
@@ -44 +65 @@
         socket.setdefaulttimeout(30)
         try:
-            telnet = telnetlib.Telnet("localhost", self.port)
+            telnet = telnetlib.Telnet(HOST, self.port)
         finally:
             socket.setdefaulttimeout(None)
@@ -62 +83 @@
 
     def testTimeoutValue(self):
-        telnet = telnetlib.Telnet("localhost", self.port, timeout=30)
+        telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
         self.assertEqual(telnet.sock.gettimeout(), 30)
         telnet.sock.close()
@@ -68 +89 @@
     def testTimeoutOpen(self):
         telnet = telnetlib.Telnet()
-        telnet.open("localhost", self.port, timeout=30)
+        telnet.open(HOST, self.port, timeout=30)
         self.assertEqual(telnet.sock.gettimeout(), 30)
         telnet.sock.close()
 
+    def testGetters(self):
+        # Test telnet getter methods
+        telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
+        t_sock = telnet.sock
+        self.assertEqual(telnet.get_socket(), t_sock)
+        self.assertEqual(telnet.fileno(), t_sock.fileno())
+        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)
+    test_support.run_unittest(GeneralTests, ReadTests, OptionTests,
+                              ExpectTests)
 
 if __name__ == '__main__':