blob: f8f854c110f39fb6073fb0417e6644555c2fd6ed [file] [log] [blame]
Guido van Rossumd8faa362007-04-27 19:54:29 +00001import socket
Jack Diederich1766b9d2009-11-06 17:20:42 +00002import select
Guido van Rossumd8faa362007-04-27 19:54:29 +00003import telnetlib
4import time
Jack Diederich1766b9d2009-11-06 17:20:42 +00005import contextlib
Guido van Rossumd8faa362007-04-27 19:54:29 +00006
Gregory P. Smith5bcd0052012-07-16 13:34:50 -07007import unittest
Guido van Rossumd8faa362007-04-27 19:54:29 +00008from unittest import TestCase
Benjamin Petersonee8712c2008-05-20 21:35:26 +00009from test import support
Victor Stinner45df8202010-04-28 22:31:17 +000010threading = support.import_module('threading')
Guido van Rossumd8faa362007-04-27 19:54:29 +000011
Benjamin Petersonee8712c2008-05-20 21:35:26 +000012HOST = support.HOST
Guido van Rossumd8faa362007-04-27 19:54:29 +000013
Jack Diederich1766b9d2009-11-06 17:20:42 +000014def server(evt, serv):
Guido van Rossumd8faa362007-04-27 19:54:29 +000015 serv.listen(5)
Christian Heimesaf98da12008-01-27 15:18:18 +000016 evt.set()
Guido van Rossumd8faa362007-04-27 19:54:29 +000017 try:
18 conn, addr = serv.accept()
Jesus Ceabc91b462011-11-08 16:24:43 +010019 conn.close()
Guido van Rossumd8faa362007-04-27 19:54:29 +000020 except socket.timeout:
21 pass
22 finally:
23 serv.close()
Guido van Rossumd8faa362007-04-27 19:54:29 +000024
25class GeneralTests(TestCase):
26
27 def setUp(self):
28 self.evt = threading.Event()
Christian Heimes5e696852008-04-09 08:37:03 +000029 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
Jesus Ceabc91b462011-11-08 16:24:43 +010030 self.sock.settimeout(60) # Safety net. Look issue 11812
Benjamin Petersonee8712c2008-05-20 21:35:26 +000031 self.port = support.bind_port(self.sock)
Jack Diederich1c8f38c2009-04-10 05:33:26 +000032 self.thread = threading.Thread(target=server, args=(self.evt,self.sock))
Jesus Ceabc91b462011-11-08 16:24:43 +010033 self.thread.setDaemon(True)
Jack Diederich1c8f38c2009-04-10 05:33:26 +000034 self.thread.start()
Christian Heimesaf98da12008-01-27 15:18:18 +000035 self.evt.wait()
Guido van Rossumd8faa362007-04-27 19:54:29 +000036
37 def tearDown(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +000038 self.thread.join()
Guido van Rossumd8faa362007-04-27 19:54:29 +000039
40 def testBasic(self):
41 # connects
Christian Heimes5e696852008-04-09 08:37:03 +000042 telnet = telnetlib.Telnet(HOST, self.port)
Guido van Rossumd8faa362007-04-27 19:54:29 +000043 telnet.sock.close()
44
45 def testTimeoutDefault(self):
Georg Brandlf78e02b2008-06-10 17:40:04 +000046 self.assertTrue(socket.getdefaulttimeout() is None)
47 socket.setdefaulttimeout(30)
48 try:
Jesus Ceabc91b462011-11-08 16:24:43 +010049 telnet = telnetlib.Telnet(HOST, self.port)
Georg Brandlf78e02b2008-06-10 17:40:04 +000050 finally:
51 socket.setdefaulttimeout(None)
Guido van Rossumd8faa362007-04-27 19:54:29 +000052 self.assertEqual(telnet.sock.gettimeout(), 30)
53 telnet.sock.close()
54
55 def testTimeoutNone(self):
56 # None, having other default
Georg Brandlf78e02b2008-06-10 17:40:04 +000057 self.assertTrue(socket.getdefaulttimeout() is None)
Guido van Rossumd8faa362007-04-27 19:54:29 +000058 socket.setdefaulttimeout(30)
59 try:
Christian Heimes5e696852008-04-09 08:37:03 +000060 telnet = telnetlib.Telnet(HOST, self.port, timeout=None)
Guido van Rossumd8faa362007-04-27 19:54:29 +000061 finally:
Georg Brandlf78e02b2008-06-10 17:40:04 +000062 socket.setdefaulttimeout(None)
63 self.assertTrue(telnet.sock.gettimeout() is None)
64 telnet.sock.close()
65
66 def testTimeoutValue(self):
Jesus Ceabc91b462011-11-08 16:24:43 +010067 telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
Guido van Rossumd8faa362007-04-27 19:54:29 +000068 self.assertEqual(telnet.sock.gettimeout(), 30)
69 telnet.sock.close()
70
Georg Brandlf78e02b2008-06-10 17:40:04 +000071 def testTimeoutOpen(self):
72 telnet = telnetlib.Telnet()
Jesus Ceabc91b462011-11-08 16:24:43 +010073 telnet.open(HOST, self.port, timeout=30)
Georg Brandlf78e02b2008-06-10 17:40:04 +000074 self.assertEqual(telnet.sock.gettimeout(), 30)
75 telnet.sock.close()
Guido van Rossumd8faa362007-04-27 19:54:29 +000076
Jack Diederich1766b9d2009-11-06 17:20:42 +000077class SocketStub(object):
78 ''' a socket proxy that re-defines sendall() '''
Gregory P. Smithdad57112012-07-15 23:42:26 -070079 def __init__(self, reads=()):
80 self.reads = list(reads) # Intentionally make a copy.
Jack Diederich1766b9d2009-11-06 17:20:42 +000081 self.writes = []
82 self.block = False
83 def sendall(self, data):
84 self.writes.append(data)
85 def recv(self, size):
86 out = b''
87 while self.reads and len(out) < size:
88 out += self.reads.pop(0)
89 if len(out) > size:
90 self.reads.insert(0, out[size:])
91 out = out[:size]
92 return out
Jack Diederich1c8f38c2009-04-10 05:33:26 +000093
Jack Diederich1766b9d2009-11-06 17:20:42 +000094class TelnetAlike(telnetlib.Telnet):
95 def fileno(self):
96 raise NotImplementedError()
97 def close(self): pass
98 def sock_avail(self):
99 return (not self.sock.block)
100 def msg(self, msg, *args):
101 with support.captured_stdout() as out:
102 telnetlib.Telnet.msg(self, msg, *args)
103 self._messages += out.getvalue()
104 return
105
Gregory P. Smithdad57112012-07-15 23:42:26 -0700106def mock_select(*s_args):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000107 block = False
108 for l in s_args:
109 for fob in l:
110 if isinstance(fob, TelnetAlike):
111 block = fob.sock.block
112 if block:
113 return [[], [], []]
114 else:
115 return s_args
116
Gregory P. Smithdad57112012-07-15 23:42:26 -0700117class MockPoller(object):
118 test_case = None # Set during TestCase setUp.
119
120 def __init__(self):
121 self._file_objs = []
122
123 def register(self, fd, eventmask):
124 self.test_case.assertTrue(hasattr(fd, 'fileno'), fd)
125 self.test_case.assertEqual(eventmask, select.POLLIN|select.POLLPRI)
126 self._file_objs.append(fd)
127
128 def poll(self, timeout=None):
129 block = False
130 for fob in self._file_objs:
131 if isinstance(fob, TelnetAlike):
132 block = fob.sock.block
133 if block:
134 return []
135 else:
136 return zip(self._file_objs, [select.POLLIN]*len(self._file_objs))
137
138 def unregister(self, fd):
139 self._file_objs.remove(fd)
140
Jack Diederich1766b9d2009-11-06 17:20:42 +0000141@contextlib.contextmanager
142def test_socket(reads):
143 def new_conn(*ignored):
144 return SocketStub(reads)
145 try:
146 old_conn = socket.create_connection
147 socket.create_connection = new_conn
148 yield None
149 finally:
150 socket.create_connection = old_conn
151 return
152
Gregory P. Smithdad57112012-07-15 23:42:26 -0700153def test_telnet(reads=(), cls=TelnetAlike, use_poll=None):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000154 ''' return a telnetlib.Telnet object that uses a SocketStub with
155 reads queued up to be read '''
156 for x in reads:
157 assert type(x) is bytes, x
158 with test_socket(reads):
159 telnet = cls('dummy', 0)
160 telnet._messages = '' # debuglevel output
Gregory P. Smithdad57112012-07-15 23:42:26 -0700161 if use_poll is not None:
162 if use_poll and not telnet._has_poll:
163 raise unittest.SkipTest('select.poll() required.')
164 telnet._has_poll = use_poll
Jack Diederich1766b9d2009-11-06 17:20:42 +0000165 return telnet
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000166
Gregory P. Smithdad57112012-07-15 23:42:26 -0700167
168class ExpectAndReadTestCase(TestCase):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000169 def setUp(self):
170 self.old_select = select.select
Gregory P. Smithdad57112012-07-15 23:42:26 -0700171 select.select = mock_select
Gregory P. Smithac14aa52012-07-16 13:38:45 -0700172 if hasattr(select, 'poll'):
173 self.old_poll = select.poll
174 select.poll = MockPoller
175 MockPoller.test_case = self
Gregory P. Smithdad57112012-07-15 23:42:26 -0700176
Jack Diederich1766b9d2009-11-06 17:20:42 +0000177 def tearDown(self):
Gregory P. Smithac14aa52012-07-16 13:38:45 -0700178 if hasattr(select, 'poll'):
179 MockPoller.test_case = None
180 select.poll = self.old_poll
Jack Diederich1766b9d2009-11-06 17:20:42 +0000181 select.select = self.old_select
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000182
Gregory P. Smithdad57112012-07-15 23:42:26 -0700183
184class ReadTests(ExpectAndReadTestCase):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000185 def test_read_until(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000186 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000187 read_until(expected, timeout=None)
188 test the blocking version of read_util
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000189 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000190 want = [b'xxxmatchyyy']
191 telnet = test_telnet(want)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000192 data = telnet.read_until(b'match')
Jack Diederich1766b9d2009-11-06 17:20:42 +0000193 self.assertEqual(data, b'xxxmatch', msg=(telnet.cookedq, telnet.rawq, telnet.sock.reads))
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000194
Jack Diederich1766b9d2009-11-06 17:20:42 +0000195 reads = [b'x' * 50, b'match', b'y' * 50]
196 expect = b''.join(reads[:-1])
197 telnet = test_telnet(reads)
198 data = telnet.read_until(b'match')
199 self.assertEqual(data, expect)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000200
Gregory P. Smithdad57112012-07-15 23:42:26 -0700201 def test_read_until_with_poll(self):
202 """Use select.poll() to implement telnet.read_until()."""
203 want = [b'x' * 10, b'match', b'y' * 10]
204 telnet = test_telnet(want, use_poll=True)
205 select.select = lambda *_: self.fail('unexpected select() call.')
206 data = telnet.read_until(b'match')
207 self.assertEqual(data, b''.join(want[:-1]))
208
209 def test_read_until_with_select(self):
210 """Use select.select() to implement telnet.read_until()."""
211 want = [b'x' * 10, b'match', b'y' * 10]
212 telnet = test_telnet(want, use_poll=False)
213 select.poll = lambda *_: self.fail('unexpected poll() call.')
214 data = telnet.read_until(b'match')
215 self.assertEqual(data, b''.join(want[:-1]))
Jack Diederich1766b9d2009-11-06 17:20:42 +0000216
217 def test_read_all(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000218 """
219 read_all()
220 Read all data until EOF; may block.
221 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000222 reads = [b'x' * 500, b'y' * 500, b'z' * 500]
223 expect = b''.join(reads)
224 telnet = test_telnet(reads)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000225 data = telnet.read_all()
Jack Diederich1766b9d2009-11-06 17:20:42 +0000226 self.assertEqual(data, expect)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000227 return
228
Jack Diederich1766b9d2009-11-06 17:20:42 +0000229 def test_read_some(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000230 """
231 read_some()
232 Read at least one byte or EOF; may block.
233 """
234 # test 'at least one byte'
Jack Diederich1766b9d2009-11-06 17:20:42 +0000235 telnet = test_telnet([b'x' * 500])
236 data = telnet.read_some()
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000237 self.assertTrue(len(data) >= 1)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000238 # test EOF
Jack Diederich1766b9d2009-11-06 17:20:42 +0000239 telnet = test_telnet()
240 data = telnet.read_some()
241 self.assertEqual(b'', data)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000242
Jack Diederich1766b9d2009-11-06 17:20:42 +0000243 def _read_eager(self, func_name):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000244 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000245 read_*_eager()
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000246 Read all data available already queued or on the socket,
247 without blocking.
248 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000249 want = b'x' * 100
250 telnet = test_telnet([want])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000251 func = getattr(telnet, func_name)
Jack Diederich1766b9d2009-11-06 17:20:42 +0000252 telnet.sock.block = True
253 self.assertEqual(b'', func())
254 telnet.sock.block = False
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000255 data = b''
256 while True:
257 try:
258 data += func()
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000259 except EOFError:
260 break
Jack Diederich1766b9d2009-11-06 17:20:42 +0000261 self.assertEqual(data, want)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000262
Jack Diederich1766b9d2009-11-06 17:20:42 +0000263 def test_read_eager(self):
264 # read_eager and read_very_eager make the same gaurantees
265 # (they behave differently but we only test the gaurantees)
266 self._read_eager('read_eager')
267 self._read_eager('read_very_eager')
268 # NB -- we need to test the IAC block which is mentioned in the
269 # docstring but not in the module docs
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000270
Jack Diederich1766b9d2009-11-06 17:20:42 +0000271 def read_very_lazy(self):
272 want = b'x' * 100
273 telnet = test_telnet([want])
274 self.assertEqual(b'', telnet.read_very_lazy())
275 while telnet.sock.reads:
276 telnet.fill_rawq()
277 data = telnet.read_very_lazy()
278 self.assertEqual(want, data)
279 self.assertRaises(EOFError, telnet.read_very_lazy)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000280
Jack Diederich1766b9d2009-11-06 17:20:42 +0000281 def test_read_lazy(self):
282 want = b'x' * 100
283 telnet = test_telnet([want])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000284 self.assertEqual(b'', telnet.read_lazy())
285 data = b''
286 while True:
287 try:
288 read_data = telnet.read_lazy()
289 data += read_data
290 if not read_data:
291 telnet.fill_rawq()
292 except EOFError:
293 break
Jack Diederich1766b9d2009-11-06 17:20:42 +0000294 self.assertTrue(want.startswith(data))
295 self.assertEqual(data, want)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000296
297class nego_collector(object):
298 def __init__(self, sb_getter=None):
299 self.seen = b''
300 self.sb_getter = sb_getter
301 self.sb_seen = b''
302
303 def do_nego(self, sock, cmd, opt):
304 self.seen += cmd + opt
305 if cmd == tl.SE and self.sb_getter:
306 sb_data = self.sb_getter()
307 self.sb_seen += sb_data
308
Jack Diederich1766b9d2009-11-06 17:20:42 +0000309tl = telnetlib
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000310
311class WriteTests(TestCase):
312 '''The only thing that write does is replace each tl.IAC for
313 tl.IAC+tl.IAC'''
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000314
315 def test_write(self):
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000316 data_sample = [b'data sample without IAC',
317 b'data sample with' + tl.IAC + b' one IAC',
318 b'a few' + tl.IAC + tl.IAC + b' iacs' + tl.IAC,
319 tl.IAC,
320 b'']
Jack Diederich1766b9d2009-11-06 17:20:42 +0000321 for data in data_sample:
322 telnet = test_telnet()
323 telnet.write(data)
324 written = b''.join(telnet.sock.writes)
325 self.assertEqual(data.replace(tl.IAC,tl.IAC+tl.IAC), written)
Jack Diederich36596a32009-07-26 22:23:04 +0000326
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000327class OptionTests(TestCase):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000328 # RFC 854 commands
329 cmds = [tl.AO, tl.AYT, tl.BRK, tl.EC, tl.EL, tl.GA, tl.IP, tl.NOP]
330
331 def _test_command(self, data):
332 """ helper for testing IAC + cmd """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000333 telnet = test_telnet(data)
334 data_len = len(b''.join(data))
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000335 nego = nego_collector()
336 telnet.set_option_negotiation_callback(nego.do_nego)
337 txt = telnet.read_all()
338 cmd = nego.seen
339 self.assertTrue(len(cmd) > 0) # we expect at least one command
Benjamin Peterson577473f2010-01-19 00:09:57 +0000340 self.assertIn(cmd[:1], self.cmds)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000341 self.assertEqual(cmd[1:2], tl.NOOPT)
Jack Diederich1766b9d2009-11-06 17:20:42 +0000342 self.assertEqual(data_len, len(txt + cmd))
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000343 nego.sb_getter = None # break the nego => telnet cycle
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000344
345 def test_IAC_commands(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000346 for cmd in self.cmds:
Jack Diederich1766b9d2009-11-06 17:20:42 +0000347 self._test_command([tl.IAC, cmd])
348 self._test_command([b'x' * 100, tl.IAC, cmd, b'y'*100])
349 self._test_command([b'x' * 10, tl.IAC, cmd, b'y'*10])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000350 # all at once
Jack Diederich1766b9d2009-11-06 17:20:42 +0000351 self._test_command([tl.IAC + cmd for (cmd) in self.cmds])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000352
353 def test_SB_commands(self):
354 # RFC 855, subnegotiations portion
355 send = [tl.IAC + tl.SB + tl.IAC + tl.SE,
356 tl.IAC + tl.SB + tl.IAC + tl.IAC + tl.IAC + tl.SE,
357 tl.IAC + tl.SB + tl.IAC + tl.IAC + b'aa' + tl.IAC + tl.SE,
358 tl.IAC + tl.SB + b'bb' + tl.IAC + tl.IAC + tl.IAC + tl.SE,
359 tl.IAC + tl.SB + b'cc' + tl.IAC + tl.IAC + b'dd' + tl.IAC + tl.SE,
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000360 ]
Jack Diederich1766b9d2009-11-06 17:20:42 +0000361 telnet = test_telnet(send)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000362 nego = nego_collector(telnet.read_sb_data)
363 telnet.set_option_negotiation_callback(nego.do_nego)
364 txt = telnet.read_all()
365 self.assertEqual(txt, b'')
366 want_sb_data = tl.IAC + tl.IAC + b'aabb' + tl.IAC + b'cc' + tl.IAC + b'dd'
367 self.assertEqual(nego.sb_seen, want_sb_data)
368 self.assertEqual(b'', telnet.read_sb_data())
369 nego.sb_getter = None # break the nego => telnet cycle
Guido van Rossumd8faa362007-04-27 19:54:29 +0000370
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000371 def test_debuglevel_reads(self):
Jack Diederich36596a32009-07-26 22:23:04 +0000372 # test all the various places that self.msg(...) is called
373 given_a_expect_b = [
374 # Telnet.fill_rawq
375 (b'a', ": recv b''\n"),
376 # Telnet.process_rawq
377 (tl.IAC + bytes([88]), ": IAC 88 not recognized\n"),
378 (tl.IAC + tl.DO + bytes([1]), ": IAC DO 1\n"),
379 (tl.IAC + tl.DONT + bytes([1]), ": IAC DONT 1\n"),
380 (tl.IAC + tl.WILL + bytes([1]), ": IAC WILL 1\n"),
381 (tl.IAC + tl.WONT + bytes([1]), ": IAC WONT 1\n"),
Jack Diederich36596a32009-07-26 22:23:04 +0000382 ]
383 for a, b in given_a_expect_b:
Jack Diederich1766b9d2009-11-06 17:20:42 +0000384 telnet = test_telnet([a])
385 telnet.set_debuglevel(1)
386 txt = telnet.read_all()
Benjamin Peterson577473f2010-01-19 00:09:57 +0000387 self.assertIn(b, telnet._messages)
Jack Diederich36596a32009-07-26 22:23:04 +0000388 return
389
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000390 def test_debuglevel_write(self):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000391 telnet = test_telnet()
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000392 telnet.set_debuglevel(1)
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000393 telnet.write(b'xxx')
394 expected = "send b'xxx'\n"
Benjamin Peterson577473f2010-01-19 00:09:57 +0000395 self.assertIn(expected, telnet._messages)
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000396
R. David Murray32ef70c2010-12-14 14:16:20 +0000397 def test_debug_accepts_str_port(self):
398 # Issue 10695
399 with test_socket([]):
400 telnet = TelnetAlike('dummy', '0')
401 telnet._messages = ''
402 telnet.set_debuglevel(1)
403 telnet.msg('test')
404 self.assertRegex(telnet._messages, r'0.*test')
405
406
Gregory P. Smithdad57112012-07-15 23:42:26 -0700407class ExpectTests(ExpectAndReadTestCase):
408 def test_expect(self):
409 """
410 expect(expected, [timeout])
411 Read until the expected string has been seen, or a timeout is
412 hit (default is no timeout); may block.
413 """
414 want = [b'x' * 10, b'match', b'y' * 10]
415 telnet = test_telnet(want)
416 (_,_,data) = telnet.expect([b'match'])
417 self.assertEqual(data, b''.join(want[:-1]))
418
419 def test_expect_with_poll(self):
420 """Use select.poll() to implement telnet.expect()."""
421 want = [b'x' * 10, b'match', b'y' * 10]
422 telnet = test_telnet(want, use_poll=True)
423 select.select = lambda *_: self.fail('unexpected select() call.')
424 (_,_,data) = telnet.expect([b'match'])
425 self.assertEqual(data, b''.join(want[:-1]))
426
427 def test_expect_with_select(self):
428 """Use select.select() to implement telnet.expect()."""
429 want = [b'x' * 10, b'match', b'y' * 10]
430 telnet = test_telnet(want, use_poll=False)
431 select.poll = lambda *_: self.fail('unexpected poll() call.')
432 (_,_,data) = telnet.expect([b'match'])
433 self.assertEqual(data, b''.join(want[:-1]))
434
435
Guido van Rossumd8faa362007-04-27 19:54:29 +0000436def test_main(verbose=None):
Gregory P. Smithdad57112012-07-15 23:42:26 -0700437 support.run_unittest(GeneralTests, ReadTests, WriteTests, OptionTests,
438 ExpectTests)
Guido van Rossumd8faa362007-04-27 19:54:29 +0000439
440if __name__ == '__main__':
441 test_main()