12 files changed, 286 insertions, 114 deletions

diff --git a/mailman/bin/docs/master.txt b/mailman/bin/docs/master.txt
index 5741617ea..333893f78 100644
--- a/mailman/bin/docs/master.txt
+++ b/mailman/bin/docs/master.txt
@@ -6,17 +6,15 @@ directories.  In normal operation, a command line script called 'mailmanctl'
 is used to start, stop and manage the queue runners.  mailmanctl actually is
 just a wrapper around the real queue runner watcher script called master.py.
 
+    >>> from mailman.configuration import config
     >>> from mailman.tests.helpers import TestableMaster
 
 Start the master in a subthread.
 
-    >>> import threading
-    >>> event = threading.Event()
-    >>> event.clear()
-    >>> master = TestableMaster(event)
-    >>> master.start_qrunners()
-    >>> threading.Thread(target=master.loop).start()
-    >>> event.wait(5.0)
+    >>> master = TestableMaster()
+    >>> master.start()
+    >>> len(list(master.qrunner_pids)) == len(config.qrunners)
+    True
 
 Now verify that all the qrunners are running.
 
@@ -26,13 +24,12 @@ Now verify that all the qrunners are running.
     >>> for pid in master.qrunner_pids:
     ...     os.kill(pid, 0)
 
-Send a SIGTERM to all the child processes, which should exit them.  cleanup()
-waits until all the child processes have exited.
+Stop the master process, which should also kill (and not restart) the child
+queue runner processes.
 
-    >>> import signal
-    >>> for pid in master.qrunner_pids:
-    ...     os.kill(pid, signal.SIGTERM)
-    >>> master.cleanup()
+    >>> master.stop()
+
+None of the children are running now.
 
     >>> import errno
     >>> for pid in master.qrunner_pids:
diff --git a/mailman/bin/master.py b/mailman/bin/master.py
index a85958836..4170ac4f4 100644
--- a/mailman/bin/master.py
+++ b/mailman/bin/master.py
@@ -89,6 +89,12 @@ level of checking.  If a process matching the host/pid described in the lock
 file is running, the master will still exit, requiring you to manually clean
 up the lock.  But if no matching process is found, the master will remove the
 apparently stale lock and make another attempt to claim the master lock."""))
+    parser.add_option('-r', '--runner',
+                      dest='runners', action='append', default=[],
+                      help=_("""\
+Override the default set of queue runners that the master watch will invoke
+instead of the default set.  Multiple -r options may be given.  The values for
+-r are passed straight through to bin/qrunner."""))
     parser.add_option('-C', '--config',
                       help=_('Alternative configuration file to use'))
     options, arguments = parser.parse_args()
@@ -258,14 +264,13 @@ class Loop:
             log.info('Master watcher caught SIGINT.  Restarting.')
         signal.signal(signal.SIGINT, sigint_handler)
 
-    def _start_runner(self, qrname, slice, count):
+    def _start_runner(self, spec):
         """Start a queue runner.
 
         All arguments are passed to the qrunner process.
 
