mirror of
https://git.notmuchmail.org/git/notmuch
synced 2024-12-23 09:54:52 +01:00
018ad3703b
crypto.gpg_path was only used when we built against gmime versions before 3.0. Since we now depend on gmime 3.0.3 or later, it is meaningless. The removal of the field from the _notmuch_config struct would be an ABI change if that struct were externally exposed, but it is not, so it's safe to unilaterally remove it. Signed-off-by: Daniel Kahn Gillmor <dkg@fifthhorseman.net>
149 lines
4.4 KiB
Python
149 lines
4.4 KiB
Python
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 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
|
|
""".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)
|