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 comand.
  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             [crypto]
  82             gpg_path=gpg
  83             """.format(tmppath=tmppath)))
  84     return MailDir(tmppath)
  85
  86
  87 class MailDir:
  88     """An interface around a correct maildir."""
  89
  90     def __init__(self, path):
  91         self._path = pathlib.Path(path)
  92         self.mailbox = mailbox.Maildir(str(path))
  93         self._idcount = 0
  94
  95     @property
  96     def path(self):
  97         """The pathname of the maildir."""
  98         return self._path
  99
 100     def _next_msgid(self):
 101         """Return a new unique message ID."""
 102         msgid = '{}@{}'.format(self._idcount, socket.getfqdn())
 103         self._idcount += 1
 104         return msgid
 105
 106     def deliver(self,
 107                 subject='Test mail',
 108                 body='This is a test mail',
 109                 to='dst@example.com',
 110                 frm='src@example.com',
 111                 headers=None,
 112                 new=False,      # Move to new dir or cur dir?
 113                 keywords=None,  # List of keywords or labels
 114                 seen=False,     # Seen flag (cur dir only)
 115                 replied=False,  # Replied flag (cur dir only)
 116                 flagged=False):  # Flagged flag (cur dir only)
 117         """Deliver a new mail message in the mbox.
 118
 119         This does only adds the message to maildir, does not insert it
 120         into the notmuch database.
 121
 122         :returns: A tuple of (msgid, pathname).
 123         """
 124         msgid = self._next_msgid()
 125         when = time.time()
 126         msg = email.message.EmailMessage()
 127         msg.add_header('Received', 'by MailDir; {}'.format(time.ctime(when)))
 128         msg.add_header('Message-ID', '<{}>'.format(msgid))
 129         msg.add_header('Date', time.ctime(when))
 130         msg.add_header('From', frm)
 131         msg.add_header('To', to)
 132         msg.add_header('Subject', subject)
 133         if headers:
 134             for h, v in headers:
 135                 msg.add_header(h, v)
 136         msg.set_content(body)
 137         mdmsg = mailbox.MaildirMessage(msg)
 138         if not new:
 139             mdmsg.set_subdir('cur')
 140         if flagged:
 141             mdmsg.add_flag('F')
 142         if replied:
 143             mdmsg.add_flag('R')
 144         if seen:
 145             mdmsg.add_flag('S')
 146         boxid = self.mailbox.add(mdmsg)
 147         basename = boxid
 148         if mdmsg.get_info():
 149             basename += mailbox.Maildir.colon + mdmsg.get_info()
 150         msgpath = self.path / mdmsg.get_subdir() / basename
 151         return (msgid, msgpath)