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
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
|
#!/usr/bin/env python3
#
# Copyright (C) 2026 Danilo M. <danix@danix.xyz>
#
# This program is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License version 2 as
# published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
"""End-to-end checks for the post-new hook against a throwaway notmuch
database. Nothing here touches the user's real mail: NOTMUCH_CONFIG points at
a generated maildir under a temp directory.
The properties worth proving are the ones that cannot be unit-tested from
mailrules.py alone:
- a rule actually tags the mail its query matches, and only that mail
- the tag:new marker is consumed on success
- the marker SURVIVES when a rule fails, so a re-run catches up
- a rule removing a protected tag is skipped whole, and the run continues
Run: ./test_post_new.py (requires notmuch on PATH)
"""
import json
import os
import subprocess
import tempfile
from pathlib import Path
HOOK = Path(__file__).resolve().parent / "post-new"
def make_message(maildir, name, sender, subject):
path = maildir / "new" / name
path.write_text(
f"From: {sender}\n"
f"To: you@example.org\n"
f"Subject: {subject}\n"
f"Message-Id: <{name}@example.org>\n"
f"Date: Mon, 11 Aug 2026 10:00:00 +0000\n"
f"\nbody\n")
def setup_database(tmp):
"""A maildir with three messages, indexed, every message carrying the
`new` marker the rules key off."""
maildir = Path(tmp) / "Mail"
for sub in ("new", "cur", "tmp"):
(maildir / sub).mkdir(parents=True)
make_message(maildir, "one", "notifications@example.com", "a notification")
make_message(maildir, "two", "friend@example.org", "a real message")
make_message(maildir, "three", "promo@example.net", "an advertisement")
config = Path(tmp) / "notmuch-config"
config.write_text(
f"[database]\npath={maildir}\n\n"
f"[new]\ntags=new;unread;inbox\n\n"
f"[user]\nname=Test\nprimary_email=you@example.org\n")
env = dict(os.environ)
env["NOTMUCH_CONFIG"] = str(config)
env["XDG_CONFIG_HOME"] = str(Path(tmp) / "config")
subprocess.run(["notmuch", "new"], env=env, capture_output=True, check=True)
return env
def write_rules(env, rules):
path = Path(env["XDG_CONFIG_HOME"]) / "mailrules" / "rules.json"
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps({"version": 1, "rules": rules}))
def count(env, query):
out = subprocess.run(["notmuch", "count", query], env=env,
capture_output=True, text=True, check=True)
return int(out.stdout.strip())
def test_a_rule_tags_only_what_it_matches():
with tempfile.TemporaryDirectory() as tmp:
env = setup_database(tmp)
write_rules(env, [{
"id": "notify",
"add": ["notify/forge"],
"query": "from:notifications@example.com",
}])
assert count(env, "tag:new") == 3
result = subprocess.run([str(HOOK)], env=env, capture_output=True,
text=True)
assert result.returncode == 0, result.stderr
assert count(env, "tag:notify/forge") == 1
assert count(env, "tag:notify/forge and from:friend@example.org") == 0
# The marker is consumed, so the next sync's rules see only new mail.
assert count(env, "tag:new") == 0
def test_stage_order_is_honoured():
with tempfile.TemporaryDirectory() as tmp:
env = setup_database(tmp)
write_rules(env, [
{"id": "late", "stage": 50, "add": ["second"],
"query": "tag:first"},
{"id": "early", "stage": 10, "add": ["first"],
"query": "from:notifications@example.com"},
])
subprocess.run([str(HOOK)], env=env, capture_output=True, check=True)
# `late` matches only what `early` tagged, so a wrong order gives 0.
assert count(env, "tag:second") == 1
def test_a_failing_rule_leaves_the_marker_in_place():
"""The property that makes a re-run safe. An invalid query fails the
notmuch call, and tag:new must survive so the next run catches up.
The query has to be one notmuch genuinely rejects, which is a narrower
set than it looks: notmuch 0.39's parser accepts unbalanced parentheses
and bare punctuation without complaint, tags nothing, and exits 0. A
malformed date range is rejected by the date parser and does exit
non-zero, which is why the fixture uses one.
"""
with tempfile.TemporaryDirectory() as tmp:
env = setup_database(tmp)
write_rules(env, [{
"id": "broken",
"add": ["x"],
"query": "date:zzz..zzz",
}])
result = subprocess.run([str(HOOK)], env=env, capture_output=True,
text=True)
assert result.returncode == 1
assert count(env, "tag:new") == 3
def test_a_disjunction_stays_inside_its_scope():
"""The parenthesisation guard, end to end. Both senders are already
indexed and out of tag:new after a first run; a rule that escaped its
scope would tag them anyway."""
with tempfile.TemporaryDirectory() as tmp:
env = setup_database(tmp)
write_rules(env, [{"id": "noop", "add": ["pass-one"],
"query": "from:nobody@example.invalid"}])
subprocess.run([str(HOOK)], env=env, capture_output=True, check=True)
assert count(env, "tag:new") == 0
write_rules(env, [{
"id": "disjunction",
"add": ["promo"],
"query": "from:friend@example.org or from:promo@example.net",
}])
subprocess.run([str(HOOK)], env=env, capture_output=True, check=True)
# Nothing carries tag:new any more, so a correctly scoped rule tags
# nothing. Unparenthesised, the `or` branch would tag one message.
assert count(env, "tag:promo") == 0
def test_a_protected_removal_is_skipped_whole_and_the_run_continues():
"""The PROTECTED_REMOVALS guard, end to end.
Four assertions in one run, because three of them pass against a guard
that is broken in a different way. A guard that skipped only the removal
would still apply the rule's adds; a guard that aborted the run would
starve every later rule; and a guard that aborted before the consumer
would strand tag:new, so every future sync would refuse the same rule
again and nothing would ever be tagged after it.
"""
with tempfile.TemporaryDirectory() as tmp:
env = setup_database(tmp)
write_rules(env, [
{"id": "over-reaching", "stage": 10,
"add": ["archived"], "remove": ["unread", "inbox"],
"query": "from:notifications@example.com"},
{"id": "well-behaved", "stage": 20, "add": ["promo"],
"query": "from:promo@example.net"},
])
assert count(env, "tag:unread") == 3
assert count(env, "tag:inbox") == 3
result = subprocess.run([str(HOOK)], env=env, capture_output=True,
text=True)
assert result.returncode == 0, result.stderr
assert "over-reaching" in result.stderr, result.stderr
# 1. the protected tags survive on the message the rule matched
assert count(env, "tag:unread and from:notifications@example.com") == 1
assert count(env, "tag:inbox and from:notifications@example.com") == 1
# 2. the rule is skipped ENTIRELY, so its adds never land either
assert count(env, "tag:archived") == 0
# 3. a later, well-behaved rule still runs
assert count(env, "tag:promo") == 1
# 4. the marker is still consumed, so the next sync is not stuck
assert count(env, "tag:new") == 0
def run_all():
for name, fn in sorted(globals().items()):
if name.startswith("test_") and callable(fn):
fn()
print(f"ok {name}")
if __name__ == "__main__":
run_all()
print("\nall passed")
|
