diff options
| author | David Bremner <david@tethera.net> | 2020-12-13 10:38:31 -0400 |
|---|---|---|
| committer | David Bremner <david@tethera.net> | 2020-12-13 10:38:31 -0400 |
| commit | 7a9c97e8a57f2662b9069dae01b6e5cb2f650563 (patch) | |
| tree | 2b7bfbf5b3848866444e61f7069ccc755301f844 /bindings/python-cffi/tests/conftest.py | |
| parent | b7ca3c23d17d247bda37645c7f861b3c0d04bf25 (diff) | |
| parent | 900ee94b0f4f48ee536bd2e9bd6bb2dfc661d615 (diff) | |
Merge tag 'debian/0.31.2-3' into debian/buster-backports
notmuch release 0.31.2-3 for unstable (sid) [dgit]
[dgit distro=debian no-split --quilt=linear]
Diffstat (limited to 'bindings/python-cffi/tests/conftest.py')
| -rw-r--r-- | bindings/python-cffi/tests/conftest.py | 149 |
1 files changed, 149 insertions, 0 deletions
diff --git a/bindings/python-cffi/tests/conftest.py b/bindings/python-cffi/tests/conftest.py new file mode 100644 index 00000000..6835fd30 --- /dev/null +++ b/bindings/python-cffi/tests/conftest.py @@ -0,0 +1,149 @@ +import email.message +import mailbox +import pathlib +import shutil +import socket +import subprocess +import textwrap +import time +import os + +import pytest + + +def pytest_report_header(): + which = shutil.which('notmuch') + vers = subprocess.run(['notmuch', '--version'], stdout=subprocess.PIPE) + return ['{} ({})'.format(vers.stdout.decode(errors='replace').strip(),which)] + + +@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 command. + + 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 + """.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) |
