src/mailman/testing/helpers.py

   1 # Copyright (C) 2008-2015 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 """Various test helpers."""
  19
  20 __all__ = [
  21     'LogFileMark',
  22     'TestableMaster',
  23     'call_api',
  24     'chdir',
  25     'configuration',
  26     'digest_mbox',
  27     'event_subscribers',
  28     'get_lmtp_client',
  29     'get_nntp_server',
  30     'get_queue_messages',
  31     'make_digest_messages',
  32     'make_testable_runner',
  33     'reset_the_world',
  34     'specialized_message_from_string',
  35     'subscribe',
  36     'temporary_db',
  37     'wait_for_webservice',
  38     ]
  39
  40
  41 import os
  42 import json
  43 import time
  44 import uuid
  45 import errno
  46 import shutil
  47 import signal
  48 import socket
  49 import logging
  50 import smtplib
  51 import datetime
  52 import threading
  53
  54 from base64 import b64encode
  55 from contextlib import contextmanager
  56 from email import message_from_string
  57 from httplib2 import Http
  58 from lazr.config import as_timedelta
  59 from mailman.bin.master import Loop as Master
  60 from mailman.config import config
  61 from mailman.database.transaction import transaction
  62 from mailman.email.message import Message
  63 from mailman.interfaces.member import MemberRole
  64 from mailman.interfaces.messages import IMessageStore
  65 from mailman.interfaces.styles import IStyleManager
  66 from mailman.interfaces.usermanager import IUserManager
  67 from mailman.runners.digest import DigestRunner
  68 from mailman.utilities.mailbox import Mailbox
  69 from unittest import mock
  70 from urllib.error import HTTPError
  71 from urllib.parse import urlencode
  72 from zope import event
  73 from zope.component import getUtility
  74
  75
  76 NL = '\n'
  77
  78
  79 \f
  80 def make_testable_runner(runner_class, name=None, predicate=None):
  81     """Create a runner that runs until its queue is empty.
  82
  83     :param runner_class: The runner class.
  84     :type runner_class: class
  85     :param name: Optional queue name; if not given, it is calculated from the
  86         class name.
  87     :type name: string or None
  88     :param predicate: Optional alternative predicate for deciding when to stop
  89         the runner.  When None (the default) it stops when the queue is empty.
  90     :type predicate: callable that gets one argument, the queue runner.
  91     :return: A runner instance.
  92     """
  93     if name is None:
  94         assert runner_class.__name__.endswith('Runner'), (
  95             'Unparseable runner class name: %s' % runner_class.__name__)
  96         name = runner_class.__name__[:-6].lower()
  97
  98     class EmptyingRunner(runner_class):
  99         """Stop processing when the queue is empty."""
 100
 101         def __init__(self, *args, **kws):
 102             super(EmptyingRunner, self).__init__(*args, **kws)
 103             # We know it's an EmptyingRunner, so really we want to see the
 104             # super class in the log files.
 105             self.__class__.__name__ = runner_class.__name__
 106
 107         def _do_periodic(self):
 108             """Stop when the queue is empty."""
 109             if predicate is None:
 110                 self._stop = (len(self.switchboard.files) == 0)
 111             else:
 112                 self._stop = predicate(self)
 113
 114     return EmptyingRunner(name)
 115
 116
 117 \f
 118 class _Bag:
 119     def __init__(self, **kws):
 120         for key, value in kws.items():
 121             setattr(self, key, value)
 122
 123
 124 def get_queue_messages(queue_name, sort_on=None):
 125     """Return and clear all the messages in the given queue.
 126
 127     :param queue_name: A string naming a queue.
 128     :param sort_on: The message header to sort on.  If None (the default),
 129         no sorting is performed.
 130     :return: A list of 2-tuples where each item contains the message and
 131         message metadata.
 132     """
 133     queue = config.switchboards[queue_name]
 134     messages = []
 135     for filebase in queue.files:
 136         msg, msgdata = queue.dequeue(filebase)
 137         messages.append(_Bag(msg=msg, msgdata=msgdata))
 138         queue.finish(filebase)
 139     if sort_on is not None:
 140         messages.sort(key=lambda item: str(item.msg[sort_on]))
 141     return messages
 142
 143
 144 \f
 145 def digest_mbox(mlist):
 146     """The mailing list's pending digest as a mailbox.
 147
 148     :param mlist: The mailing list.
 149     :return: The mailing list's pending digest as a mailbox.
 150     """
 151     path = os.path.join(mlist.data_path, 'digest.mmdf')
 152     return Mailbox(path)
 153
 154
 155 \f
 156 # Remember, Master is mailman.bin.master.Loop.
 157 class TestableMaster(Master):
 158     """A testable master loop watcher."""
 159
 160     def __init__(self, start_check=None):
 161         """Create a testable master loop watcher.
 162
 163         :param start_check: Optional callable used to check whether everything
 164             is running as the test expects.  Called in `loop()` in the
 165             subthread before the event is set.  The callback should block
 166             until the pass condition is set.
 167         :type start_check: Callable taking no arguments, returning nothing.
 168         """
 169         super(TestableMaster, self).__init__(
 170             restartable=False, config_file=config.filename)
 171         self.start_check = start_check
 172         self.event = threading.Event()
 173         self.thread = threading.Thread(target=self.loop)
 174         self.thread.daemon = True
 175         self._started_kids = None
 176
 177     def _pause(self):
 178         """See `Master`."""
 179         # No-op this because the tests generally do not signal the master,
 180         # which would mean the signal.pause() never exits.
 181         pass
 182
 183     def start(self, *runners):
 184         """Start the master."""
 185         self.start_runners(runners)
 186         self.thread.start()
 187         # Wait until all the children are definitely started.
 188         self.event.wait()
 189
 190     def stop(self):
 191         """Stop the master by killing all the children."""
 192         for pid in self.runner_pids:
 193             os.kill(pid, signal.SIGTERM)
 194         self.cleanup()
 195         self.thread.join()
 196
 197     def loop(self):
 198         """Wait until all the runners are actually running before looping."""
 199         starting_kids = set(self._kids)
 200         while starting_kids:
 201             for pid in self._kids:
 202                 try:
 203                     os.kill(pid, 0)
 204                     starting_kids.remove(pid)
 205                 except OSError as error:
 206                     if error.errno == errno.ESRCH:
 207                         # The child has not yet started.
 208                         pass
 209                     raise
 210         # Keeping a copy of all the started child processes for use by the
 211         # testing environment, even after all have exited.
 212         self._started_kids = set(self._kids)
 213         # If there are extra conditions to check, do it now.
 214         if self.start_check is not None:
 215             self.start_check()
 216         # Let the blocking thread know everything's running.
 217         self.event.set()
 218         super(TestableMaster, self).loop()
 219
 220     @property
 221     def runner_pids(self):
 222         """The pids of all the child runner processes."""
 223         for pid in self._started_kids:
 224             yield pid
 225
 226
 227 \f
 228 class LMTP(smtplib.SMTP):
 229     """Like a normal SMTP client, but for LMTP."""
 230     def lhlo(self, name=''):
 231         self.putcmd('lhlo', name or self.local_hostname)
 232         code, msg = self.getreply()
 233         self.helo_resp = msg
 234         return code, msg
 235
 236
 237 def get_lmtp_client(quiet=False):
 238     """Return a connected LMTP client."""
 239     # It's possible the process has started but is not yet accepting
 240     # connections.  Wait a little while.
 241     lmtp = LMTP()
 242     #lmtp.debuglevel = 1
 243     until = datetime.datetime.now() + as_timedelta(config.devmode.wait)
 244     while datetime.datetime.now() < until:
 245         try:
 246             response = lmtp.connect(
 247                 config.mta.lmtp_host, int(config.mta.lmtp_port))
 248             if not quiet:
 249                 print(response)
 250             return lmtp
 251         except IOError as error:
 252             if error.errno == errno.ECONNREFUSED:
 253                 time.sleep(0.1)
 254             else:
 255                 raise
 256     else:
 257         raise RuntimeError('Connection refused')
 258
 259
 260 \f
 261 def get_nntp_server(cleanups):
 262     """Create and start an NNTP server mock.
 263
 264     This can be used to retrieve the posted message for verification.
 265     """
 266     patcher = mock.patch('nntplib.NNTP')
 267     server_class = patcher.start()
 268     cleanups.append(patcher.stop)
 269     nntpd = server_class()
 270     # A class for more convenient access to the posted message.
 271     class NNTPProxy:
 272         def get_message(self):
 273             args = nntpd.post.call_args
 274             return specialized_message_from_string(args[0][0].read())
 275     return NNTPProxy()
 276
 277
 278 \f
 279 def wait_for_webservice():
 280     """Wait for the REST server to start serving requests."""
 281     until = datetime.datetime.now() + as_timedelta(config.devmode.wait)
 282     while datetime.datetime.now() < until:
 283         try:
 284             socket.socket().connect((config.webservice.hostname,
 285                                     int(config.webservice.port)))
 286         except IOError as error:
 287             if error.errno == errno.ECONNREFUSED:
 288                 time.sleep(0.1)
 289             else:
 290                 raise
 291         else:
 292             break
 293     else:
 294         raise RuntimeError('Connection refused')
 295
 296
 297 def call_api(url, data=None, method=None, username=None, password=None):
 298     """'Call a URL with a given HTTP method and return the resulting object.
 299
 300     The object will have been JSON decoded.
 301
 302     :param url: The url to open, read, and print.
 303     :type url: string
 304     :param data: Data to use to POST to a URL.
 305     :type data: dict
 306     :param method: Alternative HTTP method to use.
 307     :type method: str
 308     :param username: The HTTP Basic Auth user name.  None means use the value
 309         from the configuration.
 310     :type username: str
 311     :param password: The HTTP Basic Auth password.  None means use the value
 312         from the configuration.
 313     :type username: str
 314     :return: A 2-tuple containing the JSON decoded content (if there is any,
 315         else None) and the response object.
 316     :rtype: 2-tuple of (dict, response)
 317     :raises HTTPError: when a non-2xx return code is received.
 318     """
 319     headers = {}
 320     if data is not None:
 321         data = urlencode(data, doseq=True)
 322         headers['Content-Type'] = 'application/x-www-form-urlencoded'
 323     if method is None:
 324         if data is None:
 325             method = 'GET'
 326         else:
 327             method = 'POST'
 328     method = method.upper()
 329     if method in ('POST', 'PUT', 'PATCH') and data is None:
 330         data = urlencode({}, doseq=True)
 331     basic_auth = '{0}:{1}'.format(
 332         (config.webservice.admin_user if username is None else username),
 333         (config.webservice.admin_pass if password is None else password))
 334     # b64encode() requires a bytes, but the header value must be str.  Do the
 335     # necessary conversion dances.
 336     token = b64encode(basic_auth.encode('utf-8')).decode('ascii')
 337     headers['Authorization'] = 'Basic ' + token
 338     response, content = Http().request(url, method, data, headers)
 339     # If we did not get a 2xx status code, make this look like a urllib2
 340     # exception, for backward compatibility with existing doctests.
 341     if response.status // 100 != 2:
 342         raise HTTPError(url, response.status, content, response, None)
 343     if len(content) == 0:
 344         return None, response
 345     # XXX Workaround http://bugs.python.org/issue10038
 346     content = content.decode('utf-8')
 347     return json.loads(content), response
 348
 349
 350 \f
 351 @contextmanager
 352 def event_subscribers(*subscribers):
 353     """Temporarily extend the Zope event subscribers list.
 354
 355     :param subscribers: A sequence of event subscribers.
 356     :type subscribers: sequence of callables, each receiving one argument, the
 357         event.
 358     """
 359     old_subscribers = event.subscribers[:]
 360     event.subscribers.extend(subscribers)
 361     try:
 362         yield
 363     finally:
 364         event.subscribers[:] = old_subscribers
 365
 366
 367 \f
 368 class configuration:
 369     """A decorator/context manager for temporarily setting configurations."""
 370
 371     def __init__(self, section, **kws):
 372         self._section = section
 373         # Most tests don't care about the name given to the temporary
 374         # configuration.  Usually we'll just craft a random one, but some
 375         # tests do care, so give them a hook to set it.
 376         if '_configname' in kws:
 377             self._uuid = kws.pop('_configname')
 378         else:
 379             self._uuid = uuid.uuid4().hex
 380         self._values = kws.copy()
 381
 382     def _apply(self):
 383         lines = ['[{0}]'.format(self._section)]
 384         for key, value in self._values.items():
 385             lines.append('{0}: {1}'.format(key, value))
 386         config.push(self._uuid, NL.join(lines))
 387
 388     def _remove(self):
 389         config.pop(self._uuid)
 390
 391     def __enter__(self):
 392         self._apply()
 393
 394     def __exit__(self, *exc_info):
 395         self._remove()
 396         # Do not suppress exceptions.
 397         return False
 398
 399     def __call__(self, func):
 400         def wrapper(*args, **kws):
 401             self._apply()
 402             try:
 403                 return func(*args, **kws)
 404             finally:
 405                 self._remove()
 406         return wrapper
 407
 408
 409 \f
 410 @contextmanager
 411 def temporary_db(db):
 412     real_db = config.db
 413     config.db = db
 414     try:
 415         yield
 416     finally:
 417         config.db = real_db
 418
 419
 420 \f
 421 class chdir:
 422     """A context manager for temporary directory changing."""
 423     def __init__(self, directory):
 424         self._curdir = None
 425         self._directory = directory
 426
 427     def __enter__(self):
 428         self._curdir = os.getcwd()
 429         os.chdir(self._directory)
 430
 431     def __exit__(self, exc_type, exc_val, exc_tb):
 432         os.chdir(self._curdir)
 433         # Don't suppress exceptions.
 434         return False
 435
 436
 437 \f
 438 def subscribe(mlist, first_name, role=MemberRole.member, email=None):
 439     """Helper for subscribing a sample person to a mailing list.
 440
 441     Returns the newly created member object.
 442     """
 443     user_manager = getUtility(IUserManager)
 444     email = ('{0}person@example.com'.format(first_name[0].lower())
 445              if email is None else email)
 446     full_name = '{0} Person'.format(first_name)
 447     with transaction():
 448         person = user_manager.get_user(email)
 449         if person is None:
 450             address = user_manager.get_address(email)
 451             if address is None:
 452                 person = user_manager.create_user(email, full_name)
 453                 subscription_address = list(person.addresses)[0]
 454             else:
 455                 subscription_address = address
 456         else:
 457             subscription_address = list(person.addresses)[0]
 458         mlist.subscribe(subscription_address, role)
 459         roster = mlist.get_roster(role)
 460         return roster.get_member(email)
 461
 462
 463 \f
 464 def reset_the_world():
 465     """Reset everything:
 466
 467     * Clear out the database
 468     * Remove all residual queue and digest files
 469     * Clear the message store
 470     * Reset the global style manager
 471
 472     This should be as thorough a reset of the system as necessary to keep
 473     tests isolated.
 474     """
 475     # Reset the database between tests.
 476     config.db._reset()
 477     # Remove any digest files and members.txt file (for the file-recips
 478     # handler) in the lists' data directories.
 479     for dirpath, dirnames, filenames in os.walk(config.LIST_DATA_DIR):
 480         for filename in filenames:
 481             if filename.endswith('.mmdf') or filename == 'members.txt':
 482                 os.remove(os.path.join(dirpath, filename))
 483     # Remove all residual queue files.
 484     for dirpath, dirnames, filenames in os.walk(config.QUEUE_DIR):
 485         for filename in filenames:
 486             os.remove(os.path.join(dirpath, filename))
 487     # Clear out messages in the message store.
 488     message_store = getUtility(IMessageStore)
 489     with transaction():
 490         for message in message_store.messages:
 491             message_store.delete_message(message['message-id'])
 492     # Delete any other residual messages.
 493     for dirpath, dirnames, filenames in os.walk(config.MESSAGES_DIR):
 494         for filename in filenames:
 495             os.remove(os.path.join(dirpath, filename))
 496         shutil.rmtree(dirpath)
 497     # Reset the global style manager.
 498     getUtility(IStyleManager).populate()
 499     # Remove all dynamic header-match rules.
 500     config.chains['header-match'].flush()
 501
 502
 503 \f
 504 def specialized_message_from_string(unicode_text):
 505     """Parse text into a message object.
 506
 507     This is specialized in the sense that an instance of Mailman's own Message
 508     object is returned, and this message object has an attribute
 509     `original_size` which is the pre-calculated size in bytes of the message's
 510     text representation.
 511
 512     Also, the text must be ASCII-only unicode.
 513     """
 514     # This mimic what Switchboard.dequeue() does when parsing a message from
 515     # text into a Message instance.
 516     original_size = len(unicode_text)
 517     message = message_from_string(unicode_text, Message)
 518     message.original_size = original_size
 519     return message
 520
 521
 522 \f
 523 class LogFileMark:
 524     def __init__(self, log_name):
 525         self._log = logging.getLogger(log_name)
 526         self._filename = self._log.handlers[0].filename
 527         self._filepos = os.stat(self._filename).st_size
 528
 529     def readline(self):
 530         with open(self._filename) as fp:
 531             fp.seek(self._filepos)
 532             return fp.readline()
 533
 534     def read(self):
 535         with open(self._filename) as fp:
 536             fp.seek(self._filepos)
 537             return fp.read()
 538
 539
 540 \f
 541 def make_digest_messages(mlist, msg=None):
 542     if msg is None:
 543         msg = specialized_message_from_string("""\
 544 From: anne@example.org
 545 To: {listname}
 546 Message-ID: <testing>
 547
 548 message triggering a digest
 549 """.format(listname=mlist.fqdn_listname))
 550     mbox_path = os.path.join(mlist.data_path, 'digest.mmdf')
 551     config.handlers['to-digest'].process(mlist, msg, {})
 552     config.switchboards['digest'].enqueue(
 553         msg,
 554         listname=mlist.fqdn_listname,
 555         digest_path=mbox_path,
 556         volume=1, digest_number=1)
 557     runner = make_testable_runner(DigestRunner, 'digest')
 558     runner.run()