-        :param qrname: The name of the queue runner.
-        :param slice: The slice number.
-        :param count: The total number of slices.
+        :param spec: A queue runner spec, in a format acceptable to
+            bin/qrunner's --runner argument, e.g. name:slice:count
         :return: The process id of the child queue runner.
         """
         pid = os.fork()
@@ -275,7 +280,7 @@ class Loop:
         # Child.
         #
         # Craft the command line arguments for the exec() call.
-        rswitch = '--runner=%s:%d:%d' % (qrname, slice, count)
+        rswitch = '--runner=' + spec
         # Wherever mailmanctl lives, so too must live the qrunner script.
         exe = os.path.join(config.BIN_DIR, 'qrunner')
         # config.PYTHON, which is the absolute path to the Python interpreter,
@@ -289,13 +294,29 @@ class Loop:
         # We should never get here.
         raise RuntimeError('os.execl() failed')
 
-    def start_qrunners(self):
-        """Start all the configured qrunners."""
-        for qrname, count in config.qrunners.items():
+    def start_qrunners(self, qrunners=None):
+        """Start all the configured qrunners.
+
+        :param qrunners: If given, a sequence of queue runner names to start.
+            If not given, this sequence is taken from the configuration file.
+        """
+        if not qrunners:
+            spec_parts = config.qrunners.items()
+        else:
+            spec_parts = []
+            for qrname in qrunners:
+                if '.' in qrname:
+                    spec_parts.append((qrname, 1))
+                else:
+                    spec_parts.append((config.qrunner_shortcuts[qrname], 1))
+        for qrname, count in spec_parts:
             for slice_number in range(count):
                 # qrunner name, slice #, # of slices, restart count
                 info = (qrname, slice_number, count, 0)
-                pid = self._start_runner(qrname, slice_number, count)
+                spec = '%s:%d:%d' % (qrname, slice_number, count)
+                pid = self._start_runner(spec)
+                log = logging.getLogger('mailman.qrunner')
+                log.debug('[%d] %s', pid, spec)
                 self._kids[pid] = info
 
     def loop(self):
@@ -397,7 +418,7 @@ def main():
         loop = Loop(lock, parser.options.restartable, parser.options.config)
         loop.install_signal_handlers()
         try:
-            loop.start_qrunners()
+            loop.start_qrunners(parser.options.runners)
             loop.loop()
         finally:
             loop.cleanup()
diff --git a/mailman/bin/qrunner.py b/mailman/bin/qrunner.py
index a0032582c..342fa4203 100644
--- a/mailman/bin/qrunner.py
+++ b/mailman/bin/qrunner.py
@@ -28,7 +28,6 @@ from mailman.initialize import initialize
 
 
 COMMASPACE = ', '
-QRUNNER_SHORTCUTS = {}
 log = None
 
 
@@ -123,8 +122,8 @@ def make_qrunner(name, slice, range, once=False):
     # could be one of the shortcut names.  Or the name is a full module path,
     # use it explicitly.  If the name starts with a dot, it's a class name
     # relative to the Mailman.queue package.
-    if name in QRUNNER_SHORTCUTS:
-        classpath = QRUNNER_SHORTCUTS[name]
+    if name in config.qrunner_shortcuts:
+        classpath = config.qrunner_shortcuts[name]
     elif name.startswith('.'):
         classpath = 'mailman.queue' + name
     else:
@@ -200,23 +199,14 @@ def main():
     # value of True to propagate, which allows the 'mailman' root logger to
     # see the log messages.
     initialize(opts.config, propagate_logs=not opts.subproc)
-    log = logging.getLogger('mailman.qrunner')
-
-    # Calculate the qrunner shortcut names.
-    for qrunnerpath, slices in config.qrunners.items():
-        classname = qrunnerpath.rsplit('.', 1)[1]
-        if classname.endswith('Runner'):
-            shortname = classname[:-6].lower()
-        else:
-            shortname = classname
-        QRUNNER_SHORTCUTS[shortname] = qrunnerpath
 
     if opts.list:
-        prefixlen = max(len(shortname) for shortname in QRUNNER_SHORTCUTS)
-        for shortname in sorted(QRUNNER_SHORTCUTS):
-            runnername = QRUNNER_SHORTCUTS[shortname]
+        prefixlen = max(len(shortname)
+                        for shortname in config.qrunner_shortcuts)
+        for shortname in sorted(config.qrunner_shortcuts):
+            runnername = config.qrunner_shortcuts[shortname]
             shortname = (' ' * (prefixlen - len(shortname))) + shortname
-            print _('$shortname runs the $runnername qrunner')
+            print _('$shortname runs $runnername')
         sys.exit(0)
 
     # Fast track for one infinite runner
@@ -233,6 +223,7 @@ def main():
         loop = Loop(qrunner)
         set_signals(loop)
         # Now start up the main loop
+        log = logging.getLogger('mailman.qrunner')
         log.info('%s qrunner started.', loop.name())
         qrunner.run()
         log.info('%s qrunner exiting.', loop.name())
diff --git a/mailman/bin/testall.py b/mailman/bin/testall.py
index 9d50be111..1037b28ef 100644
--- a/mailman/bin/testall.py
+++ b/mailman/bin/testall.py
@@ -28,9 +28,7 @@ import shutil
 import optparse
 import tempfile
 import unittest
-
-import mailman
-import mailman.tests
+import pkg_resources
 
 from mailman import Version
 from mailman.configuration import config
@@ -76,8 +74,12 @@ Reduce verbosity by 1 (but not below 0)."""))
     parser.add_option('-c', '--coverage',
                       default=False, action='store_true',
                       help=_('Enable code coverage.'))
