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