Mercurial > hg
view tests/dummysmtpd.py @ 34192:da2f5f19312c
uncommit: move fb-extension to core which uncommits a changeset
uncommit extension in fb-hgext adds a uncommit command which by default
uncommits a changeset and move all the changes to the working directory. If
file names are passed, uncommit moves the changes from those files to the
working directory and left the changeset with remaining committed files.
The uncommit extension in fb-hgext does not creates an empty commit like the one
in evolve extension unless user has specified ui.alllowemptycommit to True.
The test file added is a combination of tests from test-uncommit.t,
test-uncommit-merge.t and test-uncommit-bookmark.t from fb-hgext.
.. feature::
A new uncommit extension which provides `hg uncommit` using which one can
uncommit part or all of the changeset. This command undoes the effect of a
local commit, returning the affected files to their uncommitted state.
Differential Revision: https://phab.mercurial-scm.org/D529
author | Pulkit Goyal <7895pulkit@gmail.com> |
---|---|
date | Thu, 24 Aug 2017 22:55:56 +0530 |
parents | d83ca854fa21 |
children | 75bae69747f0 |
line wrap: on
line source
#!/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()