ยปCore Development>Code coverage>Lib/test/test_telnetlib.py

Python code coverage for Lib/test/test_telnetlib.py

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