src/mailman/testing/layers.py

   1 # Copyright (C) 2008-2023 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 <https://www.gnu.org/licenses/>.
  17
  18 """Mailman test layers."""
  19
  20 # XXX 2012-03-23 BAW: Layers really really suck.  For example, the
  21 # test_owners_get_email() test requires that both the SMTPLayer and LMTPLayer
  22 # be set up, but there's apparently no way to do that and make zope.testing
  23 # happy.  This causes no test failures, but it does cause errors at the end of
  24 # the full test run.  For now, I'll ignore that, but I do want to eventually
  25 # get rid of the layers and use something like testresources or some such.
  26
  27 import os
  28 import ssl
  29 import sys
  30 import shutil
  31 import logging
  32 import datetime
  33 import tempfile
  34
  35 from importlib.resources import files, read_text
  36 from lazr.config import as_boolean
  37 from mailman.config import config
  38 from mailman.core import initialize
  39 from mailman.core.initialize import INHIBIT_CONFIG_FILE
  40 from mailman.core.logging import get_handler
  41 from mailman.database.transaction import transaction
  42 from mailman.interfaces.domain import IDomainManager
  43 from mailman.testing.helpers import (
  44     get_lmtp_client,
  45     reset_the_world,
  46     TestableMaster,
  47     wait_for_webservice,
  48 )
  49 from mailman.testing.mta import (
  50     ConnectionCountingController,
  51     ConnectionCountingSSLController,
  52     ConnectionCountingSTARTTLSController,
  53 )
  54 from mailman.utilities.string import expand
  55 from public import public
  56 from textwrap import dedent
  57 from zope.component import getUtility
  58
  59
  60 TEST_TIMEOUT = datetime.timedelta(seconds=5)
  61 NL = '\n'
  62
  63
  64 @public
  65 class MockAndMonkeyLayer:
  66     """Layer for mocking and monkey patching for testing."""
  67
  68     # Set this to True to enable predictable datetimes, uids, etc.
  69     testing_mode = False
  70
  71     # A registration of all testing factories, for resetting between tests.
  72     _resets = []
  73
  74     @classmethod
  75     def testTearDown(cls):
  76         for reset in cls._resets:
  77             reset()
  78
  79     @classmethod
  80     def register_reset(cls, reset):
  81         cls._resets.append(reset)
  82
  83
  84 @public
  85 class ConfigLayer(MockAndMonkeyLayer):
  86     """Layer for pushing and popping test configurations."""
  87
  88     var_dir = None
  89     styles = None
  90
  91     @classmethod
  92     def setUp(cls):
  93         # Set up the basic configuration stuff.  Turn off path creation until
  94         # we've pushed the testing config.
  95         config.create_paths = False
  96         initialize.initialize_1(INHIBIT_CONFIG_FILE)
  97         assert cls.var_dir is None, 'Layer already set up'
  98         # Calculate a temporary VAR_DIR directory so that run-time artifacts
  99         # of the tests won't tread on the installation's data.  This also
 100         # makes it easier to clean up after the tests are done, and insures
 101         # isolation of test suite runs.
 102         cls.var_dir = tempfile.mkdtemp()
 103         # We need a test configuration both for the foreground process and any
 104         # child processes that get spawned.  lazr.config would allow us to do
 105         # it all in a string that gets pushed, and we'll do that for the
 106         # foreground, but because we may be spawning processes (such as
 107         # runners) we'll need a file that we can specify to the with the -C
 108         # option.  Craft the full test configuration string here, push it, and
 109         # also write it out to a temp file for -C.
 110         #
 111         # Create a dummy postfix.cfg file so that the test suite doesn't try
 112         # to run the actual postmap command, which may not exist anyway.
 113         postfix_cfg = os.path.join(cls.var_dir, 'postfix.cfg')
 114         with open(postfix_cfg, 'w') as fp:
 115             print(dedent("""
 116             [postfix]
 117             postmap_command: true
 118             transport_file_type: hash
 119             """), file=fp)
 120         test_config = dedent("""
 121         [mailman]
 122         layout: testing
 123         [paths.testing]
 124         var_dir: {}
 125         [devmode]
 126         testing: yes
 127         [mta]
 128         configuration: {}
 129         """.format(cls.var_dir, postfix_cfg))
 130         # Read the testing config and push it.
 131         test_config += read_text('mailman.testing', 'testing.cfg')
 132         config.create_paths = True
 133         config.push('test config', test_config)
 134         # Initialize everything else.
 135         initialize.initialize_2(testing=True)
 136         initialize.initialize_3()
 137         # When stderr debugging is enabled, subprocess root loggers should
 138         # also be more verbose.
 139         if cls.stderr:
 140             test_config += dedent("""
 141             [logging.root]
 142             level: debug
 143             """)
 144         # Enable log message propagation and reset the log paths so that the
 145         # doctests can check the output.
 146         for logger_config in config.logger_configs:
 147             sub_name = logger_config.name.split('.')[-1]
 148             if sub_name == 'root':
 149                 continue
 150             logger_name = 'mailman.' + sub_name
 151             log = logging.getLogger(logger_name)
 152             log.propagate = cls.stderr
 153             # Reopen the file to a new path that tests can get at.  Instead of
 154             # using the configuration file path though, use a path that's
 155             # specific to the logger so that tests can find expected output
 156             # more easily.
 157             path = os.path.join(config.LOG_DIR, sub_name)
 158             get_handler(sub_name).reopen(path)
 159             log.setLevel(logging.DEBUG)
 160             # If stderr debugging is enabled, make sure subprocesses are also
 161             # more verbose.  In general though, we still don't want SQLAlchemy
 162             # debugging because it's just too verbose.  Unfortunately, if you
 163             # do want that level of debugging you currently have to manually
 164             # modify this conditional.
 165             if sub_name != 'database':
 166                 test_config += expand(dedent("""
 167                 [logging.$name]
 168                 path: $path
 169                 $extra_verbose
 170                 """), None, dict(name=sub_name, path=path,
 171                                  extra_verbose=dedent("""
 172                                      propagate: yes
 173                                      level: debug
 174                                      """) if cls.stderr else ""))
 175         # The root logger will already have a handler, but it's not the right
 176         # handler.  Remove that and set our own.
 177         if cls.stderr:
 178             console = logging.StreamHandler(sys.stderr)
 179             formatter = logging.Formatter(config.logging.root.format,
 180                                           config.logging.root.datefmt)
 181             console.setFormatter(formatter)
 182             root = logging.getLogger()
 183             del root.handlers[:]
 184             root.addHandler(console)
 185         # Write the configuration file for subprocesses and set up the config
 186         # object to pass that properly on the -C option.
 187         config_file = os.path.join(cls.var_dir, 'test.cfg')
 188         with open(config_file, 'w') as fp:
 189             fp.write(test_config)
 190             print(file=fp)
 191         config.filename = config_file
 192
 193     @classmethod
 194     def tearDown(cls):
 195         assert cls.var_dir is not None, 'Layer not set up'
 196         reset_the_world()
 197         # Destroy the test database after the tests are done so that there is
 198         # no data in case the tests are rerun with a database layer like mysql
 199         # or postgresql which are not deleted in teardown.
 200         shutil.rmtree(cls.var_dir)
 201         # Prevent the bit of post-processing on the .pop() that creates
 202         # directories.  We're basically shutting down everything and we don't
 203         # need the directories created.  Plus, doing so leaves a var directory
 204         # turd in the source tree's top-level directory.  We do it this way
 205         # rather than shutil.rmtree'ing the resulting var directory because
 206         # it's possible the user created a valid such directory for
 207         # operational or test purposes.
 208         config.create_paths = False
 209         config.pop('test config')
 210         cls.var_dir = None
 211
 212     @classmethod
 213     def testSetUp(cls):
 214         # Add an example domain.
 215         with transaction():
 216             getUtility(IDomainManager).add('example.com', 'An example domain.')
 217
 218     @classmethod
 219     def testTearDown(cls):
 220         reset_the_world()
 221
 222     # Flag to indicate that loggers should propagate to the console.
 223     stderr = False
 224
 225     # The top of our source tree, for tests that care (e.g. hooks.txt).
 226     root_directory = None
 227
 228     @classmethod
 229     def set_root_directory(cls, directory):
 230         """Set the directory at the root of our source tree.
 231
 232         zc.recipe.testrunner runs from parts/test/working-directory, but
 233         that's actually changed over the life of the package.  Some tests
 234         care, e.g. because they need to find our built-out bin directory.
 235         Fortunately, buildout can give us this information.  See the
 236         `buildout.cfg` file for where this method is called.
 237         """
 238         cls.root_directory = directory
 239
 240
 241 @public
 242 class SMTPLayer(ConfigLayer):
 243     """Layer for starting, stopping, and accessing a test SMTP server."""
 244
 245     smtpd = None
 246
 247     @classmethod
 248     def setUp(cls):
 249         assert cls.smtpd is None, 'Layer already set up'
 250         host = config.mta.smtp_host
 251         port = int(config.mta.smtp_port)
 252         cls.smtpd = ConnectionCountingController(host, port)
 253         cls.smtpd.start()
 254
 255     @classmethod
 256     def tearDown(cls):
 257         assert cls.smtpd is not None, 'Layer not set up'
 258         cls.smtpd.clear()
 259         cls.smtpd.stop()
 260
 261     @classmethod
 262     def testSetUp(cls):
 263         # Make sure we don't call our superclass's testSetUp(), otherwise the
 264         # example.com domain will get added twice.
 265         pass
 266
 267     @classmethod
 268     def testTearDown(cls):
 269         cls.smtpd.reset()
 270         cls.smtpd.clear()
 271
 272
 273 @public
 274 class SMTPSLayer(ConfigLayer):
 275     """Layer for starting, stopping, and accessing a test SMTPS server."""
 276
 277     smtpd = None
 278
 279     @classmethod
 280     def setUp(cls):
 281         assert cls.smtpd is None, 'Layer already set up'
 282         # Use a different port than the SMTP layer, since that one might
 283         # still be in use.
 284         config.push('smtps', """
 285         [mta]
 286         smtp_port: 9465
 287         smtp_secure_mode: smtps
 288         """)
 289         test_cert_path = files('mailman.testing') / 'ssl_test_cert.crt'
 290         test_key_path = files('mailman.testing') / 'ssl_test_key.key'
 291
 292         client_context = ssl.create_default_context()
 293         client_context.load_verify_locations(cafile=test_cert_path)
 294
 295         server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
 296         server_context.load_cert_chain(test_cert_path, test_key_path)
 297
 298         host = config.mta.smtp_host
 299         port = int(config.mta.smtp_port)
 300
 301         cls.smtpd = ConnectionCountingSSLController(
 302             host, port,
 303             client_context=client_context,
 304             server_context=server_context)
 305         cls.smtpd.start()
 306
 307     @classmethod
 308     def testSetUp(cls):
 309         # Make sure we don't call our superclass's testSetUp(), otherwise the
 310         # example.com domain will get added twice.
 311         pass
 312
 313     @classmethod
 314     def testTearDown(cls):
 315         cls.smtpd.reset()
 316         cls.smtpd.clear()
 317
 318     @classmethod
 319     def tearDown(cls):
 320         assert cls.smtpd is not None, 'Layer not set up'
 321         cls.smtpd.clear()
 322         cls.smtpd.stop()
 323         config.pop('smtps')
 324
 325
 326 @public
 327 class STARTTLSLayer(ConfigLayer):
 328     """Layer for starting and stopping a test SMTP server with STARTTLS."""
 329
 330     smtpd = None
 331
 332     @classmethod
 333     def setUp(cls):
 334         assert cls.smtpd is None, 'Layer already set up'
 335         # Use a different port than the SMTP and SMTPS layers, since that one
 336         # might still be in use.
 337         config.push('starttls', """
 338         [mta]
 339         smtp_port: 9587
 340         smtp_secure_mode: starttls
 341         """)
 342         test_cert_path = files('mailman.testing') / 'ssl_test_cert.crt'
 343         test_key_path = files('mailman.testing') / 'ssl_test_key.key'
 344
 345         client_context = ssl.create_default_context()
 346         client_context.load_verify_locations(cafile=test_cert_path)
 347
 348         server_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
 349         server_context.load_cert_chain(test_cert_path, test_key_path)
 350
 351         host = config.mta.smtp_host
 352         port = int(config.mta.smtp_port)
 353
 354         cls.smtpd = ConnectionCountingSTARTTLSController(
 355             host, port,
 356             client_context=client_context,
 357             server_context=server_context)
 358         cls.smtpd.start()
 359
 360     @classmethod
 361     def testSetUp(cls):
 362         # Make sure we don't call our superclass's testSetUp(), otherwise the
 363         # example.com domain will get added twice.
 364         pass
 365
 366     @classmethod
 367     def testTearDown(cls):
 368         cls.smtpd.reset()
 369         cls.smtpd.clear()
 370
 371     @classmethod
 372     def tearDown(cls):
 373         assert cls.smtpd is not None, 'Layer not set up'
 374         cls.smtpd.clear()
 375         cls.smtpd.stop()
 376         config.pop('starttls')
 377
 378
 379 @public
 380 class LMTPLayer(ConfigLayer):
 381     """Layer for starting, stopping, and accessing a test LMTP server."""
 382
 383     lmtpd = None
 384
 385     @staticmethod
 386     def _wait_for_lmtp_server():
 387         get_lmtp_client(quiet=True)
 388
 389     @classmethod
 390     def setUp(cls):
 391         assert cls.lmtpd is None, 'Layer already set up'
 392         cls.lmtpd = TestableMaster(cls._wait_for_lmtp_server)
 393         cls.lmtpd.start('lmtp')
 394
 395     @classmethod
 396     def tearDown(cls):
 397         assert cls.lmtpd is not None, 'Layer not set up'
 398         cls.lmtpd.stop()
 399         cls.lmtpd = None
 400
 401     @classmethod
 402     def testSetUp(cls):
 403         # Make sure we don't call our superclass's testSetUp(), otherwise the
 404         # example.com domain will get added twice.
 405         pass
 406
 407
 408 @public
 409 class RESTLayer(SMTPLayer):
 410     """Layer for starting, stopping, and accessing the test REST layer."""
 411
 412     server = None
 413     stderr = True
 414
 415     @classmethod
 416     def setUp(cls):
 417         assert cls.server is None, 'Layer already set up'
 418         cls.server = TestableMaster(wait_for_webservice)
 419         cls.server.start('rest')
 420
 421     @classmethod
 422     def tearDown(cls):
 423         assert cls.server is not None, 'Layer not set up'
 424         cls.server.stop()
 425         cls.server = None
 426
 427
 428 @public
 429 def is_testing():
 430     """Return a 'testing' flag for use with the predictable factories.
 431
 432     :return: True when in testing mode.
 433     :rtype: bool
 434     """
 435     return (MockAndMonkeyLayer.testing_mode or
 436             as_boolean(config.devmode.testing))