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

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