1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
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)
|