notmuch/bindings/python-cffi/tests/conftest.py

import email.message
import mailbox
import pathlib
import socket
import subprocess
import textwrap
import time
import os

import pytest


def pytest_report_header():
    vers = subprocess.run(['notmuch', '--version'], stdout=subprocess.PIPE)
    which = subprocess.run(['which', 'notmuch'], stdout=subprocess.PIPE)
    return ['{} ({})'.format(vers.stdout.decode(errors='replace').strip(),
                             which.stdout.decode(errors='replace').strip())]


@pytest.fixture(scope='function')
def tmppath(tmpdir):
    """The tmpdir fixture wrapped in pathlib.Path."""
    return pathlib.Path(str(tmpdir))


@pytest.fixture
def notmuch(maildir):
    """Return a function which runs notmuch commands on our test maildir.

    This uses the notmuch-config file created by the ``maildir``
    fixture.
    """
    def run(*args):
        """Run a notmuch comand.

        This function runs with a timeout error as many notmuch
        commands may block if multiple processes are trying to open
        the database in write-mode.  It is all too easy to
        accidentally do this in the unittests.
        """
        cfg_fname = maildir.path / 'notmuch-config'
        cmd = ['notmuch'] + list(args)
        env = os.environ.copy()
        env['NOTMUCH_CONFIG'] = str(cfg_fname)
        proc = subprocess.run(cmd,
                              timeout=5,
                              env=env)
        proc.check_returncode()
    return run


@pytest.fixture
def maildir(tmppath):
    """A basic test interface to a valid maildir directory.

    This creates a valid maildir and provides a simple mechanism to
    deliver test emails to it.  It also writes a notmuch-config file
    in the top of the maildir.
    """
    cur = tmppath / 'cur'
    cur.mkdir()
    new = tmppath / 'new'
    new.mkdir()
    tmp = tmppath / 'tmp'
    tmp.mkdir()
    cfg_fname = tmppath/'notmuch-config'
    with cfg_fname.open('w') as fp:
        fp.write(textwrap.dedent("""\
            [database]
            path={tmppath!s}
            [user]
            name=Some Hacker
            primary_email=dst@example.com
            [new]
            tags=unread;inbox;
            ignore=
            [search]
            exclude_tags=deleted;spam;
            [maildir]
            synchronize_flags=true
            [crypto]
            gpg_path=gpg
            """.format(tmppath=tmppath)))
    return MailDir(tmppath)


class MailDir:
    """An interface around a correct maildir."""

    def __init__(self, path):
        self._path = pathlib.Path(path)
        self.mailbox = mailbox.Maildir(str(path))
        self._idcount = 0

    @property
    def path(self):
        """The pathname of the maildir."""
        return self._path

    def _next_msgid(self):
        """Return a new unique message ID."""
        msgid = '{}@{}'.format(self._idcount, socket.getfqdn())
        self._idcount += 1
        return msgid

    def deliver(self,
                subject='Test mail',
                body='This is a test mail',
                to='dst@example.com',
                frm='src@example.com',
                headers=None,
                new=False,      # Move to new dir or cur dir?
                keywords=None,  # List of keywords or labels
                seen=False,     # Seen flag (cur dir only)
                replied=False,  # Replied flag (cur dir only)
                flagged=False):  # Flagged flag (cur dir only)
        """Deliver a new mail message in the mbox.

        This does only adds the message to maildir, does not insert it
        into the notmuch database.

        :returns: A tuple of (msgid, pathname).
        """
        msgid = self._next_msgid()
        when = time.time()
        msg = email.message.EmailMessage()
        msg.add_header('Received', 'by MailDir; {}'.format(time.ctime(when)))
        msg.add_header('Message-ID', '<{}>'.format(msgid))
        msg.add_header('Date', time.ctime(when))
        msg.add_header('From', frm)
        msg.add_header('To', to)
        msg.add_header('Subject', subject)
        if headers:
            for h, v in headers:
                msg.add_header(h, v)
        msg.set_content(body)
        mdmsg = mailbox.MaildirMessage(msg)
        if not new:
            mdmsg.set_subdir('cur')
        if flagged:
            mdmsg.add_flag('F')
        if replied:
            mdmsg.add_flag('R')
        if seen:
            mdmsg.add_flag('S')
        boxid = self.mailbox.add(mdmsg)
        basename = boxid
        if mdmsg.get_info():
            basename += mailbox.Maildir.colon + mdmsg.get_info()
        msgpath = self.path / mdmsg.get_subdir() / basename
        return (msgid, msgpath)