dummysmtpd.py
121 lines
| 3.3 KiB
| text/x-python
|
PythonLexer
/ tests / dummysmtpd.py
r48294 | #!/usr/bin/env python | |||
|
r29332 | |||
"""dummy SMTP server for use in tests""" | ||||
from __future__ import absolute_import | ||||
import asyncore | ||||
import optparse | ||||
import smtpd | ||||
import ssl | ||||
import sys | ||||
|
r35794 | import traceback | ||
|
r29332 | |||
from mercurial import ( | ||||
|
r36584 | pycompat, | ||
|
r30506 | server, | ||
|
r29556 | sslutil, | ||
ui as uimod, | ||||
|
r29332 | ) | ||
|
r43346 | |||
|
r29332 | def log(msg): | ||
sys.stdout.write(msg) | ||||
sys.stdout.flush() | ||||
|
r43346 | |||
|
r29332 | class dummysmtpserver(smtpd.SMTPServer): | ||
def __init__(self, localaddr): | ||||
smtpd.SMTPServer.__init__(self, localaddr, remoteaddr=None) | ||||
|
r39064 | def process_message(self, peer, mailfrom, rcpttos, data, **kwargs): | ||
|
r29332 | log('%s from=%s to=%s\n' % (peer[0], mailfrom, ', '.join(rcpttos))) | ||
|
r35794 | def handle_error(self): | ||
# On Windows, a bad SSL connection sometimes generates a WSAECONNRESET. | ||||
# The default handler will shutdown this server, and then both the | ||||
# current connection and subsequent ones fail on the client side with | ||||
# "No connection could be made because the target machine actively | ||||
# refused it". If we eat the error, then the client properly aborts in | ||||
# the expected way, and the server is available for subsequent requests. | ||||
traceback.print_exc() | ||||
|
r43346 | |||
|
r29332 | class dummysmtpsecureserver(dummysmtpserver): | ||
def __init__(self, localaddr, certfile): | ||||
dummysmtpserver.__init__(self, localaddr) | ||||
self._certfile = certfile | ||||
def handle_accept(self): | ||||
pair = self.accept() | ||||
if not pair: | ||||
return | ||||
conn, addr = pair | ||||
|
r30559 | ui = uimod.ui.load() | ||
|
r29332 | try: | ||
# wrap_socket() would block, but we don't care | ||||
|
r29556 | conn = sslutil.wrapserversocket(conn, ui, certfile=self._certfile) | ||
|
r29332 | except ssl.SSLError: | ||
log('%s ssl error\n' % addr[0]) | ||||
conn.close() | ||||
return | ||||
smtpd.SMTPChannel(self, conn, addr) | ||||
|
r43346 | |||
|
r29332 | def run(): | ||
try: | ||||
asyncore.loop() | ||||
except KeyboardInterrupt: | ||||
pass | ||||
|
r43346 | |||
|
r36584 | def _encodestrsonly(v): | ||
if isinstance(v, type(u'')): | ||||
return v.encode('ascii') | ||||
return v | ||||
|
r43346 | |||
|
r36584 | def bytesvars(obj): | ||
unidict = vars(obj) | ||||
bd = {k.encode('ascii'): _encodestrsonly(v) for k, v in unidict.items()} | ||||
if bd[b'daemon_postexec'] is not None: | ||||
bd[b'daemon_postexec'] = [ | ||||
|
r43346 | _encodestrsonly(v) for v in bd[b'daemon_postexec'] | ||
] | ||||
|
r36584 | return bd | ||
|
r43346 | |||
|
r29332 | def main(): | ||
op = optparse.OptionParser() | ||||
op.add_option('-d', '--daemon', action='store_true') | ||||
op.add_option('--daemon-postexec', action='append') | ||||
op.add_option('-p', '--port', type=int, default=8025) | ||||
op.add_option('-a', '--address', default='localhost') | ||||
op.add_option('--pid-file', metavar='FILE') | ||||
op.add_option('--tls', choices=['none', 'smtps'], default='none') | ||||
op.add_option('--certificate', metavar='FILE') | ||||
opts, args = op.parse_args() | ||||
if opts.tls == 'smtps' and not opts.certificate: | ||||
op.error('--certificate must be specified') | ||||
addr = (opts.address, opts.port) | ||||
|
r43346 | |||
|
r29332 | def init(): | ||
if opts.tls == 'none': | ||||
dummysmtpserver(addr) | ||||
else: | ||||
dummysmtpsecureserver(addr, opts.certificate) | ||||
log('listening at %s:%d\n' % addr) | ||||
|
r36584 | server.runservice( | ||
|
r43346 | bytesvars(opts), | ||
initfn=init, | ||||
runfn=run, | ||||
runargs=[pycompat.sysexecutable, pycompat.fsencode(__file__)] | ||||
+ pycompat.sysargv[1:], | ||||
) | ||||
|
r29332 | |||
if __name__ == '__main__': | ||||
main() | ||||