annotate tests/dummysmtpd.py @ 37271:0194dac77c93

scmutil: add method for looking up a context given a revision symbol

changectx's constructor currently supports a mix if inputs:

  * integer revnums
  * binary nodeids
  * '.', 'tip', 'null'
  * stringified revnums
  * namespaced identifiers (e.g. bookmarks and tags)
  * hex nodeids
  * partial hex nodeids

The first two are always internal [1]. The other five can be specified
by the user. The third type ('.', 'tip', 'null') often comes from
either the user or internal callers. We probably have some internal
callers that pass hex nodeids too, perhaps even partial ones
(histedit?). There are only a few callers that pass user-supplied
strings: revsets.stringset, peer.lookup, webutil.changeidctx, and
maybe one or two more.

Supporting this mix of things in the constructor is convenient, but a
bit strange, IMO. For example, if repo[node] is given a node that's
not in the repo, it will first check if it's bookmark etc before
raising an exception. Of course, the risk of it being a bookmark is
extremely small, but it just feels ugly.

Also, a problem with having this code in the constructor (whether it
supports a mix of types or not) is that it's harder to override (I'd
like to override it, and that's how this series started).

This patch starts moving out the handling of user-supplied strings by
introducing scmutil.revsymbol(). So far, that just checks that the
input is indeed a string, and then delegates to repo[symbol]. The
patch also calls it from revsets.stringset to prove that it works.

 [1] Well, you probably can enter a 20-byte binary nodeid on the
     command line, but I don't think we should care to preserve
     support for that.

Differential Revision: https://phab.mercurial-scm.org/D3024

#!/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
import traceback

from mercurial import (
    pycompat,
    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)))

    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()

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 _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():
    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(
        bytesvars(opts), initfn=init, runfn=run,
        runargs=[pycompat.sysexecutable,
                 pycompat.fsencode(__file__)] + pycompat.sysargv[1:])

if __name__ == '__main__':
    main()