Python源码示例：smtpd.SMTPServer()
示例1
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例2
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例3
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例4
def __init__(self):
localaddr = settings.LMTP_HOST, int(settings.LMTP_PORT)
log.info("lmtp_server_start", address=localaddr)
smtpd.SMTPServer.__init__(self, localaddr, remoteaddr=None)
super().__init__()
示例5
def __init__(self, debug=False, esmtp=True):
super(SMTPServer, self).__init__()
self.esmtp = esmtp
self.debug = debug
self._server = None
self._wait_for_server = threading.Semaphore(0)
示例6
def stop(self):
self._server.close()
super(SMTPServer, self).stop()
示例7
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例8
def __init__(self, *args, **kw):
# The base SMTP server turns these on automatically, but our test
# server is set up to munge the EHLO response, so we need to provide
# them as well. And yes, the call is to SMTPServer not SimSMTPServer.
self._extra_features = ['SMTPUTF8', '8BITMIME']
smtpd.SMTPServer.__init__(self, *args, **kw)
示例9
def __init__(self, *args, **kwargs):
smtpd.SMTPServer.__init__(self, *args, **kwargs)
self.messages = []
if self._decode_data:
self.return_status = 'return status'
else:
self.return_status = b'return status'
示例10
def test_process_message_unimplemented(self):
server = smtpd.SMTPServer((support.HOST, 0), ('b', 0),
decode_data=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
def write_line(line):
channel.socket.queue_recv(line)
channel.handle_read()
write_line(b'HELO example')
write_line(b'MAIL From:eggs@example')
write_line(b'RCPT To:spam@example')
write_line(b'DATA')
self.assertRaises(NotImplementedError, write_line, b'spam\r\n.\r\n')
示例11
def test_decode_data_default_warns(self):
with self.assertWarns(DeprecationWarning):
smtpd.SMTPServer((support.HOST, 0), ('b', 0))
示例12
def test_socket_uses_IPv6(self):
server = smtpd.SMTPServer((support.HOSTv6, 0), (support.HOST, 0),
decode_data=False)
self.assertEqual(server.socket.family, socket.AF_INET6)
示例13
def test_socket_uses_IPv4(self):
server = smtpd.SMTPServer((support.HOST, 0), (support.HOSTv6, 0),
decode_data=False)
self.assertEqual(server.socket.family, socket.AF_INET)
示例14
def __init__(self, addr, handler, poll_interval, sockmap):
smtpd.SMTPServer.__init__(self, addr, None, map=sockmap,
decode_data=True)
self.port = self.socket.getsockname()[1]
self._handler = handler
self._thread = None
self.poll_interval = poll_interval
示例15
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例16
def __init__(self, localaddr, remoteaddr):
smtpd.SMTPServer.__init__(self, localaddr, remoteaddr)
self.messages = []
示例17
def test_process_message_unimplemented(self):
server = smtpd.SMTPServer('a', 'b')
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr)
def write_line(line):
channel.socket.queue_recv(line)
channel.handle_read()
write_line(b'HELO example')
write_line(b'MAIL From:eggs@example')
write_line(b'RCPT To:spam@example')
write_line(b'DATA')
self.assertRaises(NotImplementedError, write_line, b'spam\r\n.\r\n')
示例18
def __init__(self, addr, handler, poll_interval, sockmap):
smtpd.SMTPServer.__init__(self, addr, None, map=sockmap)
self.port = self.socket.getsockname()[1]
self._handler = handler
self._thread = None
self.poll_interval = poll_interval
示例19
def __init__(self, *args, **kwargs):
self.messages = []
smtpd.SMTPServer.__init__(self, *args, **kwargs)
示例20
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例21
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例22
def __init__(self, *args, **kw):
# The base SMTP server turns these on automatically, but our test
# server is set up to munge the EHLO response, so we need to provide
# them as well. And yes, the call is to SMTPServer not SimSMTPServer.
self._extra_features = ['SMTPUTF8', '8BITMIME']
smtpd.SMTPServer.__init__(self, *args, **kw)
示例23
def __init__(self, *args, **kwargs):
smtpd.SMTPServer.__init__(self, *args, **kwargs)
self.messages = []
if self._decode_data:
self.return_status = 'return status'
else:
self.return_status = b'return status'
示例24
def test_process_message_unimplemented(self):
server = smtpd.SMTPServer((support.HOST, 0), ('b', 0),
decode_data=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
def write_line(line):
channel.socket.queue_recv(line)
channel.handle_read()
write_line(b'HELO example')
write_line(b'MAIL From:eggs@example')
write_line(b'RCPT To:spam@example')
write_line(b'DATA')
self.assertRaises(NotImplementedError, write_line, b'spam\r\n.\r\n')
示例25
def test_decode_data_default_warns(self):
with self.assertWarns(DeprecationWarning):
smtpd.SMTPServer((support.HOST, 0), ('b', 0))
示例26
def test_socket_uses_IPv6(self):
server = smtpd.SMTPServer((support.HOSTv6, 0), (support.HOST, 0),
decode_data=False)
self.assertEqual(server.socket.family, socket.AF_INET6)
示例27
def test_socket_uses_IPv4(self):
server = smtpd.SMTPServer((support.HOST, 0), (support.HOSTv6, 0),
decode_data=False)
self.assertEqual(server.socket.family, socket.AF_INET)
示例28
def __init__(self, addr, handler, poll_interval, sockmap):
smtpd.SMTPServer.__init__(self, addr, None, map=sockmap,
decode_data=True)
self.port = self.socket.getsockname()[1]
self._handler = handler
self._thread = None
self.poll_interval = poll_interval
示例29
def __init__(self, *args, **kw):
self._extra_features = []
smtpd.SMTPServer.__init__(self, *args, **kw)
示例30
def __init__(self, *args, **kwargs):
threading.Thread.__init__(self)
# New kwarg added in Python 3.5; default switching to False in 3.6.
# Setting a value only silences a deprecation warning in Python 3.5.
kwargs['decode_data'] = True
smtpd.SMTPServer.__init__(self, *args, **kwargs)
self._sink = []
self.active = False
self.active_lock = threading.Lock()
self.sink_lock = threading.Lock()
