Lines Matching refs:self
58 def __init__(self, conn):
59 asynchat.async_chat.__init__(self, conn)
60 self.set_terminator(b"\r\n")
61 self.in_buffer = []
62 self.push('+OK dummy pop3 server ready. <timestamp>')
63 self.tls_active = False
64 self.tls_starting = False
66 def collect_incoming_data(self, data):
67 self.in_buffer.append(data)
69 def found_terminator(self):
70 line = b''.join(self.in_buffer)
72 self.in_buffer = []
79 if hasattr(self, 'cmd_' + cmd):
80 method = getattr(self, 'cmd_' + cmd)
83 self.push('-ERR unrecognized POP3 command "%s".' %cmd)
85 def handle_error(self):
88 def push(self, data):
89 asynchat.async_chat.push(self, data.encode("ISO-8859-1") + b'\r\n')
91 def cmd_echo(self, arg):
93 self.push(arg)
95 def cmd_user(self, arg):
97 self.push("-ERR no such user")
98 self.push('+OK password required')
100 def cmd_pass(self, arg):
102 self.push("-ERR wrong password")
103 self.push('+OK 10 messages')
105 def cmd_stat(self, arg):
106 self.push('+OK 10 100')
108 def cmd_list(self, arg):
110 self.push('+OK %s %s' % (arg, arg))
112 self.push('+OK')
113 asynchat.async_chat.push(self, LIST_RESP)
117 def cmd_retr(self, arg):
118 self.push('+OK %s bytes' %len(RETR_RESP))
119 asynchat.async_chat.push(self, RETR_RESP)
123 def cmd_dele(self, arg):
124 self.push('+OK message marked for deletion.')
126 def cmd_noop(self, arg):
127 self.push('+OK done nothing.')
129 def cmd_rpop(self, arg):
130 self.push('+OK done nothing.')
132 def cmd_apop(self, arg):
133 self.push('+OK done nothing.')
135 def cmd_quit(self, arg):
136 self.push('+OK closing.')
137 self.close_when_done()
139 def _get_capas(self):
140 _capas = dict(self.CAPAS)
141 if not self.tls_active and SUPPORTS_SSL:
145 def cmd_capa(self, arg):
146 self.push('+OK Capability list follows')
147 if self._get_capas():
148 for cap, params in self._get_capas().items():
152 self.push(' '.join(_ln))
153 self.push('.')
155 def cmd_utf8(self, arg):
156 self.push('+OK I know RFC6856'
157 if self.enable_UTF8
162 def cmd_stls(self, arg):
163 if self.tls_active is False:
164 self.push('+OK Begin TLS negotiation')
167 tls_sock = context.wrap_socket(self.socket,
171 self.del_channel()
172 self.set_socket(tls_sock)
173 self.tls_active = True
174 self.tls_starting = True
175 self.in_buffer = []
176 self._do_tls_handshake()
178 self.push('-ERR Command not permitted when TLS active')
180 def _do_tls_handshake(self):
182 self.socket.do_handshake()
188 return self.handle_close()
192 return self.handle_close()
196 return self.handle_close()
198 self.tls_active = True
199 self.tls_starting = False
201 def handle_read(self):
202 if self.tls_starting:
203 self._do_tls_handshake()
206 asynchat.async_chat.handle_read(self)
208 self.handle_close()
214 def __init__(self, address, af=socket.AF_INET):
215 threading.Thread.__init__(self)
216 asyncore.dispatcher.__init__(self)
217 self.daemon = True
218 self.create_socket(af, socket.SOCK_STREAM)
219 self.bind(address)
220 self.listen(5)
221 self.active = False
222 self.active_lock = threading.Lock()
223 self.host, self.port = self.socket.getsockname()[:2]
224 self.handler_instance = None
226 def start(self):
227 assert not self.active
228 self.__flag = threading.Event()
229 threading.Thread.start(self)
230 self.__flag.wait()
232 def run(self):
233 self.active = True
234 self.__flag.set()
236 while self.active and asyncore.socket_map:
237 with self.active_lock:
242 def stop(self):
243 assert self.active
244 self.active = False
245 self.join()
247 def handle_accepted(self, conn, addr):
248 self.handler_instance = self.handler(conn)
250 def handle_connect(self):
251 self.close()
254 def writable(self):
257 def handle_error(self):
262 def assertOK(self, resp):
263 self.assertTrue(resp.startswith(b"+OK"))
265 def setUp(self):
266 self.server = DummyPOP3Server((HOST, PORT))
267 self.server.start()
268 self.client = poplib.POP3(self.server.host, self.server.port,
271 def tearDown(self):
272 self.client.close()
273 self.server.stop()
275 self.server = None
277 def test_getwelcome(self):
278 self.assertEqual(self.client.getwelcome(),
281 def test_exceptions(self):
282 self.assertRaises(poplib.error_proto, self.client._shortcmd, 'echo -err')
284 def test_user(self):
285 self.assertOK(self.client.user('guido'))
286 self.assertRaises(poplib.error_proto, self.client.user, 'invalid')
288 def test_pass_(self):
289 self.assertOK(self.client.pass_('python'))
290 self.assertRaises(poplib.error_proto, self.client.user, 'invalid')
292 def test_stat(self):
293 self.assertEqual(self.client.stat(), (10, 100))
295 def test_list(self):
296 self.assertEqual(self.client.list()[1:],
299 self.assertTrue(self.client.list('1').endswith(b"OK 1 1"))
301 def test_retr(self):
307 foo = self.client.retr('foo')
308 self.assertEqual(foo, expected)
310 def test_too_long_lines(self):
311 self.assertRaises(poplib.error_proto, self.client._shortcmd,
314 def test_dele(self):
315 self.assertOK(self.client.dele('foo'))
317 def test_noop(self):
318 self.assertOK(self.client.noop())
320 def test_rpop(self):
321 self.assertOK(self.client.rpop('foo'))
324 def test_apop_normal(self):
325 self.assertOK(self.client.apop('foo', 'dummypassword'))
328 def test_apop_REDOS(self):
334 with test_support.swap_attr(self.client, 'welcome', evil_welcome):
336 self.assertRaises(poplib.error_proto, self.client.apop, 'a', 'kb')
338 def test_top(self):
344 self.assertEqual(self.client.top(1, 1), expected)
346 def test_uidl(self):
347 self.client.uidl()
348 self.client.uidl('foo')
350 def test_utf8_raises_if_unsupported(self):
351 self.server.handler.enable_UTF8 = False
352 self.assertRaises(poplib.error_proto, self.client.utf8)
354 def test_utf8(self):
355 self.server.handler.enable_UTF8 = True
357 result = self.client.utf8()
358 self.assertEqual(result, expected)
360 def test_capa(self):
361 capa = self.client.capa()
362 self.assertTrue('IMPLEMENTATION' in capa.keys())
364 def test_quit(self):
365 resp = self.client.quit()
366 self.assertTrue(resp)
367 self.assertIsNone(self.client.sock)
368 self.assertIsNone(self.client.file)
371 def test_stls_capa(self):
372 capa = self.client.capa()
373 self.assertTrue('STLS' in capa.keys())
376 def test_stls(self):
378 resp = self.client.stls()
379 self.assertEqual(resp, expected)
382 def test_stls_context(self):
386 self.assertEqual(ctx.verify_mode, ssl.CERT_REQUIRED)
387 self.assertEqual(ctx.check_hostname, True)
388 with self.assertRaises(ssl.CertificateError):
389 resp = self.client.stls(context=ctx)
390 self.client = poplib.POP3("localhost", self.server.port,
392 resp = self.client.stls(context=ctx)
393 self.assertEqual(resp, expected)
401 def __init__(self, conn):
402 asynchat.async_chat.__init__(self, conn)
403 self.secure_connection()
404 self.set_terminator(b"\r\n")
405 self.in_buffer = []
406 self.push('+OK dummy pop3 server ready. <timestamp>')
407 self.tls_active = True
408 self.tls_starting = False
415 def setUp(self):
416 self.server = DummyPOP3Server((HOST, PORT))
417 self.server.handler = DummyPOP3_SSLHandler
418 self.server.start()
419 self.client = poplib.POP3_SSL(self.server.host, self.server.port)
421 def test__all__(self):
422 self.assertIn('POP3_SSL', poplib.__all__)
424 def test_context(self):
428 self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
429 self.server.port, keyfile=CERTFILE, context=ctx)
430 self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
431 self.server.port, certfile=CERTFILE, context=ctx)
432 self.assertRaises(ValueError, poplib.POP3_SSL, self.server.host,
433 self.server.port, keyfile=CERTFILE,
436 self.client.quit()
437 self.client = poplib.POP3_SSL(self.server.host, self.server.port,
439 self.assertIsInstance(self.client.sock, ssl.SSLSocket)
440 self.assertIs(self.client.sock.context, ctx)
441 self.assertTrue(self.client.noop().startswith(b'+OK'))
443 def test_stls(self):
444 self.assertRaises(poplib.error_proto, self.client.stls)
448 def test_stls_capa(self):
449 capa = self.client.capa()
450 self.assertFalse('STLS' in capa.keys())
457 def setUp(self):
458 self.server = DummyPOP3Server((HOST, PORT))
459 self.server.start()
460 self.client = poplib.POP3(self.server.host, self.server.port,
462 self.client.stls()
464 def tearDown(self):
465 if self.client.file is not None and self.client.sock is not None:
467 self.client.quit()
472 self.client.close()
473 self.server.stop()
475 self.server = None
477 def test_stls(self):
478 self.assertRaises(poplib.error_proto, self.client.stls)
482 def test_stls_capa(self):
483 capa = self.client.capa()
484 self.assertFalse(b'STLS' in capa.keys())
489 def setUp(self):
490 self.evt = threading.Event()
491 self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
492 self.sock.settimeout(60) # Safety net. Look issue 11812
493 self.port = socket_helper.bind_port(self.sock)
494 self.thread = threading.Thread(target=self.server, args=(self.evt, self.sock))
495 self.thread.daemon = True
496 self.thread.start()
497 self.evt.wait()
499 def tearDown(self):
500 self.thread.join()
502 self.thread = None
504 def server(self, evt, serv):
516 def testTimeoutDefault(self):
517 self.assertIsNone(socket.getdefaulttimeout())
520 pop = poplib.POP3(HOST, self.port)
523 self.assertEqual(pop.sock.gettimeout(), test_support.LOOPBACK_TIMEOUT)
526 def testTimeoutNone(self):
527 self.assertIsNone(socket.getdefaulttimeout())
530 pop = poplib.POP3(HOST, self.port, timeout=None)
533 self.assertIsNone(pop.sock.gettimeout())
536 def testTimeoutValue(self):
537 pop = poplib.POP3(HOST, self.port, timeout=test_support.LOOPBACK_TIMEOUT)
538 self.assertEqual(pop.sock.gettimeout(), test_support.LOOPBACK_TIMEOUT)
540 with self.assertRaises(ValueError):
541 poplib.POP3(HOST, self.port, timeout=0)