Lines Matching refs:self
29 def setUp(self):
30 self.evt = threading.Event()
31 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
32 self.sock.settimeout(60) # Safety net. Look issue 11812
33 self.port = socket_helper.bind_port(self.sock)
34 self.thread = threading.Thread(target=server, args=(self.evt,self.sock))
35 self.thread.daemon = True
36 self.thread.start()
37 self.evt.wait()
39 def tearDown(self):
40 self.thread.join()
41 del self.thread # Clear out any dangling Thread objects.
43 def testBasic(self):
45 telnet = telnetlib.Telnet(HOST, self.port)
48 def testContextManager(self):
49 with telnetlib.Telnet(HOST, self.port) as tn:
50 self.assertIsNotNone(tn.get_socket())
51 self.assertIsNone(tn.get_socket())
53 def testTimeoutDefault(self):
54 self.assertTrue(socket.getdefaulttimeout() is None)
57 telnet = telnetlib.Telnet(HOST, self.port)
60 self.assertEqual(telnet.sock.gettimeout(), 30)
63 def testTimeoutNone(self):
65 self.assertTrue(socket.getdefaulttimeout() is None)
68 telnet = telnetlib.Telnet(HOST, self.port, timeout=None)
71 self.assertTrue(telnet.sock.gettimeout() is None)
74 def testTimeoutValue(self):
75 telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
76 self.assertEqual(telnet.sock.gettimeout(), 30)
79 def testTimeoutOpen(self):
81 telnet.open(HOST, self.port, timeout=30)
82 self.assertEqual(telnet.sock.gettimeout(), 30)
85 def testGetters(self):
87 telnet = telnetlib.Telnet(HOST, self.port, timeout=30)
89 self.assertEqual(telnet.get_socket(), t_sock)
90 self.assertEqual(telnet.fileno(), t_sock.fileno())
95 def __init__(self, reads=()):
96 self.reads = list(reads) # Intentionally make a copy.
97 self.writes = []
98 self.block = False
99 def sendall(self, data):
100 self.writes.append(data)
101 def recv(self, size):
103 while self.reads and len(out) < size:
104 out += self.reads.pop(0)
106 self.reads.insert(0, out[size:])
111 def fileno(self):
113 def close(self): pass
114 def sock_avail(self):
115 return (not self.sock.block)
116 def msg(self, msg, *args):
118 telnetlib.Telnet.msg(self, msg, *args)
119 self._messages += out.getvalue()
124 def __init__(self):
125 self.keys = {}
128 def resolution(self):
131 def register(self, fileobj, events, data=None):
133 self.keys[fileobj] = key
136 def unregister(self, fileobj):
137 return self.keys.pop(fileobj)
139 def select(self, timeout=None):
141 for fileobj in self.keys:
148 return [(key, key.events) for key in self.keys.values()]
150 def get_map(self):
151 return self.keys
177 def setUp(self):
178 self.old_selector = telnetlib._TelnetSelector
180 def tearDown(self):
181 telnetlib._TelnetSelector = self.old_selector
184 def test_read_until(self):
192 self.assertEqual(data, b'xxxmatch', msg=(telnet.cookedq, telnet.rawq, telnet.sock.reads))
198 self.assertEqual(data, expect)
201 def test_read_all(self):
210 self.assertEqual(data, expect)
213 def test_read_some(self):
221 self.assertTrue(len(data) >= 1)
225 self.assertEqual(b'', data)
227 def _read_eager(self, func_name):
237 self.assertEqual(b'', func())
245 self.assertEqual(data, want)
247 def test_read_eager(self):
250 self._read_eager('read_eager')
251 self._read_eager('read_very_eager')
255 def read_very_lazy(self):
258 self.assertEqual(b'', telnet.read_very_lazy())
262 self.assertEqual(want, data)
263 self.assertRaises(EOFError, telnet.read_very_lazy)
265 def test_read_lazy(self):
268 self.assertEqual(b'', telnet.read_lazy())
278 self.assertTrue(want.startswith(data))
279 self.assertEqual(data, want)
282 def __init__(self, sb_getter=None):
283 self.seen = b''
284 self.sb_getter = sb_getter
285 self.sb_seen = b''
287 def do_nego(self, sock, cmd, opt):
288 self.seen += cmd + opt
289 if cmd == tl.SE and self.sb_getter:
290 sb_data = self.sb_getter()
291 self.sb_seen += sb_data
299 def test_write(self):
309 self.assertEqual(data.replace(tl.IAC,tl.IAC+tl.IAC), written)
315 def _test_command(self, data):
323 self.assertTrue(len(cmd) > 0) # we expect at least one command
324 self.assertIn(cmd[:1], self.cmds)
325 self.assertEqual(cmd[1:2], tl.NOOPT)
326 self.assertEqual(data_len, len(txt + cmd))
329 def test_IAC_commands(self):
330 for cmd in self.cmds:
331 self._test_command([tl.IAC, cmd])
332 self._test_command([b'x' * 100, tl.IAC, cmd, b'y'*100])
333 self._test_command([b'x' * 10, tl.IAC, cmd, b'y'*10])
335 self._test_command([tl.IAC + cmd for (cmd) in self.cmds])
337 def test_SB_commands(self):
349 self.assertEqual(txt, b'')
351 self.assertEqual(nego.sb_seen, want_sb_data)
352 self.assertEqual(b'', telnet.read_sb_data())
355 def test_debuglevel_reads(self):
356 # test all the various places that self.msg(...) is called
371 self.assertIn(b, telnet._messages)
374 def test_debuglevel_write(self):
379 self.assertIn(expected, telnet._messages)
381 def test_debug_accepts_str_port(self):
388 self.assertRegex(telnet._messages, r'0.*test')
392 def test_expect(self):
401 self.assertEqual(data, b''.join(want[:-1]))