Details
Details
- Reviewers
 - None
 - Group Reviewers
 hg-reviewers - Commits
 - rHGed96d1116302: tests: help dummysmtpd work on python 3
 
Diff Detail
Diff Detail
- Repository
 - rHG Mercurial
 - Lint
 Lint Skipped - Unit
 Unit Tests Skipped 
| hg-reviewers | 
| Lint Skipped | 
| Unit Tests Skipped | 
| Path | Packages | |||
|---|---|---|---|---|
| M | tests/dummysmtpd.py (20 lines) | 
| Status | Author | Revision | |
|---|---|---|---|
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Abandoned | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Closed | durin42 | ||
| Abandoned | durin42 | 
| #!/usr/bin/env python | #!/usr/bin/env python | ||||
| """dummy SMTP server for use in tests""" | """dummy SMTP server for use in tests""" | ||||
| from __future__ import absolute_import | from __future__ import absolute_import | ||||
| import asyncore | import asyncore | ||||
| import optparse | import optparse | ||||
| import smtpd | import smtpd | ||||
| import ssl | import ssl | ||||
| import sys | import sys | ||||
| import traceback | import traceback | ||||
| from mercurial import ( | from mercurial import ( | ||||
| pycompat, | |||||
| server, | server, | ||||
| sslutil, | sslutil, | ||||
| ui as uimod, | ui as uimod, | ||||
| ) | ) | ||||
| def log(msg): | def log(msg): | ||||
| sys.stdout.write(msg) | sys.stdout.write(msg) | ||||
| sys.stdout.flush() | sys.stdout.flush() | ||||
| smtpd.SMTPChannel(self, conn, addr) | smtpd.SMTPChannel(self, conn, addr) | ||||
| def run(): | def run(): | ||||
| try: | try: | ||||
| asyncore.loop() | asyncore.loop() | ||||
| except KeyboardInterrupt: | except KeyboardInterrupt: | ||||
| pass | pass | ||||
| def _encodestrsonly(v): | |||||
| if isinstance(v, type(u'')): | |||||
| return v.encode('ascii') | |||||
| return v | |||||
| 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'] = [ | |||||
| _encodestrsonly(v) for v in bd[b'daemon_postexec']] | |||||
| return bd | |||||
| def main(): | def main(): | ||||
| op = optparse.OptionParser() | op = optparse.OptionParser() | ||||
| op.add_option('-d', '--daemon', action='store_true') | op.add_option('-d', '--daemon', action='store_true') | ||||
| op.add_option('--daemon-postexec', action='append') | op.add_option('--daemon-postexec', action='append') | ||||
| op.add_option('-p', '--port', type=int, default=8025) | op.add_option('-p', '--port', type=int, default=8025) | ||||
| op.add_option('-a', '--address', default='localhost') | op.add_option('-a', '--address', default='localhost') | ||||
| op.add_option('--pid-file', metavar='FILE') | op.add_option('--pid-file', metavar='FILE') | ||||
| op.add_option('--tls', choices=['none', 'smtps'], default='none') | op.add_option('--tls', choices=['none', 'smtps'], default='none') | ||||
| op.add_option('--certificate', metavar='FILE') | op.add_option('--certificate', metavar='FILE') | ||||
| opts, args = op.parse_args() | opts, args = op.parse_args() | ||||
| if opts.tls == 'smtps' and not opts.certificate: | if opts.tls == 'smtps' and not opts.certificate: | ||||
| op.error('--certificate must be specified') | op.error('--certificate must be specified') | ||||
| addr = (opts.address, opts.port) | addr = (opts.address, opts.port) | ||||
| def init(): | def init(): | ||||
| if opts.tls == 'none': | if opts.tls == 'none': | ||||
| dummysmtpserver(addr) | dummysmtpserver(addr) | ||||
| else: | else: | ||||
| dummysmtpsecureserver(addr, opts.certificate) | dummysmtpsecureserver(addr, opts.certificate) | ||||
| log('listening at %s:%d\n' % addr) | log('listening at %s:%d\n' % addr) | ||||
| server.runservice(vars(opts), initfn=init, runfn=run, | server.runservice( | ||||
| runargs=[sys.executable, __file__] + sys.argv[1:]) | bytesvars(opts), initfn=init, runfn=run, | ||||
| runargs=[pycompat.sysexecutable, | |||||
| pycompat.fsencode(__file__)] + pycompat.sysargv[1:]) | |||||
| if __name__ == '__main__': | if __name__ == '__main__': | ||||
| main() | main() | ||||