Python code coverage for Lib/test/test_telnetlib.py

#	count	content
1	n/a	import socket
2	n/a	import selectors
3	n/a	import telnetlib
4	n/a	import contextlib
5	n/a
6	n/a	from test import support
7	n/a	import unittest
8	n/a	threading = support.import_module('threading')
9	n/a
10	n/a	HOST = support.HOST
11	n/a
12	n/a	def server(evt, serv):
13	n/a	serv.listen()
14	n/a	evt.set()
15	n/a	try:
16	n/a	conn, addr = serv.accept()
17	n/a	conn.close()
18	n/a	except socket.timeout:
19	n/a	pass
20	n/a	finally:
21	n/a	serv.close()
22	n/a
23	n/a	class GeneralTests(unittest.TestCase):
24	n/a
25	n/a	def setUp(self):
26	n/a	self.evt = threading.Event()
27	n/a	self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
28	n/a	self.sock.settimeout(60) # Safety net. Look issue 11812
29	n/a	self.port = support.bind_port(self.sock)
30	n/a	self.thread = threading.Thread(target=server, args=(self.evt,self.sock))
31	n/a	self.thread.setDaemon(True)
32	n/a	self.thread.start()
33	n/a	self.evt.wait()
34	n/a
35	n/a	def tearDown(self):
36	n/a	self.thread.join()
37	n/a	del self.thread # Clear out any dangling Thread objects.
38	n/a
39	n/a	def testBasic(self):
40	n/a	# connects
41	n/a	telnet = telnetlib.Telnet(HOST, self.port)
42	n/a	telnet.sock.close()
43	n/a
44	n/a	def testContextManager(self):
45	n/a	with telnetlib.Telnet(HOST, self.port) as tn:
46	n/a	self.assertIsNotNone(tn.get_socket())
47	n/a	self.assertIsNone(tn.get_socket())
48	n/a
49	n/a	def testTimeoutDefault(self):
50	n/a	self.assertTrue(socket.getdefaulttimeout() is None)
51	n/a	socket.setdefaulttimeout(30)
52	n/a	try:
53	n/a	telnet = telnetlib.Telnet(HOST, self.port)
54	n/a	finally:
55	n/a	socket.setdefaulttimeout(None)
56	n/a	self.assertEqual(telnet.sock.gettimeout(), 30)
57	n/a	telnet.sock.close()
58	n/a
59	n/a	def testTimeoutNone(self):
60	n/a	# None, having other default
61	n/a	self.assertTrue(socket.getdefaulttimeout() is None)
62	n/a	socket.setdefaulttimeout(30)
63	n/a	try:
64	n/a	telnet = telnetlib.Telnet(HOST, self.port, timeout=None)
65	n/a	finally:
66	n/a	socket.setdefaulttimeout(None)
67	n/a	self.assertTrue(telnet.sock.gettimeout() is None)
68	n/a	telnet.sock.close()
69	n/a
70	n/a	def testTimeoutValue(self):
71	n/a	telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
72	n/a	self.assertEqual(telnet.sock.gettimeout(), 30)
73	n/a	telnet.sock.close()
74	n/a
75	n/a	def testTimeoutOpen(self):
76	n/a	telnet = telnetlib.Telnet()
77	n/a	telnet.open(HOST, self.port, timeout=30)
78	n/a	self.assertEqual(telnet.sock.gettimeout(), 30)
79	n/a	telnet.sock.close()
80	n/a
81	n/a	def testGetters(self):
82	n/a	# Test telnet getter methods
83	n/a	telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
84	n/a	t_sock = telnet.sock
85	n/a	self.assertEqual(telnet.get_socket(), t_sock)
86	n/a	self.assertEqual(telnet.fileno(), t_sock.fileno())
87	n/a	telnet.sock.close()
88	n/a
89	n/a	class SocketStub(object):
90	n/a	''' a socket proxy that re-defines sendall() '''
91	n/a	def __init__(self, reads=()):
92	n/a	self.reads = list(reads) # Intentionally make a copy.
93	n/a	self.writes = []
94	n/a	self.block = False
95	n/a	def sendall(self, data):
96	n/a	self.writes.append(data)
97	n/a	def recv(self, size):
98	n/a	out = b''
99	n/a	while self.reads and len(out) < size:
100	n/a	out += self.reads.pop(0)
101	n/a	if len(out) > size:
102	n/a	self.reads.insert(0, out[size:])
103	n/a	out = out[:size]
104	n/a	return out
105	n/a
106	n/a	class TelnetAlike(telnetlib.Telnet):
107	n/a	def fileno(self):
108	n/a	raise NotImplementedError()
109	n/a	def close(self): pass
110	n/a	def sock_avail(self):
111	n/a	return (not self.sock.block)
112	n/a	def msg(self, msg, *args):
113	n/a	with support.captured_stdout() as out:
114	n/a	telnetlib.Telnet.msg(self, msg, *args)
115	n/a	self._messages += out.getvalue()
116	n/a	return
117	n/a
118	n/a	class MockSelector(selectors.BaseSelector):
119	n/a
120	n/a	def __init__(self):
121	n/a	self.keys = {}
122	n/a
123	n/a	@property
124	n/a	def resolution(self):
125	n/a	return 1e-3
126	n/a
127	n/a	def register(self, fileobj, events, data=None):
128	n/a	key = selectors.SelectorKey(fileobj, 0, events, data)
129	n/a	self.keys[fileobj] = key
130	n/a	return key
131	n/a
132	n/a	def unregister(self, fileobj):
133	n/a	return self.keys.pop(fileobj)
134	n/a
135	n/a	def select(self, timeout=None):
136	n/a	block = False
137	n/a	for fileobj in self.keys:
138	n/a	if isinstance(fileobj, TelnetAlike):
139	n/a	block = fileobj.sock.block
140	n/a	break
141	n/a	if block:
142	n/a	return []
143	n/a	else:
144	n/a	return [(key, key.events) for key in self.keys.values()]
145	n/a
146	n/a	def get_map(self):
147	n/a	return self.keys
148	n/a
149	n/a
150	n/a	@contextlib.contextmanager
151	n/a	def test_socket(reads):
152	n/a	def new_conn(*ignored):
153	n/a	return SocketStub(reads)
154	n/a	try:
155	n/a	old_conn = socket.create_connection
156	n/a	socket.create_connection = new_conn
157	n/a	yield None
158	n/a	finally:
159	n/a	socket.create_connection = old_conn
160	n/a	return
161	n/a
162	n/a	def test_telnet(reads=(), cls=TelnetAlike):
163	n/a	''' return a telnetlib.Telnet object that uses a SocketStub with
164	n/a	reads queued up to be read '''
165	n/a	for x in reads:
166	n/a	assert type(x) is bytes, x
167	n/a	with test_socket(reads):
168	n/a	telnet = cls('dummy', 0)
169	n/a	telnet._messages = '' # debuglevel output
170	n/a	return telnet
171	n/a
172	n/a	class ExpectAndReadTestCase(unittest.TestCase):
173	n/a	def setUp(self):
174	n/a	self.old_selector = telnetlib._TelnetSelector
175	n/a	telnetlib._TelnetSelector = MockSelector
176	n/a	def tearDown(self):
177	n/a	telnetlib._TelnetSelector = self.old_selector
178	n/a
179	n/a	class ReadTests(ExpectAndReadTestCase):
180	n/a	def test_read_until(self):
181	n/a	"""
182	n/a	read_until(expected, timeout=None)
183	n/a	test the blocking version of read_util
184	n/a	"""
185	n/a	want = [b'xxxmatchyyy']
186	n/a	telnet = test_telnet(want)
187	n/a	data = telnet.read_until(b'match')
188	n/a	self.assertEqual(data, b'xxxmatch', msg=(telnet.cookedq, telnet.rawq, telnet.sock.reads))
189	n/a
190	n/a	reads = [b'x' * 50, b'match', b'y' * 50]
191	n/a	expect = b''.join(reads[:-1])
192	n/a	telnet = test_telnet(reads)
193	n/a	data = telnet.read_until(b'match')
194	n/a	self.assertEqual(data, expect)
195	n/a
196	n/a
197	n/a	def test_read_all(self):
198	n/a	"""
199	n/a	read_all()
200	n/a	Read all data until EOF; may block.
201	n/a	"""
202	n/a	reads = [b'x' * 500, b'y' * 500, b'z' * 500]
203	n/a	expect = b''.join(reads)
204	n/a	telnet = test_telnet(reads)
205	n/a	data = telnet.read_all()
206	n/a	self.assertEqual(data, expect)
207	n/a	return
208	n/a
209	n/a	def test_read_some(self):
210	n/a	"""
211	n/a	read_some()
212	n/a	Read at least one byte or EOF; may block.
213	n/a	"""
214	n/a	# test 'at least one byte'
215	n/a	telnet = test_telnet([b'x' * 500])
216	n/a	data = telnet.read_some()
217	n/a	self.assertTrue(len(data) >= 1)
218	n/a	# test EOF
219	n/a	telnet = test_telnet()
220	n/a	data = telnet.read_some()
221	n/a	self.assertEqual(b'', data)
222	n/a
223	n/a	def _read_eager(self, func_name):
224	n/a	"""
225	n/a	read_*_eager()
226	n/a	Read all data available already queued or on the socket,
227	n/a	without blocking.
228	n/a	"""
229	n/a	want = b'x' * 100
230	n/a	telnet = test_telnet([want])
231	n/a	func = getattr(telnet, func_name)
232	n/a	telnet.sock.block = True
233	n/a	self.assertEqual(b'', func())
234	n/a	telnet.sock.block = False
235	n/a	data = b''
236	n/a	while True:
237	n/a	try:
238	n/a	data += func()
239	n/a	except EOFError:
240	n/a	break
241	n/a	self.assertEqual(data, want)
242	n/a
243	n/a	def test_read_eager(self):
244	n/a	# read_eager and read_very_eager make the same guarantees
245	n/a	# (they behave differently but we only test the guarantees)
246	n/a	self._read_eager('read_eager')
247	n/a	self._read_eager('read_very_eager')
248	n/a	# NB -- we need to test the IAC block which is mentioned in the
249	n/a	# docstring but not in the module docs
250	n/a
251	n/a	def read_very_lazy(self):
252	n/a	want = b'x' * 100
253	n/a	telnet = test_telnet([want])
254	n/a	self.assertEqual(b'', telnet.read_very_lazy())
255	n/a	while telnet.sock.reads:
256	n/a	telnet.fill_rawq()
257	n/a	data = telnet.read_very_lazy()
258	n/a	self.assertEqual(want, data)
259	n/a	self.assertRaises(EOFError, telnet.read_very_lazy)
260	n/a
261	n/a	def test_read_lazy(self):
262	n/a	want = b'x' * 100
263	n/a	telnet = test_telnet([want])
264	n/a	self.assertEqual(b'', telnet.read_lazy())
265	n/a	data = b''
266	n/a	while True:
267	n/a	try:
268	n/a	read_data = telnet.read_lazy()
269	n/a	data += read_data
270	n/a	if not read_data:
271	n/a	telnet.fill_rawq()
272	n/a	except EOFError:
273	n/a	break
274	n/a	self.assertTrue(want.startswith(data))
275	n/a	self.assertEqual(data, want)
276	n/a
277	n/a	class nego_collector(object):
278	n/a	def __init__(self, sb_getter=None):
279	n/a	self.seen = b''
280	n/a	self.sb_getter = sb_getter
281	n/a	self.sb_seen = b''
282	n/a
283	n/a	def do_nego(self, sock, cmd, opt):
284	n/a	self.seen += cmd + opt
285	n/a	if cmd == tl.SE and self.sb_getter:
286	n/a	sb_data = self.sb_getter()
287	n/a	self.sb_seen += sb_data
288	n/a
289	n/a	tl = telnetlib
290	n/a
291	n/a	class WriteTests(unittest.TestCase):
292	n/a	'''The only thing that write does is replace each tl.IAC for
293	n/a	tl.IAC+tl.IAC'''
294	n/a
295	n/a	def test_write(self):
296	n/a	data_sample = [b'data sample without IAC',
297	n/a	b'data sample with' + tl.IAC + b' one IAC',
298	n/a	b'a few' + tl.IAC + tl.IAC + b' iacs' + tl.IAC,
299	n/a	tl.IAC,
300	n/a	b'']
301	n/a	for data in data_sample:
302	n/a	telnet = test_telnet()
303	n/a	telnet.write(data)
304	n/a	written = b''.join(telnet.sock.writes)
305	n/a	self.assertEqual(data.replace(tl.IAC,tl.IAC+tl.IAC), written)
306	n/a
307	n/a	class OptionTests(unittest.TestCase):
308	n/a	# RFC 854 commands
309	n/a	cmds = [tl.AO, tl.AYT, tl.BRK, tl.EC, tl.EL, tl.GA, tl.IP, tl.NOP]
310	n/a
311	n/a	def _test_command(self, data):
312	n/a	""" helper for testing IAC + cmd """
313	n/a	telnet = test_telnet(data)
314	n/a	data_len = len(b''.join(data))
315	n/a	nego = nego_collector()
316	n/a	telnet.set_option_negotiation_callback(nego.do_nego)
317	n/a	txt = telnet.read_all()
318	n/a	cmd = nego.seen
319	n/a	self.assertTrue(len(cmd) > 0) # we expect at least one command
320	n/a	self.assertIn(cmd[:1], self.cmds)
321	n/a	self.assertEqual(cmd[1:2], tl.NOOPT)
322	n/a	self.assertEqual(data_len, len(txt + cmd))
323	n/a	nego.sb_getter = None # break the nego => telnet cycle
324	n/a
325	n/a	def test_IAC_commands(self):
326	n/a	for cmd in self.cmds:
327	n/a	self._test_command([tl.IAC, cmd])
328	n/a	self._test_command([b'x' * 100, tl.IAC, cmd, b'y'*100])
329	n/a	self._test_command([b'x' * 10, tl.IAC, cmd, b'y'*10])
330	n/a	# all at once
331	n/a	self._test_command([tl.IAC + cmd for (cmd) in self.cmds])
332	n/a
333	n/a	def test_SB_commands(self):
334	n/a	# RFC 855, subnegotiations portion
335	n/a	send = [tl.IAC + tl.SB + tl.IAC + tl.SE,
336	n/a	tl.IAC + tl.SB + tl.IAC + tl.IAC + tl.IAC + tl.SE,
337	n/a	tl.IAC + tl.SB + tl.IAC + tl.IAC + b'aa' + tl.IAC + tl.SE,
338	n/a	tl.IAC + tl.SB + b'bb' + tl.IAC + tl.IAC + tl.IAC + tl.SE,
339	n/a	tl.IAC + tl.SB + b'cc' + tl.IAC + tl.IAC + b'dd' + tl.IAC + tl.SE,
340	n/a	]
341	n/a	telnet = test_telnet(send)
342	n/a	nego = nego_collector(telnet.read_sb_data)
343	n/a	telnet.set_option_negotiation_callback(nego.do_nego)
344	n/a	txt = telnet.read_all()
345	n/a	self.assertEqual(txt, b'')
346	n/a	want_sb_data = tl.IAC + tl.IAC + b'aabb' + tl.IAC + b'cc' + tl.IAC + b'dd'
347	n/a	self.assertEqual(nego.sb_seen, want_sb_data)
348	n/a	self.assertEqual(b'', telnet.read_sb_data())
349	n/a	nego.sb_getter = None # break the nego => telnet cycle
350	n/a
351	n/a	def test_debuglevel_reads(self):
352	n/a	# test all the various places that self.msg(...) is called
353	n/a	given_a_expect_b = [
354	n/a	# Telnet.fill_rawq
355	n/a	(b'a', ": recv b''\n"),
356	n/a	# Telnet.process_rawq
357	n/a	(tl.IAC + bytes([88]), ": IAC 88 not recognized\n"),
358	n/a	(tl.IAC + tl.DO + bytes([1]), ": IAC DO 1\n"),
359	n/a	(tl.IAC + tl.DONT + bytes([1]), ": IAC DONT 1\n"),
360	n/a	(tl.IAC + tl.WILL + bytes([1]), ": IAC WILL 1\n"),
361	n/a	(tl.IAC + tl.WONT + bytes([1]), ": IAC WONT 1\n"),
362	n/a	]
363	n/a	for a, b in given_a_expect_b:
364	n/a	telnet = test_telnet([a])
365	n/a	telnet.set_debuglevel(1)
366	n/a	txt = telnet.read_all()
367	n/a	self.assertIn(b, telnet._messages)
368	n/a	return
369	n/a
370	n/a	def test_debuglevel_write(self):
371	n/a	telnet = test_telnet()
372	n/a	telnet.set_debuglevel(1)
373	n/a	telnet.write(b'xxx')
374	n/a	expected = "send b'xxx'\n"
375	n/a	self.assertIn(expected, telnet._messages)
376	n/a
377	n/a	def test_debug_accepts_str_port(self):
378	n/a	# Issue 10695
379	n/a	with test_socket([]):
380	n/a	telnet = TelnetAlike('dummy', '0')
381	n/a	telnet._messages = ''
382	n/a	telnet.set_debuglevel(1)
383	n/a	telnet.msg('test')
384	n/a	self.assertRegex(telnet._messages, r'0.*test')
385	n/a
386	n/a
387	n/a	class ExpectTests(ExpectAndReadTestCase):
388	n/a	def test_expect(self):
389	n/a	"""
390	n/a	expect(expected, [timeout])
391	n/a	Read until the expected string has been seen, or a timeout is
392	n/a	hit (default is no timeout); may block.
393	n/a	"""
394	n/a	want = [b'x' * 10, b'match', b'y' * 10]
395	n/a	telnet = test_telnet(want)
396	n/a	(_,_,data) = telnet.expect([b'match'])
397	n/a	self.assertEqual(data, b''.join(want[:-1]))
398	n/a
399	n/a
400	n/a	if __name__ == '__main__':
401	n/a	unittest.main()