blob: d57fda133747c3a76fa7bd4b8886fa0da7e18403 [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 threading
4import telnetlib
5import time
Jack Diederich1766b9d2009-11-06 17:20:42 +00006import contextlib
Guido van Rossumd8faa362007-04-27 19:54:29 +00007
8from unittest import TestCase
Benjamin Petersonee8712c2008-05-20 21:35:26 +00009from test import support
Guido van Rossumd8faa362007-04-27 19:54:29 +000010
Benjamin Petersonee8712c2008-05-20 21:35:26 +000011HOST = support.HOST
Guido van Rossumd8faa362007-04-27 19:54:29 +000012
Jack Diederich1766b9d2009-11-06 17:20:42 +000013def server(evt, serv):
Guido van Rossumd8faa362007-04-27 19:54:29 +000014 serv.listen(5)
Christian Heimesaf98da12008-01-27 15:18:18 +000015 evt.set()
Guido van Rossumd8faa362007-04-27 19:54:29 +000016 try:
17 conn, addr = serv.accept()
18 except socket.timeout:
19 pass
20 finally:
21 serv.close()
22 evt.set()
23
24class GeneralTests(TestCase):
25
26 def setUp(self):
27 self.evt = threading.Event()
Christian Heimes5e696852008-04-09 08:37:03 +000028 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
29 self.sock.settimeout(3)
Benjamin Petersonee8712c2008-05-20 21:35:26 +000030 self.port = support.bind_port(self.sock)
Jack Diederich1c8f38c2009-04-10 05:33:26 +000031 self.thread = threading.Thread(target=server, args=(self.evt,self.sock))
32 self.thread.start()
Christian Heimesaf98da12008-01-27 15:18:18 +000033 self.evt.wait()
34 self.evt.clear()
35 time.sleep(.1)
Guido van Rossumd8faa362007-04-27 19:54:29 +000036
37 def tearDown(self):
38 self.evt.wait()
Jack Diederich1c8f38c2009-04-10 05:33:26 +000039 self.thread.join()
Guido van Rossumd8faa362007-04-27 19:54:29 +000040
41 def testBasic(self):
42 # connects
Christian Heimes5e696852008-04-09 08:37:03 +000043 telnet = telnetlib.Telnet(HOST, self.port)
Guido van Rossumd8faa362007-04-27 19:54:29 +000044 telnet.sock.close()
45
46 def testTimeoutDefault(self):
Georg Brandlf78e02b2008-06-10 17:40:04 +000047 self.assertTrue(socket.getdefaulttimeout() is None)
48 socket.setdefaulttimeout(30)
49 try:
50 telnet = telnetlib.Telnet("localhost", self.port)
51 finally:
52 socket.setdefaulttimeout(None)
Guido van Rossumd8faa362007-04-27 19:54:29 +000053 self.assertEqual(telnet.sock.gettimeout(), 30)
54 telnet.sock.close()
55
56 def testTimeoutNone(self):
57 # None, having other default
Georg Brandlf78e02b2008-06-10 17:40:04 +000058 self.assertTrue(socket.getdefaulttimeout() is None)
Guido van Rossumd8faa362007-04-27 19:54:29 +000059 socket.setdefaulttimeout(30)
60 try:
Christian Heimes5e696852008-04-09 08:37:03 +000061 telnet = telnetlib.Telnet(HOST, self.port, timeout=None)
Guido van Rossumd8faa362007-04-27 19:54:29 +000062 finally:
Georg Brandlf78e02b2008-06-10 17:40:04 +000063 socket.setdefaulttimeout(None)
64 self.assertTrue(telnet.sock.gettimeout() is None)
65 telnet.sock.close()
66
67 def testTimeoutValue(self):
68 telnet = telnetlib.Telnet("localhost", self.port, timeout=30)
Guido van Rossumd8faa362007-04-27 19:54:29 +000069 self.assertEqual(telnet.sock.gettimeout(), 30)
70 telnet.sock.close()
71
Georg Brandlf78e02b2008-06-10 17:40:04 +000072 def testTimeoutOpen(self):
73 telnet = telnetlib.Telnet()
74 telnet.open("localhost", self.port, timeout=30)
75 self.assertEqual(telnet.sock.gettimeout(), 30)
76 telnet.sock.close()
Guido van Rossumd8faa362007-04-27 19:54:29 +000077
Jack Diederich1766b9d2009-11-06 17:20:42 +000078class SocketStub(object):
79 ''' a socket proxy that re-defines sendall() '''
80 def __init__(self, reads=[]):
81 self.reads = reads
82 self.writes = []
83 self.block = False
84 def sendall(self, data):
85 self.writes.append(data)
86 def recv(self, size):
87 out = b''
88 while self.reads and len(out) < size:
89 out += self.reads.pop(0)
90 if len(out) > size:
91 self.reads.insert(0, out[size:])
92 out = out[:size]
93 return out
Jack Diederich1c8f38c2009-04-10 05:33:26 +000094
Jack Diederich1766b9d2009-11-06 17:20:42 +000095class TelnetAlike(telnetlib.Telnet):
96 def fileno(self):
97 raise NotImplementedError()
98 def close(self): pass
99 def sock_avail(self):
100 return (not self.sock.block)
101 def msg(self, msg, *args):
102 with support.captured_stdout() as out:
103 telnetlib.Telnet.msg(self, msg, *args)
104 self._messages += out.getvalue()
105 return
106
107def new_select(*s_args):
108 block = False
109 for l in s_args:
110 for fob in l:
111 if isinstance(fob, TelnetAlike):
112 block = fob.sock.block
113 if block:
114 return [[], [], []]
115 else:
116 return s_args
117
118@contextlib.contextmanager
119def test_socket(reads):
120 def new_conn(*ignored):
121 return SocketStub(reads)
122 try:
123 old_conn = socket.create_connection
124 socket.create_connection = new_conn
125 yield None
126 finally:
127 socket.create_connection = old_conn
128 return
129
130def test_telnet(reads=[], cls=TelnetAlike):
131 ''' return a telnetlib.Telnet object that uses a SocketStub with
132 reads queued up to be read '''
133 for x in reads:
134 assert type(x) is bytes, x
135 with test_socket(reads):
136 telnet = cls('dummy', 0)
137 telnet._messages = '' # debuglevel output
138 return telnet
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000139
140class ReadTests(TestCase):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000141 def setUp(self):
142 self.old_select = select.select
143 select.select = new_select
144 def tearDown(self):
145 select.select = self.old_select
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000146
Jack Diederich1766b9d2009-11-06 17:20:42 +0000147 def test_read_until(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000148 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000149 read_until(expected, timeout=None)
150 test the blocking version of read_util
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000151 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000152 want = [b'xxxmatchyyy']
153 telnet = test_telnet(want)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000154 data = telnet.read_until(b'match')
Jack Diederich1766b9d2009-11-06 17:20:42 +0000155 self.assertEqual(data, b'xxxmatch', msg=(telnet.cookedq, telnet.rawq, telnet.sock.reads))
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000156
Jack Diederich1766b9d2009-11-06 17:20:42 +0000157 reads = [b'x' * 50, b'match', b'y' * 50]
158 expect = b''.join(reads[:-1])
159 telnet = test_telnet(reads)
160 data = telnet.read_until(b'match')
161 self.assertEqual(data, expect)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000162
Jack Diederich1766b9d2009-11-06 17:20:42 +0000163
164 def test_read_all(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000165 """
166 read_all()
167 Read all data until EOF; may block.
168 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000169 reads = [b'x' * 500, b'y' * 500, b'z' * 500]
170 expect = b''.join(reads)
171 telnet = test_telnet(reads)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000172 data = telnet.read_all()
Jack Diederich1766b9d2009-11-06 17:20:42 +0000173 self.assertEqual(data, expect)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000174 return
175
Jack Diederich1766b9d2009-11-06 17:20:42 +0000176 def test_read_some(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000177 """
178 read_some()
179 Read at least one byte or EOF; may block.
180 """
181 # test 'at least one byte'
Jack Diederich1766b9d2009-11-06 17:20:42 +0000182 telnet = test_telnet([b'x' * 500])
183 data = telnet.read_some()
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000184 self.assertTrue(len(data) >= 1)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000185 # test EOF
Jack Diederich1766b9d2009-11-06 17:20:42 +0000186 telnet = test_telnet()
187 data = telnet.read_some()
188 self.assertEqual(b'', data)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000189
Jack Diederich1766b9d2009-11-06 17:20:42 +0000190 def _read_eager(self, func_name):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000191 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000192 read_*_eager()
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000193 Read all data available already queued or on the socket,
194 without blocking.
195 """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000196 want = b'x' * 100
197 telnet = test_telnet([want])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000198 func = getattr(telnet, func_name)
Jack Diederich1766b9d2009-11-06 17:20:42 +0000199 telnet.sock.block = True
200 self.assertEqual(b'', func())
201 telnet.sock.block = False
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000202 data = b''
203 while True:
204 try:
205 data += func()
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000206 except EOFError:
207 break
Jack Diederich1766b9d2009-11-06 17:20:42 +0000208 self.assertEqual(data, want)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000209
Jack Diederich1766b9d2009-11-06 17:20:42 +0000210 def test_read_eager(self):
211 # read_eager and read_very_eager make the same gaurantees
212 # (they behave differently but we only test the gaurantees)
213 self._read_eager('read_eager')
214 self._read_eager('read_very_eager')
215 # NB -- we need to test the IAC block which is mentioned in the
216 # docstring but not in the module docs
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000217
Jack Diederich1766b9d2009-11-06 17:20:42 +0000218 def read_very_lazy(self):
219 want = b'x' * 100
220 telnet = test_telnet([want])
221 self.assertEqual(b'', telnet.read_very_lazy())
222 while telnet.sock.reads:
223 telnet.fill_rawq()
224 data = telnet.read_very_lazy()
225 self.assertEqual(want, data)
226 self.assertRaises(EOFError, telnet.read_very_lazy)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000227
Jack Diederich1766b9d2009-11-06 17:20:42 +0000228 def test_read_lazy(self):
229 want = b'x' * 100
230 telnet = test_telnet([want])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000231 self.assertEqual(b'', telnet.read_lazy())
232 data = b''
233 while True:
234 try:
235 read_data = telnet.read_lazy()
236 data += read_data
237 if not read_data:
238 telnet.fill_rawq()
239 except EOFError:
240 break
Jack Diederich1766b9d2009-11-06 17:20:42 +0000241 self.assertTrue(want.startswith(data))
242 self.assertEqual(data, want)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000243
244class nego_collector(object):
245 def __init__(self, sb_getter=None):
246 self.seen = b''
247 self.sb_getter = sb_getter
248 self.sb_seen = b''
249
250 def do_nego(self, sock, cmd, opt):
251 self.seen += cmd + opt
252 if cmd == tl.SE and self.sb_getter:
253 sb_data = self.sb_getter()
254 self.sb_seen += sb_data
255
Jack Diederich1766b9d2009-11-06 17:20:42 +0000256tl = telnetlib
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000257
258class WriteTests(TestCase):
259 '''The only thing that write does is replace each tl.IAC for
260 tl.IAC+tl.IAC'''
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000261
262 def test_write(self):
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000263 data_sample = [b'data sample without IAC',
264 b'data sample with' + tl.IAC + b' one IAC',
265 b'a few' + tl.IAC + tl.IAC + b' iacs' + tl.IAC,
266 tl.IAC,
267 b'']
Jack Diederich1766b9d2009-11-06 17:20:42 +0000268 for data in data_sample:
269 telnet = test_telnet()
270 telnet.write(data)
271 written = b''.join(telnet.sock.writes)
272 self.assertEqual(data.replace(tl.IAC,tl.IAC+tl.IAC), written)
Jack Diederich36596a32009-07-26 22:23:04 +0000273
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000274class OptionTests(TestCase):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000275 # RFC 854 commands
276 cmds = [tl.AO, tl.AYT, tl.BRK, tl.EC, tl.EL, tl.GA, tl.IP, tl.NOP]
277
278 def _test_command(self, data):
279 """ helper for testing IAC + cmd """
Jack Diederich1766b9d2009-11-06 17:20:42 +0000280 telnet = test_telnet(data)
281 data_len = len(b''.join(data))
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000282 nego = nego_collector()
283 telnet.set_option_negotiation_callback(nego.do_nego)
284 txt = telnet.read_all()
285 cmd = nego.seen
286 self.assertTrue(len(cmd) > 0) # we expect at least one command
287 self.assertTrue(cmd[:1] in self.cmds)
288 self.assertEqual(cmd[1:2], tl.NOOPT)
Jack Diederich1766b9d2009-11-06 17:20:42 +0000289 self.assertEqual(data_len, len(txt + cmd))
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000290 nego.sb_getter = None # break the nego => telnet cycle
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000291
292 def test_IAC_commands(self):
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000293 for cmd in self.cmds:
Jack Diederich1766b9d2009-11-06 17:20:42 +0000294 self._test_command([tl.IAC, cmd])
295 self._test_command([b'x' * 100, tl.IAC, cmd, b'y'*100])
296 self._test_command([b'x' * 10, tl.IAC, cmd, b'y'*10])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000297 # all at once
Jack Diederich1766b9d2009-11-06 17:20:42 +0000298 self._test_command([tl.IAC + cmd for (cmd) in self.cmds])
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000299
300 def test_SB_commands(self):
301 # RFC 855, subnegotiations portion
302 send = [tl.IAC + tl.SB + tl.IAC + tl.SE,
303 tl.IAC + tl.SB + tl.IAC + tl.IAC + tl.IAC + tl.SE,
304 tl.IAC + tl.SB + tl.IAC + tl.IAC + b'aa' + tl.IAC + tl.SE,
305 tl.IAC + tl.SB + b'bb' + tl.IAC + tl.IAC + tl.IAC + tl.SE,
306 tl.IAC + tl.SB + b'cc' + tl.IAC + tl.IAC + b'dd' + tl.IAC + tl.SE,
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000307 ]
Jack Diederich1766b9d2009-11-06 17:20:42 +0000308 telnet = test_telnet(send)
Jack Diederich1c8f38c2009-04-10 05:33:26 +0000309 nego = nego_collector(telnet.read_sb_data)
310 telnet.set_option_negotiation_callback(nego.do_nego)
311 txt = telnet.read_all()
312 self.assertEqual(txt, b'')
313 want_sb_data = tl.IAC + tl.IAC + b'aabb' + tl.IAC + b'cc' + tl.IAC + b'dd'
314 self.assertEqual(nego.sb_seen, want_sb_data)
315 self.assertEqual(b'', telnet.read_sb_data())
316 nego.sb_getter = None # break the nego => telnet cycle
Guido van Rossumd8faa362007-04-27 19:54:29 +0000317
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000318 def test_debuglevel_reads(self):
Jack Diederich36596a32009-07-26 22:23:04 +0000319 # test all the various places that self.msg(...) is called
320 given_a_expect_b = [
321 # Telnet.fill_rawq
322 (b'a', ": recv b''\n"),
323 # Telnet.process_rawq
324 (tl.IAC + bytes([88]), ": IAC 88 not recognized\n"),
325 (tl.IAC + tl.DO + bytes([1]), ": IAC DO 1\n"),
326 (tl.IAC + tl.DONT + bytes([1]), ": IAC DONT 1\n"),
327 (tl.IAC + tl.WILL + bytes([1]), ": IAC WILL 1\n"),
328 (tl.IAC + tl.WONT + bytes([1]), ": IAC WONT 1\n"),
Jack Diederich36596a32009-07-26 22:23:04 +0000329 ]
330 for a, b in given_a_expect_b:
Jack Diederich1766b9d2009-11-06 17:20:42 +0000331 telnet = test_telnet([a])
332 telnet.set_debuglevel(1)
333 txt = telnet.read_all()
334 self.assertTrue(b in telnet._messages)
Jack Diederich36596a32009-07-26 22:23:04 +0000335 return
336
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000337 def test_debuglevel_write(self):
Jack Diederich1766b9d2009-11-06 17:20:42 +0000338 telnet = test_telnet()
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000339 telnet.set_debuglevel(1)
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000340 telnet.write(b'xxx')
341 expected = "send b'xxx'\n"
Jack Diederich1766b9d2009-11-06 17:20:42 +0000342 self.assertTrue(expected in telnet._messages)
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000343
Guido van Rossumd8faa362007-04-27 19:54:29 +0000344def test_main(verbose=None):
Jack Diederichf31f7bb2009-09-03 20:37:58 +0000345 support.run_unittest(GeneralTests, ReadTests, WriteTests, OptionTests)
Guido van Rossumd8faa362007-04-27 19:54:29 +0000346
347if __name__ == '__main__':
348 test_main()