mirror of
https://github.com/facebook/sapling.git
synced 2024-10-16 19:57:18 +03:00
dfda82e492
Summary: Mostly empty lines removed and added. A few bugfixes on excessive line splitting. Reviewed By: quark-zju Differential Revision: D8199128 fbshipit-source-id: 90c1616061bfd7cfbba0b75f03f89683340374d5
90 lines
2.3 KiB
Python
Executable File
90 lines
2.3 KiB
Python
Executable File
#!/usr/bin/env python
|
|
|
|
"""dummy SMTP server for use in tests"""
|
|
|
|
from __future__ import absolute_import
|
|
|
|
import asyncore
|
|
import optparse
|
|
import smtpd
|
|
import ssl
|
|
import sys
|
|
|
|
from mercurial import server, sslutil, ui as uimod
|
|
|
|
|
|
def log(msg):
|
|
sys.stdout.write(msg)
|
|
sys.stdout.flush()
|
|
|
|
|
|
class dummysmtpserver(smtpd.SMTPServer):
|
|
def __init__(self, localaddr):
|
|
smtpd.SMTPServer.__init__(self, localaddr, remoteaddr=None)
|
|
|
|
def process_message(self, peer, mailfrom, rcpttos, data):
|
|
log("%s from=%s to=%s\n" % (peer[0], mailfrom, ", ".join(rcpttos)))
|
|
|
|
|
|
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
|
|
ui = uimod.ui.load()
|
|
try:
|
|
# wrap_socket() would block, but we don't care
|
|
conn = sslutil.wrapserversocket(conn, ui, certfile=self._certfile)
|
|
except ssl.SSLError:
|
|
log("%s ssl error\n" % addr[0])
|
|
conn.close()
|
|
return
|
|
smtpd.SMTPChannel(self, conn, addr)
|
|
|
|
|
|
def run():
|
|
try:
|
|
asyncore.loop()
|
|
except KeyboardInterrupt:
|
|
pass
|
|
|
|
|
|
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)
|
|
|
|
def init():
|
|
if opts.tls == "none":
|
|
dummysmtpserver(addr)
|
|
else:
|
|
dummysmtpsecureserver(addr, opts.certificate)
|
|
log("listening at %s:%d\n" % addr)
|
|
|
|
server.runservice(
|
|
vars(opts),
|
|
initfn=init,
|
|
runfn=run,
|
|
runargs=[sys.executable, __file__] + sys.argv[1:],
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|