git.notmuchmail.org Git - notmuch/blob - bindings/python-cffi/tests/conftest.py

   1 import email.message
   2 import mailbox
   3 import pathlib
   4 import shutil
   5 import socket
   6 import subprocess
   7 import textwrap
   8 import time
   9 import os
  10
  11 import pytest
  12
  13
  14 def pytest_report_header():
  15     which = shutil.which('notmuch')
  16     vers = subprocess.run(['notmuch', '--version'], stdout=subprocess.PIPE)
  17     return ['{} ({})'.format(vers.stdout.decode(errors='replace').strip(),which)]
  18
  19
  20 @pytest.fixture(scope='function')
  21 def tmppath(tmpdir):
  22     """The tmpdir fixture wrapped in pathlib.Path."""
  23     return pathlib.Path(str(tmpdir))
  24
  25
  26 @pytest.fixture
  27 def notmuch(maildir):
  28     """Return a function which runs notmuch commands on our test maildir.
  29
  30     This uses the notmuch-config file created by the ``maildir``
  31     fixture.
  32     """
  33     def run(*args):
  34         """Run a notmuch command.
  35
  36         This function runs with a timeout error as many notmuch
  37         commands may block if multiple processes are trying to open
  38         the database in write-mode.  It is all too easy to
  39         accidentally do this in the unittests.
  40         """
  41         cfg_fname = maildir.path / 'notmuch-config'
  42         cmd = ['notmuch'] + list(args)
  43         env = os.environ.copy()
  44         env['NOTMUCH_CONFIG'] = str(cfg_fname)
  45         proc = subprocess.run(cmd,
  46                               timeout=5,
  47                               env=env)
  48         proc.check_returncode()
  49     return run
  50
  51
  52 @pytest.fixture
  53 def maildir(tmppath):
  54     """A basic test interface to a valid maildir directory.
  55
  56     This creates a valid maildir and provides a simple mechanism to
  57     deliver test emails to it.  It also writes a notmuch-config file
  58     in the top of the maildir.
  59     """
  60     cur = tmppath / 'cur'
  61     cur.mkdir()
  62     new = tmppath / 'new'
  63     new.mkdir()
  64     tmp = tmppath / 'tmp'
  65     tmp.mkdir()
  66     cfg_fname = tmppath/'notmuch-config'
  67     with cfg_fname.open('w') as fp:
  68         fp.write(textwrap.dedent("""\
  69             [database]
  70             path={tmppath!s}
  71             [user]
  72             name=Some Hacker
  73             primary_email=dst@example.com
  74             [new]
  75             tags=unread;inbox;
  76             ignore=
  77             [search]
  78             exclude_tags=deleted;spam;
  79             [maildir]
  80             synchronize_flags=true
  81             """.format(tmppath=tmppath)))
  82     return MailDir(tmppath)
  83
  84
  85 class MailDir:
  86     """An interface around a correct maildir."""
  87
  88     def __init__(self, path):
  89         self._path = pathlib.Path(path)
  90         self.mailbox = mailbox.Maildir(str(path))
  91         self._idcount = 0
  92
  93     @property
  94     def path(self):
  95         """The pathname of the maildir."""
  96         return self._path
  97
  98     def _next_msgid(self):
  99         """Return a new unique message ID."""
 100         msgid = '{}@{}'.format(self._idcount, socket.getfqdn())
 101         self._idcount += 1
 102         return msgid
 103
 104     def deliver(self,
 105                 subject='Test mail',
 106                 body='This is a test mail',
 107                 to='dst@example.com',
 108                 frm='src@example.com',
 109                 headers=None,
 110                 new=False,      # Move to new dir or cur dir?
 111                 keywords=None,  # List of keywords or labels
 112                 seen=False,     # Seen flag (cur dir only)
 113                 replied=False,  # Replied flag (cur dir only)
 114                 flagged=False):  # Flagged flag (cur dir only)
 115         """Deliver a new mail message in the mbox.
 116
 117         This does only adds the message to maildir, does not insert it
 118         into the notmuch database.
 119
 120         :returns: A tuple of (msgid, pathname).
 121         """
 122         msgid = self._next_msgid()
 123         when = time.time()
 124         msg = email.message.EmailMessage()
 125         msg.add_header('Received', 'by MailDir; {}'.format(time.ctime(when)))
 126         msg.add_header('Message-ID', '<{}>'.format(msgid))
 127         msg.add_header('Date', time.ctime(when))
 128         msg.add_header('From', frm)
 129         msg.add_header('To', to)
 130         msg.add_header('Subject', subject)
 131         if headers:
 132             for h, v in headers:
 133                 msg.add_header(h, v)
 134         msg.set_content(body)
 135         mdmsg = mailbox.MaildirMessage(msg)
 136         if not new:
 137             mdmsg.set_subdir('cur')
 138         if flagged:
 139             mdmsg.add_flag('F')
 140         if replied:
 141             mdmsg.add_flag('R')
 142         if seen:
 143             mdmsg.add_flag('S')
 144         boxid = self.mailbox.add(mdmsg)
 145         basename = boxid
 146         if mdmsg.get_info():
 147             basename += mailbox.Maildir.colon + mdmsg.get_info()
 148         msgpath = self.path / mdmsg.get_subdir() / basename
 149         return (msgid, msgpath)