src/mailman/testing/layers.py

   1 # Copyright (C) 2008-2009 by the Free Software Foundation, Inc.
   2 #
   3 # This file is part of GNU Mailman.
   4 #
   5 # GNU Mailman is free software: you can redistribute it and/or modify it under
   6 # the terms of the GNU General Public License as published by the Free
   7 # Software Foundation, either version 3 of the License, or (at your option)
   8 # any later version.
   9 #
  10 # GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
  11 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  12 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  13 # more details.
  14 #
  15 # You should have received a copy of the GNU General Public License along with
  16 # GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.
  17
  18 """Mailman test layers."""
  19
  20 from __future__ import absolute_import, unicode_literals
  21
  22 __metaclass__ = type
  23 __all__ = [
  24     'ConfigLayer',
  25     'MockAndMonkeyLayer',
  26     'RESTLayer',
  27     'SMTPLayer',
  28     ]
  29
  30
  31 import os
  32 import sys
  33 import shutil
  34 import logging
  35 import datetime
  36 import tempfile
  37
  38 from pkg_resources import resource_string
  39 from textwrap import dedent
  40 from urllib2 import urlopen, URLError
  41 from zope.component import getUtility
  42
  43 from mailman.config import config
  44 from mailman.core import initialize
  45 from mailman.core.logging import get_handler
  46 from mailman.i18n import _
  47 from mailman.interfaces.domain import IDomainManager
  48 from mailman.interfaces.messages import IMessageStore
  49 from mailman.testing.helpers import TestableMaster
  50 from mailman.testing.mta import ConnectionCountingController
  51 from mailman.utilities.datetime import factory
  52 from mailman.utilities.string import expand
  53
  54
  55 TEST_TIMEOUT = datetime.timedelta(seconds=5)
  56 NL = '\n'
  57
  58
  59 \f
  60 class MockAndMonkeyLayer:
  61     """Layer for mocking and monkey patching for testing."""
  62
  63     @classmethod
  64     def setUp(cls):
  65         factory.testing_mode = True
  66
  67     @classmethod
  68     def tearDown(cls):
  69         factory.testing_mode = False
  70
  71     @classmethod
  72     def testTearDown(cls):
  73         factory.reset()
  74
  75
  76 \f
  77 class ConfigLayer(MockAndMonkeyLayer):
  78     """Layer for pushing and popping test configurations."""
  79
  80     var_dir = None
  81     styles = None
  82
  83     @classmethod
  84     def setUp(cls):
  85         # Set up the basic configuration stuff.
  86         initialize.initialize_1()
  87         assert cls.var_dir is None, 'Layer already set up'
  88         # Calculate a temporary VAR_DIR directory so that run-time artifacts
  89         # of the tests won't tread on the installation's data.  This also
  90         # makes it easier to clean up after the tests are done, and insures
  91         # isolation of test suite runs.
  92         cls.var_dir = tempfile.mkdtemp()
  93         # We need a test configuration both for the foreground process and any
  94         # child processes that get spawned.  lazr.config would allow us to do
  95         # it all in a string that gets pushed, and we'll do that for the
  96         # foreground, but because we may be spawning processes (such as queue
  97         # runners) we'll need a file that we can specify to the with the -C
  98         # option.  Craft the full test configuration string here, push it, and
  99         # also write it out to a temp file for -C.
 100         test_config = dedent("""
 101         [mailman]
 102         var_dir: %s
 103         """ % cls.var_dir)
 104         # Read the testing config and push it.
 105         test_config += resource_string('mailman.testing', 'testing.cfg')
 106         config.push('test config', test_config)
 107         # Initialize everything else.
 108         initialize.initialize_2()
 109         initialize.initialize_3()
 110         # When stderr debugging is enabled, subprocess root loggers should
 111         # also be more verbose.
 112         if cls.stderr:
 113             test_config += dedent("""
 114             [logging.root]
 115             propagate: yes
 116             level: debug
 117             """)
 118         # Enable log message propagation and reset the log paths so that the
 119         # doctests can check the output.
 120         for logger_config in config.logger_configs:
 121             sub_name = logger_config.name.split('.')[-1]
 122             if sub_name == 'root':
 123                 continue
 124             logger_name = 'mailman.' + sub_name
 125             log = logging.getLogger(logger_name)
 126             log.propagate = True
 127             # Reopen the file to a new path that tests can get at.  Instead of
 128             # using the configuration file path though, use a path that's
 129             # specific to the logger so that tests can find expected output
 130             # more easily.
 131             path = os.path.join(config.LOG_DIR, sub_name)
 132             get_handler(sub_name).reopen(path)
 133             log.setLevel(logging.DEBUG)
 134             # If stderr debugging is enabled, make sure subprocesses are also
 135             # more verbose.
 136             if cls.stderr:
 137                 test_config += expand(dedent("""
 138                 [logging.$name]
 139                 propagate: yes
 140                 level: debug
 141                 """), dict(name=sub_name, path=path))
 142         # zope.testing sets up logging before we get to our own initialization
 143         # function.  This messes with the root logger, so explicitly set it to
 144         # go to stderr.
 145         if cls.stderr:
 146             console = logging.StreamHandler(sys.stderr)
 147             formatter = logging.Formatter(config.logging.root.format,
 148                                           config.logging.root.datefmt)
 149             console.setFormatter(formatter)
 150             logging.getLogger().addHandler(console)
 151         # Write the configuration file for subprocesses and set up the config
 152         # object to pass that properly on the -C option.
 153         config_file = os.path.join(cls.var_dir, 'test.cfg')
 154         with open(config_file, 'w') as fp:
 155             fp.write(test_config)
 156             print >> fp
 157         config.filename = config_file
 158
 159     @classmethod
 160     def tearDown(cls):
 161         assert cls.var_dir is not None, 'Layer not set up'
 162         config.pop('test config')
 163         shutil.rmtree(cls.var_dir)
 164         cls.var_dir = None
 165
 166     @classmethod
 167     def testSetUp(cls):
 168         # Add an example domain.
 169         IDomainManager(config).add(
 170             'example.com', 'An example domain.',
 171             'http://lists.example.com', 'postmaster@example.com')
 172         config.db.commit()
 173
 174     @classmethod
 175     def testTearDown(cls):
 176         # Reset the database between tests.
 177         config.db._reset()
 178         # Remove all residual queue files.
 179         for dirpath, dirnames, filenames in os.walk(config.QUEUE_DIR):
 180             for filename in filenames:
 181                 os.remove(os.path.join(dirpath, filename))
 182         # Clear out messages in the message store.
 183         message_store = getUtility(IMessageStore)
 184         for message in message_store.messages:
 185             message_store.delete_message(message['message-id'])
 186         config.db.commit()
 187         # Reset the global style manager.
 188         config.style_manager.populate()
 189
 190     # Flag to indicate that loggers should propagate to the console.
 191     stderr = False
 192
 193     @classmethod
 194     def handle_stderr(cls, *ignore):
 195         cls.stderr = True
 196
 197     @classmethod
 198     def hack_options_parser(cls):
 199         """Hack our way into the zc.testing framework.
 200
 201         Add our custom command line option parsing into zc.testing's.  We do
 202         the imports here so that if zc.testing isn't invoked, this stuff never
 203         gets in the way.  This is pretty fragile, depend on changes in the
 204         zc.testing package.  There should be a better way!
 205         """
 206         from zope.testing.testrunner.options import parser
 207         parser.add_option(str('-e'), str('--stderr'),
 208                           action='callback', callback=cls.handle_stderr,
 209                           help=_('Propagate log errors to stderr.'))
 210
 211
 212 \f
 213 class SMTPLayer(ConfigLayer):
 214     """Layer for starting, stopping, and accessing a test SMTP server."""
 215
 216     smtpd = None
 217
 218     @classmethod
 219     def setUp(cls):
 220         assert cls.smtpd is None, 'Layer already set up'
 221         host = config.mta.smtp_host
 222         port = int(config.mta.smtp_port)
 223         cls.smtpd = ConnectionCountingController(host, port)
 224         cls.smtpd.start()
 225
 226     @classmethod
 227     def tearDown(cls):
 228         assert cls.smtpd is not None, 'Layer not set up'
 229         cls.smtpd.clear()
 230         cls.smtpd.stop()
 231
 232     @classmethod
 233     def testSetUp(cls):
 234         pass
 235
 236     @classmethod
 237     def testTearDown(cls):
 238         pass
 239
 240
 241 \f
 242 class RESTLayer(SMTPLayer):
 243     """Layer for starting, stopping, and accessing the test REST layer."""
 244
 245     server = None
 246
 247     @staticmethod
 248     def _wait_for_rest_server():
 249         until = datetime.datetime.now() + TEST_TIMEOUT
 250         while datetime.datetime.now() < until:
 251             try:
 252                 fp = urlopen('http://localhost:8001/3.0/system')
 253             except URLError:
 254                 pass
 255             else:
 256                 fp.close()
 257                 break
 258         else:
 259             raise RuntimeError('REST server did not start up')
 260
 261     @classmethod
 262     def setUp(cls):
 263         assert cls.server is None, 'Layer already set up'
 264         cls.server = TestableMaster(cls._wait_for_rest_server)
 265         cls.server.start('rest')
 266
 267     @classmethod
 268     def tearDown(cls):
 269         assert cls.server is not None, 'Layer not set up'
 270         cls.server.stop()
 271         cls.server = None