-    opts, args = parser.parse_args()
-    return parser, opts, args
+    options, arguments = parser.parse_args()
+    if len(arguments) == 0:
+        arguments = ['.']
+    parser.options = options
+    parser.arguments = arguments
+    return parser
 
 
 
@@ -147,39 +149,57 @@ def suite(patterns=None):
 def main():
     global basedir
 
-    parser, opts, args = parseargs()
-    if not args:
-        args = ['.']
+    parser = parseargs()
 
-    # Store the options some place that other code can get to it.
-    config.opts = opts
+    # Set verbosity level for test_documentation.py.  XXX There should be a
+    # better way to do this.
+    config.verbosity = parser.options.verbosity
 
     # Turn on code coverage if selected.
-    if opts.coverage:
+    if parser.options.coverage:
         try:
             import coverage
         except ImportError:
-            opts.coverage = False
+            parser.options.coverage = False
         else:
             coverage.start()
 
     # Set up the testing configuration file both for this process, and for all
     # sub-processes testing will spawn (e.g. the qrunners).
     #
+    # Also create a logging.cfg file with values reflecting verbosity and
+    # stderr propagation.  Enable it only if necessary.
+    fd, logging_cfg = tempfile.mkstemp(suffix='.cfg')
+    os.close(fd)
+    enable_logging_cfg = False
+    with open(logging_cfg, 'w') as fp:
+        print >> fp, '[*]'
+        if parser.options.stderr:
+            print >> fp, 'propagate = True'
+            enable_logging_cfg = True
+        if parser.options.verbosity > 2:
+            print >> fp, 'level = DEBUG'
+            enable_logging_cfg = True
+
+    cfg_in = pkg_resources.resource_string(
+        'mailman.tests', 'testing.cfg.in')
+    fd, cfg_out = tempfile.mkstemp(suffix='.cfg')
+    os.close(fd)
+    with open(cfg_out, 'w') as fp:
+        fp.write(cfg_in)
+        if enable_logging_cfg:
+            print >> fp, 'LOG_CONFIG_FILE = "%s"' % logging_cfg
+
     # Calculate a temporary VAR_DIR directory so that run-time artifacts of
     # the tests won't tread on the installation's data.  This also makes it
     # easier to clean up after the tests are done, and insures isolation of
     # test suite runs.
-    cfg_in = os.path.join(os.path.dirname(mailman.tests.__file__),
-                           'testing.cfg.in')
-    fd, cfg_out = tempfile.mkstemp(suffix='.cfg')
-    os.close(fd)
-    shutil.copyfile(cfg_in, cfg_out)
-
     var_dir = tempfile.mkdtemp()
-    if opts.verbosity > 2:
+    if parser.options.verbosity > 2:
         print 'VAR_DIR :', var_dir
         print 'config file:', cfg_out
+        if enable_logging_cfg:
+            print 'logging config file:', logging_cfg
 
     user_id = os.getuid()
     user_name = pwd.getpwuid(user_id).pw_name
@@ -195,7 +215,7 @@ def main():
             print >> fp, 'MAILMAN_GID = %d' % group_id
             print >> fp, "LANGUAGES = 'en'"
 
-        initialize_1(cfg_out, propagate_logs=opts.stderr)
+        initialize_1(cfg_out, propagate_logs=parser.options.stderr)
         mailman_uid = pwd.getpwnam(config.MAILMAN_USER).pw_uid
         mailman_gid = grp.getgrnam(config.MAILMAN_GROUP).gr_gid
         os.chmod(cfg_out, 0660)
@@ -208,28 +228,31 @@ def main():
         os.chmod(config.dbfile, 0660)
         os.chown(config.dbfile, mailman_uid, mailman_gid)
 
-        # Patch ups
+        # Patch ups.
         test_engine_url = 'sqlite:///' + config.dbfile
-        config.SQLALCHEMY_ENGINE_URL = test_engine_url
+        config.DEFAULT_DATABASE_URL = test_engine_url
 
         # Write this to the config file so subprocesses share the same testing
         # database file.
         with open(cfg_out, 'a') as fp:
-            print >> fp, 'SQLALCHEMY_ENGINE_URL = "%s"' % test_engine_url
+            print >> fp, 'DEFAULT_DATABASE_URL = "%s"' % test_engine_url
 
         # With -vvv, turn on engine debugging.
-        initialize_2(opts.verbosity > 3)
+        initialize_2(parser.options.verbosity > 3)
 
