src/mailman/testing/documentation.py

   1 # Copyright (C) 2007-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 """Harness for testing Mailman's documentation.
  19
  20 Note that doctest extraction does not currently work for zip file
  21 distributions.  doctest discovery currently requires file system traversal.
  22 """
  23
  24 import os
  25 import sys
  26
  27 from click.testing import CliRunner
  28 from contextlib import ExitStack
  29 from importlib import import_module
  30 from mailman.testing.helpers import call_api
  31 from mailman.testing.layers import SMTPLayer
  32 from public import public
  33 from subprocess import PIPE, run, STDOUT
  34 from urllib.error import HTTPError
  35
  36
  37 DOT = '.'
  38 COMMASPACE = ', '
  39
  40
  41 def stop():
  42     """Call into pdb.set_trace()"""
  43     # Do the import here so that you get the wacky special hacked pdb instead
  44     # of Python's normal pdb.
  45     import pdb
  46     pdb.set_trace()
  47
  48
  49 def dump_msgdata(msgdata, *additional_skips):
  50     """Dump in a more readable way a message metadata dictionary."""
  51     if len(msgdata) == 0:
  52         print('*Empty*')
  53         return
  54     skips = set(additional_skips)
  55     # Some stuff we always want to skip, because their values will always be
  56     # variable data.
  57     skips.add('received_time')
  58     longest = max(len(key) for key in msgdata if key not in skips)
  59     for key in sorted(msgdata):
  60         if key in skips:
  61             continue
  62         print('{0:{2}}: {1}'.format(key, msgdata[key], longest))
  63
  64
  65 def dump_list(list_of_things, key=str):
  66     """Print items in a string to get rid of stupid u'' prefixes."""
  67     # List of things may be a generator.
  68     list_of_things = list(list_of_things)
  69     if len(list_of_things) == 0:
  70         print('*Empty*')
  71     if key is not None:
  72         list_of_things = sorted(list_of_things, key=key)
  73     for item in list_of_things:
  74         print(item)
  75
  76
  77 def call_http(url, data=None, method=None, username=None, password=None):
  78     """'Call a URL with a given HTTP method and return the resulting object.
  79
  80     The object will have been JSON decoded.
  81
  82     :param url: The url to open, read, and print.
  83     :type url: string
  84     :param data: Data to use to POST to a URL.
  85     :type data: dict
  86     :param method: Alternative HTTP method to use.
  87     :type method: str
  88     :param username: The HTTP Basic Auth user name.  None means use the value
  89         from the configuration.
  90     :type username: str
  91     :param password: The HTTP Basic Auth password.  None means use the value
  92         from the configuration.
  93     :type username: str
  94     :return: The decoded JSON data structure.
  95     :raises HTTPError: when a non-2xx return code is received.
  96     """
  97     try:
  98         content, response = call_api(url, data, method, username, password)
  99     except HTTPError as ex:
 100         print(ex)
 101         return
 102     if content is None:
 103         # We used to use httplib2 here, which included the status code in the
 104         # response headers in the `status` key.  requests doesn't do this, but
 105         # the doctests expect it so for backward compatibility, include the
 106         # status code in the printed response.
 107         headers = dict(status=response.status_code)
 108         headers.update({
 109             field.lower(): response.headers[field]
 110             for field in response.headers
 111             })
 112         # Remove the connection: close header from the response.
 113         headers.pop('connection')
 114         for field in sorted(headers):
 115             print('{}: {}'.format(field, headers[field]))
 116         return None
 117     return content
 118
 119
 120 def _print_dict(data, depth=0):
 121     for item, value in sorted(data.items()):
 122         if isinstance(value, dict):
 123             _print_dict(value, depth+1)
 124         print('    ' * depth + '{}: {}'.format(item, value))
 125
 126
 127 def dump_json(url, data=None, method=None, username=None, password=None,
 128               sort_entries=None):
 129     """Print the JSON dictionary read from a URL.
 130
 131     :param url: The url to open, read, and print.
 132     :type url: string
 133     :param data: Data to use to POST to a URL.
 134     :type data: dict
 135     :param method: Alternative HTTP method to use.
 136     :type method: str
 137     :param username: The HTTP Basic Auth user name.  None means use the value
 138         from the configuration.
 139     :type username: str
 140     :param password: The HTTP Basic Auth password.  None means use the value
 141         from the configuration.
 142     :type password: str
 143     :param sort_entries: The key to sort entiries.
 144     :type sort_entries: str
 145     """
 146     results = call_http(url, data, method, username, password)
 147     if results is None:
 148         return
 149     for key in sorted(results):
 150         value = results[key]
 151         if key == 'entries':
 152             entries = value
 153             if sort_entries:
 154                 entries = sorted(entries, key=lambda x: x[sort_entries])
 155             for i, entry in enumerate(entries):
 156                 # entry is a dictionary.
 157                 print('entry %d:' % i)
 158                 for entry_key in sorted(entry):
 159                     print('    {}: {}'.format(entry_key, entry[entry_key]))
 160         elif isinstance(value, list):
 161             printable_value = COMMASPACE.join(
 162                 "'{}'".format(s) for s in sorted(value))
 163             print('{}: [{}]'.format(key, printable_value))
 164         elif isinstance(value, dict):
 165             print('{}:'.format(key))
 166             _print_dict(value, 1)
 167         else:
 168             print('{}: {}'.format(key, value))
 169
 170
 171 @public
 172 def cli(command_path):
 173     """Call a CLI command in doctests.
 174
 175     Use this to invoke click commands in doctests.  This returns a partial
 176     that accepts a sequence of command line options, invokes the click
 177     command, and returns the results (unless the keyword argument 'quiet')
 178     is True.
 179     """
 180     package_path, dot, name = command_path.rpartition('.')
 181     command = getattr(import_module(package_path), name)
 182     def inner(command_string, quiet=False, input=None):           # noqa: E306
 183         args = command_string.split()
 184         assert args[0] == 'mailman', args
 185         assert args[1] == command.name, args
 186         # The first two will be `mailman <command>`.  That's just for
 187         # documentation purposes, and aren't useful for the test.
 188         result = CliRunner().invoke(command, args[2:], input=input)
 189         if not quiet:
 190             # Print the output, with any trailing newlines stripped, unless
 191             # the quiet flag is set.  The extra newlines just make the
 192             # doctests uglier and usually all we care about is the stdout
 193             # text.
 194             print(result.output.rstrip('\n'))
 195     return inner
 196
 197
 198 @public
 199 def run_mailman(args, **overrides):
 200     """Execute `mailman` command with the given arguments and return output."""
 201     exe = os.path.join(os.path.dirname(sys.executable), 'mailman')
 202     env = os.environ.copy()
 203     env.update(overrides)
 204     run_args = [exe]
 205     # When running tests as root, just add the flag to force run mailman
 206     # command without errors.
 207     if os.geteuid() == 0:
 208         run_args.append('--run-as-root')
 209     run_args.extend(args)
 210     proc = run(
 211         run_args, env=env, stdout=PIPE, stderr=STDOUT, universal_newlines=True)
 212     return proc
 213
 214
 215 @public
 216 def setup(testobj):
 217     """Test setup."""
 218     # In general, I don't like adding convenience functions, since I think
 219     # doctests should do the imports themselves.  It makes for better
 220     # documentation that way.  However, a few are really useful, or help to
 221     # hide some icky test implementation details.
 222     testobj.globs['smtpd'] = SMTPLayer.smtpd
 223     testobj.globs['stop'] = stop
 224     # Add this so that cleanups can be automatically added by the doctest.
 225     testobj.globs['cleanups'] = ExitStack()
 226
 227
 228 @public
 229 def teardown(testobj):
 230     testobj.globs['cleanups'].close()