Introduce CFFI-based python bindings

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

diff --git a/bindings/python-cffi/tests/conftest.py b/bindings/python-cffi/tests/conftest.py
new file mode 100644 (file)
index 0000000..1b7bbc3
--- /dev/null
+++ b/bindings/python-cffi/tests/conftest.py
@@ -0,0 +1,142 @@
+import email.message
+import mailbox
+import pathlib
+import socket
+import subprocess
+import textwrap
+import time
+
+import pytest
+
+
+@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)
+        print('Invoking: {}'.format(' '.join(cmd)))
+        proc = subprocess.run(cmd,
+                              timeout=5,
+                              env={'NOTMUCH_CONFIG': str(cfg_fname)})
+        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)