-        # Run the tests
+        # Run the tests.  XXX I'm not sure if basedir can be converted to
+        # pkg_resources.
+        import mailman
         basedir = os.path.dirname(mailman.__file__)
-        runner = unittest.TextTestRunner(verbosity=opts.verbosity)
-        results = runner.run(suite(args))
+        runner = unittest.TextTestRunner(verbosity=parser.options.verbosity)
+        results = runner.run(suite(parser.arguments))
     finally:
         os.remove(cfg_out)
+        os.remove(logging_cfg)
         shutil.rmtree(var_dir)
 
     # Turn off coverage and print a report
-    if opts.coverage:
+    if parser.options.coverage:
         coverage.stop()
         modules = [module for name, module in sys.modules.items()
                    if module
diff --git a/mailman/configuration.py b/mailman/configuration.py
index 1f82c8304..83fadac5d 100644
--- a/mailman/configuration.py
+++ b/mailman/configuration.py
@@ -48,6 +48,7 @@ class Configuration(object):
         self.domains = {}       # email host -> web host
         self._reverse = None
         self.qrunners = {}
+        self.qrunner_shortcuts = {}
         self.QFILE_SCHEMA_VERSION = Version.QFILE_SCHEMA_VERSION
 
     def load(self, filename=None):
@@ -212,22 +213,34 @@ class Configuration(object):
     def add_qrunner(self, name, count=1):
         """Convenient interface for adding additional qrunners.
 
-        name is the qrunner name and it must not include the 'Runner' suffix.
-        E.g. 'HTTP' or 'LMTP'.  count is the number of qrunner slices to
-        create, by default, 1.
+        :param name: the qrunner name, which must not include the 'Runner'
+            suffix.  E.g. 'HTTP' or 'LMTP'.
+        :param count: is the number of qrunner slices to create, default: 1.
         """
         if name.startswith('.'):
             name = 'mailman.queue' + name
         self.qrunners[name] = count
+        # Calculate the queue runner shortcut name.
+        classname = name.rsplit('.', 1)[1]
+        if classname.endswith('Runner'):
+            shortname = classname[:-6].lower()
+        else:
+            shortname = classname
+        self.qrunner_shortcuts[shortname] = name
 
     def del_qrunner(self, name):
         """Remove the named qrunner so that it does not start.
 
-        name is the qrunner name and it must not include the 'Runner' suffix.
+        :param name: the qrunner name, which must not include the 'Runner'
+            suffix.
         """
         if name.startswith('.'):
             name = 'mailman.queue' + name
         self.qrunners.pop(name)
+        for shortname, classname in self.qrunner_shortcuts:
+            if name == classname:
+                del self.qrunner_shortcuts[shortname]
+                break
 
     @property
     def paths(self):
diff --git a/mailman/database/__init__.py b/mailman/database/__init__.py
index 67f803454..1615c291f 100644
--- a/mailman/database/__init__.py
+++ b/mailman/database/__init__.py
@@ -120,6 +120,7 @@ class StockDatabase:
 
     def _reset(self):
         from mailman.database.model import ModelMeta
+        self.store.rollback()
         ModelMeta._reset(self.store)
 
 
diff --git a/mailman/loginit.py b/mailman/loginit.py
index 2fc6d88ef..844c2543e 100644
--- a/mailman/loginit.py
+++ b/mailman/loginit.py
@@ -56,18 +56,45 @@ _handlers = []
 
 
 class ReallySafeConfigParser(ConfigParser.SafeConfigParser):
+    """Like `SafeConfigParser` but catches `NoOptionError`."""
+
     def getstring(self, section, option, default=None):
         try:
             return ConfigParser.SafeConfigParser.get(self, section, option)
         except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
-            return default
+            # Try again with the '*' section.
+            try:
+                return ConfigParser.SafeConfigParser.get(self, '*', option)
+            except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
+                return default
 
     def getboolean(self, section, option, default=None):
         try:
             return ConfigParser.SafeConfigParser.getboolean(
                 self, section, option)
         except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
-            return default
+            # Try again with the '*' section.
+            try:
+                return ConfigParser.SafeConfigParser.getboolean(
+                    self, '*', option)
+            except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
+                return default
+
+    def getlevel(self, section, option, default=None):
+        try:
+            level_str = ConfigParser.SafeConfigParser.get(
+                self, section, option)
+        except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
+            # Try again with the '*' section.
+            try:
+                level_str = ConfigParser.SafeConfigParser.get(
+                    self, '*', option)
+            except (ConfigParser.NoSectionError, ConfigParser.NoOptionError):
+                level_str = default
+        # Convert the level string into an integer.
+        level_str = level_str.upper()
+        level_def = (logging.DEBUG if section == 'debug' else logging.INFO)
+        return getattr(logging, level_str, level_def)
 
 
 
@@ -136,9 +163,7 @@ def initialize(propagate=False):
         # does not set an override, the default level will be INFO except for
         # the 'debug' logger.  It doesn't make much sense for the debug logger
         # to ignore debug level messages!
-        level_str = cp.getstring(logger, 'level', 'INFO').upper()
-        level_def = (logging.DEBUG if logger == 'debug' else logging.INFO)
-        level_int = getattr(logging, level_str, level_def)
+        level_int = cp.getlevel(logger, 'level', 'INFO')
         log.setLevel(level_int)
         # Create a formatter for this logger, then a handler, and link the
         # formatter to the handler.
diff --git a/mailman/queue/docs/lmtp.txt b/mailman/queue/docs/lmtp.txt
new file mode 100644
index 000000000..7c886c281
--- /dev/null
+++ b/mailman/queue/docs/lmtp.txt
@@ -0,0 +1,49 @@
+LTMP server
+===========
+
+Mailman can accept messages via LMTP (RFC 2033).  Most modern mail servers
+support LMTP local delivery, so this is a very portable way to connect Mailman
+with your mail server.
+
+Our LMTP server is fairly simple though; all it does is make sure that the
+message is destined for a valid endpoint, e.g. mylist-join@example.com.
+
+Let's start a testable LMTP queue runner.
+
+    >>> from mailman.tests import helpers
+    >>> master = helpers.TestableMaster()
+    >>> master.start('lmtp')
+
+It also helps to have a nice LMTP client.
+
+    >>> lmtp = helpers.get_lmtp_client()
+    (220, '... Python LMTP queue runner 1.0')
+    >>> lmtp.lhlo('remote.example.org')
+    (250, ...)
+
+
+Invalid mailing list
+--------------------
+
+If the mail server tries to send a message to a non-existant mailing list, it
+will get a 550 error.
+
+    >>> lmtp.sendmail(
+    ...     'anne.person@example.com',
+    ...     ['mylist@example.com'], """\
+    ... From: anne.person@example.com
+    ... To: mylist@example.com
+    ... Subject: An interesting message
+    ... Message-ID: <aardvark>
+    ...
+    ... This is an interesting message.
+    ... """)
+    Traceback (most recent call last):
+    ...
+    SMTPDataError: (550, 'Requested action not taken: mailbox unavailable')
+
+
+Clean up
+--------
+
+    >>> master.stop()
diff --git a/mailman/queue/lmtp.py b/mailman/queue/lmtp.py
index 71ae2b9cc..5001b3aac 100644
--- a/mailman/queue/lmtp.py
+++ b/mailman/queue/lmtp.py
@@ -19,17 +19,13 @@
 
 Most mail servers can be configured to deliver local messages via 'LMTP'[1].
 This module is actually an LMTP server rather than a standard queue runner.
-Once it enters its main asyncore loop, it does not respond to mailmanctl
-signals the same way as other runners do.  All signals will kill this process,
-but the normal mailmanctl watchdog will restart it upon exit.
 
 The LMTP runner opens a local TCP port and waits for the mail server to
 connect to it.  The messages it receives over LMTP are very minimally parsed
 for sanity and if they look okay, they are accepted and injected into
-Mailman's incoming queue for processing through the normal pipeline.  If they
-don't look good, or are destined for a bogus sub-queue address, they are
-rejected right away, hopefully so that the peer mail server can provide better
-diagnostics.
+Mailman's incoming queue for normal processing.  If they don't look good, or
+are destined for a bogus sub-queue address, they are rejected right away,
+hopefully so that the peer mail server can provide better diagnostics.
 
 [1] RFC 2033 Local Mail Transport Protocol
     http://www.faqs.org/rfcs/rfc2033.html
@@ -48,14 +44,15 @@ from email.utils import parseaddr
 
 from mailman.Message import Message
 from mailman.configuration import config
-from mailman.runner import Runner, Switchboard
+from mailman.queue import Runner, Switchboard
 
 elog = logging.getLogger('mailman.error')
 qlog = logging.getLogger('mailman.qrunner')
 
-# We only care about the listname and the subq as in listname@ or
+
+# We only care about the listname and the subqueue as in listname@ or
 # listname-request@
-subqnames = (
+SUBQUEUE_NAMES = (
     'bounces',  'confirm',  'join', '       leave',
     'owner',    'request',  'subscribe',    'unsubscribe',
     )
@@ -63,6 +60,7 @@ subqnames = (
 DASH    = '-'
 CRLF    = '\r\n'
 ERR_451 = '451 Requested action aborted: error in processing'
+ERR_501 = '501 Message has defects'
 ERR_502 = '502 Error: command HELO not implemented'
 ERR_550 = config.LMTP_ERR_550
 
@@ -71,12 +69,12 @@ smtpd.__version__ = 'Python LMTP queue runner 1.0'
 
 
 
-def getlistq(address):
+def split_recipient(address):
     localpart, domain = address.split('@', 1)
     localpart = localpart.split(config.VERP_DELIMITER, 1)[0]
-    l = localpart.split(DASH)
-    if l[-1] in subqnames:
-        listname = DASH.join(l[:-1])
+    parts = localpart.split(DASH)
+    if parts[-1] in SUBQUEUE_NAMES:
+        listname = DASH.join(parts[:-1])
         subq = l[-1]
     else:
         listname = localpart
@@ -85,16 +83,20 @@ def getlistq(address):
 
 
 
-class SMTPChannel(smtpd.SMTPChannel):
-    # Override smtpd.SMTPChannel don't can't change the class name so that we
-    # don't have to reverse engineer Python's name mangling scheme.
-    #
-    # LMTP greeting is LHLO and no HELO/EHLO
+class Channel(smtpd.SMTPChannel):
+    """An LMTP channel."""
+
+    def __init__(self, server, conn, addr):
+        smtpd.SMTPChannel.__init__(self, server, conn, addr)
+        # Stash this here since the subclass uses private attributes. :(
+        self._server = server
 
     def smtp_LHLO(self, arg):
+        """The LMTP greeting, used instead of HELO/EHLO."""
         smtpd.SMTPChannel.smtp_HELO(self, arg)
 
     def smtp_HELO(self, arg):
+        """HELO is not a valid LMTP command."""
         self.push(ERR_502)
 
 
@@ -110,7 +112,7 @@ class LMTPRunner(Runner, smtpd.SMTPServer):
 
     def handle_accept(self):
         conn, addr = self.accept()
-        channel = SMTPChannel(self, conn, addr)
+        channel = Channel(self, conn, addr)
 
     def process_message(self, peer, mailfrom, rcpttos, data):
         try:
@@ -118,10 +120,12 @@ class LMTPRunner(Runner, smtpd.SMTPServer):
             # since the set of mailing lists could have changed.  However, on
             # a big site this could be fairly expensive, so we may need to
             # cache this in some way.
-            listnames = set(config.list_manager.names)
-            # Parse the message data.  XXX Should we reject the message
-            # immediately if it has defects?  Usually only spam has defects.
+            listnames = set(config.db.list_manager.names)
+            # Parse the message data.  If there are any defects in the
+            # message, reject it right away; it's probably spam. 
             msg = email.message_from_string(data, Message)
+            if msg.defects:
+                return ERR_501
             msg['X-MailFrom'] = mailfrom
         except Exception, e:
             elog.error('%s', e)
@@ -135,7 +139,7 @@ class LMTPRunner(Runner, smtpd.SMTPServer):
         for to in rcpttos:
             try:
                 to = parseaddr(to)[1].lower()
-                listname, subq, domain = getlistq(to)
+                listname, subq, domain = split_recipient(to)
                 listname += '@' + domain
                 if listname not in listnames:
                     status.append(ERR_550)
@@ -180,12 +184,12 @@ class LMTPRunner(Runner, smtpd.SMTPServer):
         # response to the LMTP client.
         return CRLF.join(status)
 
-    def _cleanup(self):
-        pass
-
+    def run(self):
+        """See `IRunner`."""
+        asyncore.loop()
 
-server = LMTPRunner()
-qlog.info('LMTPRunner qrunner started.')
-asyncore.loop()
-# We'll never get here, but just in case...
-qlog.info('LMTPRunner qrunner exiting.')
+    def stop(self):
+        """See `IRunner`."""
+        asyncore.socket_map.clear()
+        asyncore.close_all()
+        self.close()
diff --git a/mailman/tests/helpers.py b/mailman/tests/helpers.py
index 4dc289458..71b0914cb 100644
--- a/mailman/tests/helpers.py
+++ b/mailman/tests/helpers.py
@@ -23,6 +23,7 @@ __metaclass__ = type
 __all__ = [
     'TestableMaster',
     'digest_mbox',
+    'get_lmtp_client',
     'get_queue_messages',
     'make_testable_runner',
     ]
@@ -31,6 +32,8 @@ __all__ = [
 import os
 import time
 import errno
+import signal
+import socket
 import mailbox
 import smtplib
 import tempfile
@@ -102,12 +105,27 @@ def digest_mbox(mlist):
 class TestableMaster(Master):
     """A testable master loop watcher."""
 
-    def __init__(self, event):
+    def __init__(self):
         super(TestableMaster, self).__init__(
             restartable=False, config_file=config.filename)
-        self._event = event
+        self.event = threading.Event()
+        self.thread = threading.Thread(target=self.loop)
         self._started_kids = None
 
+    def start(self, *qrunners):
+        """Start the master."""
+        self.start_qrunners(qrunners)
+        self.thread.start()
+        # Wait until all the children are definitely started, or timeout.
+        self.event.wait(2.0)
+
+    def stop(self):
+        """Stop the master by killing all the children."""
+        for pid in self.qrunner_pids:
+            os.kill(pid, signal.SIGTERM)
+        self.cleanup()
+        self.thread.join()
+
     def loop(self):
         """Wait until all the qrunners are actually running before looping."""
         starting_kids = set(self._kids)
@@ -125,7 +143,7 @@ class TestableMaster(Master):
         # testing environment, even after all have exited.
         self._started_kids = set(self._kids)
         # Let the blocking thread know everything's running.
-        self._event.set()
+        self.event.set()
         super(TestableMaster, self).loop()
 
     @property
@@ -173,9 +191,38 @@ class SMTPServer:
             else:
                 self._messages.append(message)
             yield message
-            
+
     def clear(self):
         """Clear all messages from the queue."""
         # Just throw these away.
         list(self._messages)
         self._messages = []
+
+
+
+class LMTP(smtplib.SMTP):
+    """Like a normal SMTP client, but for LMTP."""
+    def lhlo(self, name=''):
+        self.putcmd('lhlo', name or self.local_hostname)
+        code, msg = self.getreply()
+        self.helo_resp = msg
+        return code, msg
+
+
+def get_lmtp_client():
+    """Return a connected LMTP client."""
+    # It's possible the process has started but is not yet accepting
+    # connections.  Wait a little while.
+    lmtp = LMTP()
+    for attempts in range(3):
+        try:
+            response = lmtp.connect(config.LMTP_HOST, config.LMTP_PORT)
+            print response
+            return lmtp
+        except socket.error, error:
+            if error[0] == errno.ECONNREFUSED:
+                time.sleep(1)
+            else:
+                raise
+    else:
+        raise RuntimeError('Connection refused')
diff --git a/mailman/tests/test_documentation.py b/mailman/tests/test_documentation.py
index d55e2b4bb..d11d4bd70 100644
--- a/mailman/tests/test_documentation.py
+++ b/mailman/tests/test_documentation.py
@@ -88,7 +88,7 @@ def test_suite():
     flags = (doctest.ELLIPSIS |
              doctest.NORMALIZE_WHITESPACE |
              doctest.REPORT_NDIFF)
-    if config.opts.verbosity <= 2:
+    if config.verbosity <= 2:
         flags |= doctest.REPORT_ONLY_FIRST_FAILURE
     # Add all the doctests in all subpackages.
     for docsdir in packages:
diff --git a/mailman/tests/testing.cfg.in b/mailman/tests/testing.cfg.in
index 074806cb1..e7b23ac6d 100644
--- a/mailman/tests/testing.cfg.in
+++ b/mailman/tests/testing.cfg.in
@@ -7,6 +7,7 @@
 SMTPPORT                = 10825
 MAX_RESTARTS            = 1
 MTA                     = None
+USE_LMTP                = Yes
 
 add_domain('example.com', 'www.example.com')