mirror of
https://github.com/python/cpython.git
synced 2025-09-23 08:53:45 +00:00
e0c22206e4
to avoid failing due to the select() file descriptor limit. Contributed by Akintayo Holder and under the Google contributor agreement.
453 lines
15 KiB
Python
453 lines
15 KiB
Python
import socket
|
|
import telnetlib
|
|
import time
|
|
import Queue
|
|
|
|
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()
|