test/test.py

   1 #!/usr/bin/env python3
   2
   3 """Test processing logic on known feeds.
   4 """
   5
   6 import difflib as _difflib
   7 import glob as _glob
   8 import io as _io
   9 import logging as _logging
  10 import os as _os
  11 import re as _re
  12
  13 import rss2email as _rss2email
  14 import rss2email.config as _rss2email_config
  15 import rss2email.feed as _rss2email_feed
  16
  17
  18 # Get a copy of the internal rss2email.CONFIG for copying
  19 _stringio = _io.StringIO()
  20 _rss2email_config.CONFIG.write(_stringio)
  21 BASE_CONFIG_STRING = _stringio.getvalue()
  22 del _stringio
  23
  24 MESSAGE_ID_REGEXP = _re.compile(
  25     '^Message-ID: <[^@]*@dev.null.invalid>$', _re.MULTILINE)
  26 USER_AGENT_REGEXP = _re.compile(
  27     '^User-Agent: rss2email/[0-9.]* \+\S*$', _re.MULTILINE)
  28 BOUNDARY_REGEXP = _re.compile('===============[^=]+==')
  29
  30
  31 class Send (list):
  32     def __call__(self, sender, message):
  33         self.append((sender, message))
  34
  35     def as_string(self):
  36         chunks = [
  37             'SENT BY: {}\n{}\n'.format(sender, message.as_string())
  38             for sender,message in self]
  39         return '\n'.join(chunks)
  40
  41
  42 def clean_result(text):
  43     """Cleanup dynamic portions of the generated email headers
  44
  45     >>> text = (
  46     ...      'Content-Type: multipart/digest;\\n'
  47     ...      '  boundary="===============7509425281347501533=="\\n'
  48     ...      'MIME-Version: 1.0\\n'
  49     ...      'Date: Tue, 23 Aug 2011 15:57:37 -0000\\n'
  50     ...      'Message-ID: <9dff03db-f5a7@dev.null.invalid>\\n'
  51     ...      'User-Agent: rss2email/3.5 +https://github.com/wking/rss2email\\n'
  52     ...      )
  53     >>> print(clean_result(text).rstrip())
  54     Content-Type: multipart/digest;
  55       boundary="===============...=="
  56     MIME-Version: 1.0
  57     Date: Tue, 23 Aug 2011 15:57:37 -0000
  58     Message-ID: <...@dev.null.invalid>
  59     User-Agent: rss2email/...
  60     """
  61     for regexp,replacement in [
  62             (MESSAGE_ID_REGEXP, 'Message-ID: <...@dev.null.invalid>'),
  63             (USER_AGENT_REGEXP, 'User-Agent: rss2email/...'),
  64             (BOUNDARY_REGEXP, '===============...=='),
  65             ]:
  66         text = regexp.sub(replacement, text)
  67     return text
  68
  69 def test(dirname=None, config_path=None, force=False):
  70     if dirname is None:
  71         dirname = _os.path.dirname(config_path)
  72     if config_path is None:
  73         _rss2email.LOG.info('testing {}'.format(dirname))
  74         for config_path in _glob.glob(_os.path.join(dirname, '*.config')):
  75             test(dirname=dirname, config_path=config_path, force=force)
  76         return
  77     feed_path = _glob.glob(_os.path.join(dirname, 'feed.*'))[0]
  78     _rss2email.LOG.info('testing {}'.format(config_path))
  79     config = _rss2email_config.Config()
  80     config.read_string(BASE_CONFIG_STRING)
  81     read_paths = config.read([config_path])
  82     feed = _rss2email_feed.Feed(name='test', url=feed_path, config=config)
  83     expected_path = config_path.replace('config', 'expected')
  84     with open(expected_path, 'r') as f:
  85         expected = clean_result(f.read())
  86     feed._send = Send()
  87     feed.run()
  88     generated = feed._send.as_string()
  89     if force:
  90         with open(expected_path, 'w') as f:
  91             f.write(generated)
  92     generated = clean_result(generated)
  93     if generated != expected:
  94         diff_lines = _difflib.unified_diff(
  95             expected.splitlines(), generated.splitlines(),
  96             'expected', 'generated', lineterm='')
  97         raise ValueError(
  98             'error processing {}\n{}'.format(
  99                 config_path,
 100                 '\n'.join(diff_lines)))
 101
 102
 103 if __name__ == '__main__':
 104     import argparse
 105
 106     parser = argparse.ArgumentParser(description=__doc__)
 107     parser.add_argument(
 108         '--force', action='store_const', const=True,
 109         help=(
 110             "write output files (useful for figuring out what's expected "
 111             'from a new feed).'))
 112     parser.add_argument(
 113         '-V', '--verbose', default=0, action='count',
 114         help='increment verbosity')
 115     parser.add_argument(
 116         'dir', nargs='*',
 117         help='select subdirs to test (tests all subdirs by default)')
 118
 119     args = parser.parse_args()
 120
 121     if args.verbose:
 122         _rss2email.LOG.setLevel(
 123             max(_logging.DEBUG, _logging.ERROR - 10 * args.verbose))
 124
 125     # no paths on the command line, find all subdirectories
 126     this_dir = _os.path.dirname(__file__)
 127     if not args.dir:
 128         for basename in _os.listdir(this_dir):
 129             path = _os.path.join(this_dir, basename)
 130             if _os.path.isdir(path):
 131                 args.dir.append(path)
 132
 133     # we need standardized URLs, so change to `this_dir` and adjust paths
 134     orig_dir = _os.getcwd()
 135     _os.chdir(this_dir)
 136
 137     # run tests
 138     for orig_path in args.dir:
 139         this_path = _os.path.relpath(orig_path, start=this_dir)
 140         if _os.path.isdir(this_path):
 141             test(dirname=this_path, force=args.force)
 142         else:
 143             test(config_path=this_path, force=args.force)