Python源码示例：smtpd.DebuggingServer()
示例1
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = StringIO.StringIO()
sys.stdout = self.output
self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例2
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = StringIO.StringIO()
sys.stdout = self.output
self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例3
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = StringIO.StringIO()
sys.stdout = self.output
self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例4
def setUp(self):
self.real_getfqdn = socket.getfqdn
socket.getfqdn = mock_socket.getfqdn
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = io.StringIO()
sys.stdout = self.output
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Capture SMTPChannel debug output
self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
smtpd.DEBUGSTREAM = io.StringIO()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1),
decode_data=True)
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例5
def test_process_message_with_decode_data_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
decode_data=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nhello\n')
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
From: test
X-Peer: peer-address
hello
------------ END MESSAGE ------------
"""))
示例6
def test_process_message_with_enable_SMTPUTF8_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
enable_SMTPUTF8=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
b'From: test'
b'X-Peer: peer-address'
b''
b'h\\xc3\\xa9llo\\xff'
------------ END MESSAGE ------------
"""))
示例7
def test_process_SMTPUTF8_message_with_enable_SMTPUTF8_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
enable_SMTPUTF8=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n',
enable_SMTPUTF8=True)
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
mail options: ['BODY=8BITMIME', 'SMTPUTF8']
b'From: test'
b'X-Peer: peer-address'
b''
b'h\\xc3\\xa9llo\\xff'
------------ END MESSAGE ------------
"""))
示例8
def setUp(self):
self.real_getfqdn = socket.getfqdn
socket.getfqdn = mock_socket.getfqdn
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = io.StringIO()
sys.stdout = self.output
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Capture SMTPChannel debug output
self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
smtpd.DEBUGSTREAM = io.StringIO()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例9
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = StringIO.StringIO()
sys.stdout = self.output
self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例10
def setUp(self):
self.real_getfqdn = socket.getfqdn
socket.getfqdn = mock_socket.getfqdn
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = io.StringIO()
sys.stdout = self.output
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Capture SMTPChannel debug output
self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
smtpd.DEBUGSTREAM = io.StringIO()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1),
decode_data=True)
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例11
def test_process_message_with_decode_data_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
decode_data=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nhello\n')
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
From: test
X-Peer: peer-address
hello
------------ END MESSAGE ------------
"""))
示例12
def test_process_message_with_enable_SMTPUTF8_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
enable_SMTPUTF8=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
b'From: test'
b'X-Peer: peer-address'
b''
b'h\\xc3\\xa9llo\\xff'
------------ END MESSAGE ------------
"""))
示例13
def test_process_SMTPUTF8_message_with_enable_SMTPUTF8_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
enable_SMTPUTF8=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n',
enable_SMTPUTF8=True)
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
mail options: ['BODY=8BITMIME', 'SMTPUTF8']
b'From: test'
b'X-Peer: peer-address'
b''
b'h\\xc3\\xa9llo\\xff'
------------ END MESSAGE ------------
"""))
示例14
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = StringIO.StringIO()
sys.stdout = self.output
self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例15
def setUp(self):
self.real_getfqdn = socket.getfqdn
socket.getfqdn = mock_socket.getfqdn
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = io.StringIO()
sys.stdout = self.output
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Capture SMTPChannel debug output
self.old_DEBUGSTREAM = smtpd.DEBUGSTREAM
smtpd.DEBUGSTREAM = io.StringIO()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1),
decode_data=True)
# Keep a note of what server host and port were assigned
self.host, self.port = self.serv.socket.getsockname()[:2]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例16
def testSendMessage(self):
m = email.mime.text.MIMEText('A test message')
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
smtp.send_message(m, from_addr='John', to_addrs='Sally')
# XXX (see comment in testSend)
time.sleep(0.01)
smtp.quit()
self.client_evt.set()
self.serv_evt.wait()
self.output.flush()
# Remove the X-Peer header that DebuggingServer adds as figuring out
# exactly what IP address format is put there is not easy (and
# irrelevant to our test). Typically 127.0.0.1 or ::1, but it is
# not always the same as socket.gethostbyname(HOST). :(
test_output = self.get_output_without_xpeer()
del m['X-Peer']
mexpect = '%s%s\n%s' % (MSG_BEGIN, m.as_string(), MSG_END)
self.assertEqual(test_output, mexpect)
示例17
def test_process_message_with_decode_data_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
decode_data=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, decode_data=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nhello\n')
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
From: test
X-Peer: peer-address
hello
------------ END MESSAGE ------------
"""))
示例18
def test_process_message_with_enable_SMTPUTF8_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
enable_SMTPUTF8=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n')
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
b'From: test'
b'X-Peer: peer-address'
b''
b'h\\xc3\\xa9llo\\xff'
------------ END MESSAGE ------------
"""))
示例19
def test_process_SMTPUTF8_message_with_enable_SMTPUTF8_true(self):
server = smtpd.DebuggingServer((support.HOST, 0), ('b', 0),
enable_SMTPUTF8=True)
conn, addr = server.accept()
channel = smtpd.SMTPChannel(server, conn, addr, enable_SMTPUTF8=True)
with support.captured_stdout() as s:
self.send_data(channel, b'From: test\n\nh\xc3\xa9llo\xff\n',
enable_SMTPUTF8=True)
stdout = s.getvalue()
self.assertEqual(stdout, textwrap.dedent("""\
---------- MESSAGE FOLLOWS ----------
mail options: ['BODY=8BITMIME', 'SMTPUTF8']
b'From: test'
b'X-Peer: peer-address'
b''
b'h\\xc3\\xa9llo\\xff'
------------ END MESSAGE ------------
"""))
示例20
def setUp(self):
# temporarily replace sys.stdout to capture DebuggingServer output
self.old_stdout = sys.stdout
self.output = StringIO.StringIO()
sys.stdout = self.output
self._threads = test_support.threading_setup()
self.serv_evt = threading.Event()
self.client_evt = threading.Event()
# Pick a random unused port by passing 0 for the port number
self.serv = smtpd.DebuggingServer((HOST, 0), ('nowhere', -1))
# Keep a note of what port was assigned
self.port = self.serv.socket.getsockname()[1]
serv_args = (self.serv, self.serv_evt, self.client_evt)
self.thread = threading.Thread(target=debugging_server, args=serv_args)
self.thread.start()
# wait until server thread has assigned a port number
self.serv_evt.wait()
self.serv_evt.clear()
示例21
def testNotImplemented(self):
# EHLO isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
expected = (502, 'Error: command "EHLO" not implemented')
self.assertEqual(smtp.ehlo(), expected)
smtp.quit()
示例22
def testVRFY(self):
# VRFY isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=15)
expected = (502, 'Error: command "VRFY" not implemented')
self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
smtp.quit()
示例23
def handle_error(self):
raise
# Test various SMTP & ESMTP commands/behaviors that require a simulated server
# (i.e., something with more features than DebuggingServer)
示例24
def testNotImplemented(self):
# EHLO isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
expected = (502, 'Error: command "EHLO" not implemented')
self.assertEqual(smtp.ehlo(), expected)
smtp.quit()
示例25
def testVRFY(self):
# VRFY isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
expected = (502, 'Error: command "VRFY" not implemented')
self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
smtp.quit()
示例26
def handle_error(self):
raise
# Test various SMTP & ESMTP commands/behaviors that require a simulated server
# (i.e., something with more features than DebuggingServer)
示例27
def testNotImplemented(self):
# EHLO isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
expected = (502, 'Error: command "EHLO" not implemented')
self.assertEqual(smtp.ehlo(), expected)
smtp.quit()
示例28
def testVRFY(self):
# VRFY isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
expected = (502, 'Error: command "VRFY" not implemented')
self.assertEqual(smtp.vrfy('nobody@nowhere.com'), expected)
self.assertEqual(smtp.verify('nobody@nowhere.com'), expected)
smtp.quit()
示例29
def handle_error(self):
raise
# Test various SMTP & ESMTP commands/behaviors that require a simulated server
# (i.e., something with more features than DebuggingServer)
示例30
def testELHO(self):
# EHLO isn't implemented in DebuggingServer
smtp = smtplib.SMTP(HOST, self.port, local_hostname='localhost', timeout=3)
expected = (250, b'\nSIZE 33554432\nHELP')
self.assertEqual(smtp.ehlo(), expected)
smtp.quit()
