diff options
| author | Danilo M. <danix@danix.xyz> | 2026-08-23 20:39:06 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-08-23 20:39:06 +0200 |
| commit | f7948d98fe09c551856b04ba8a473dbd40649dc3 (patch) | |
| tree | a3757a879fb49ad0b797c1f7ce062801bd80f516 /assets/hooks | |
| parent | a9d1cf73a91b5eef79a9722ec9921c97b9ec5c81 (diff) | |
| download | qtmaildir-f7948d98fe09c551856b04ba8a473dbd40649dc3.tar.gz qtmaildir-f7948d98fe09c551856b04ba8a473dbd40649dc3.zip | |
feat(hooks): own the notmuch hooks, and keep sent mail out of the inbox
The post-new hook and its rule store move here from the companion mailctl
project, which is being retired. Nothing else was shared between the two, so
this is a plain move: mailrules.py is stdlib-only and post-new imports only
it.
With that in hand, the hook learns the one thing it could not know before.
notmuch's new.tags applies `inbox` to every file it indexes, and it cannot
tell an arrival from the copy this application files into a sent folder after
a send, so sent mail turned up in the inbox view and in any hand-typed
tag:inbox search. Drafts arrived the same way, through the composer's
autosave. 786 messages were affected on the developer's own index.
qtmaildirconf.py reads the sent and drafts folders out of qtmaildir.conf, so
adding an account fixes itself. Reading the application's own config is not
the cross-repo coupling it would have been last week: this repo owns the hook
now.
Three properties are load-bearing:
- it is NOT a relaxation of PROTECTED_REMOVALS, which is about a rule
removing `inbox` from mail whose provenance the hook cannot judge. Here
the provenance is the file's own path, and `inbox` was never true of it.
- only `inbox`. maildir.synchronize_flags is true, so removing `unread`
would rewrite Maildir filenames and reach the server on the next mbsync.
- an empty folder list means NOTHING, never an empty query, which notmuch
reads as "match everything". A system with no qtmaildir config must be
left alone rather than have every new message stripped.
Trash is deliberately not in the list: Delete leaves `inbox` on a trashed
message so Restore can put it back where it came from.
The three Python suites run under ctest rather than beside it as scripts
someone remembers to run, since this code tags real mail unattended on every
sync.
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Q2koFevoSxTLhfexJTZWQd
Diffstat (limited to 'assets/hooks')
| -rwxr-xr-x | assets/hooks/mailrules.py | 261 | ||||
| -rwxr-xr-x | assets/hooks/post-new | 188 | ||||
| -rwxr-xr-x | assets/hooks/qtmaildirconf.py | 134 | ||||
| -rwxr-xr-x | assets/hooks/test_mailrules.py | 278 | ||||
| -rwxr-xr-x | assets/hooks/test_post_new.py | 340 | ||||
| -rwxr-xr-x | assets/hooks/test_qtmaildirconf.py | 179 |
6 files changed, 1380 insertions, 0 deletions
diff --git a/assets/hooks/mailrules.py b/assets/hooks/mailrules.py new file mode 100755 index 0000000..dbb80a5 --- /dev/null +++ b/assets/hooks/mailrules.py @@ -0,0 +1,261 @@ +#!/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. +"""Shared notmuch tagging-rule store. + +The rules live in ~/.config/mailrules/rules.json and are read by both this +tool and qtmaildir, so the format belongs to neither: a field one tool does +not understand is preserved verbatim across a save by the other. + +A rule carries NO scope. The post-new hook supplies `tag:new`, a dry run +supplies nothing and counts against the whole corpus. This is what lets one +rule serve arrivals, a dry run, and (later) a backfill over history. + +Stdlib only, deliberately: this module is imported by a notmuch hook that +runs on every sync, and mailctl has no dependencies to inherit. +""" + +import json +import os +import re +import tempfile +from dataclasses import dataclass, field +from pathlib import Path + +FORMAT_VERSION = 1 +DEFAULT_STAGE = 50 + +# Fields this version understands. Anything else in a rule object is kept in +# `unknown` and written back untouched, which is what makes the file neutral +# rather than this tool's file that another program may read. +KNOWN_KEYS = {"id", "stage", "enabled", "add", "remove", "query", "note"} + +# An id is a handle, not a display name: a UI selects on it and a diff tracks +# it. Tags may contain '/' and may be renamed; ids may not. +ID_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$") + + +@dataclass +class Rule: + id: str + query: str + add: list = field(default_factory=list) + remove: list = field(default_factory=list) + stage: int = DEFAULT_STAGE + enabled: bool = True + note: str = "" + unknown: dict = field(default_factory=dict) + + +@dataclass +class Store: + rules: list = field(default_factory=list) + warnings: list = field(default_factory=list) + unknown: dict = field(default_factory=dict) + # Distinguishes "no file yet" from "a file that would not load". The hook + # treats them differently: the first is a fresh install, the second must + # not consume tag:new. + missing: bool = False + failed: bool = False + + +def default_path(): + """$XDG_CONFIG_HOME/mailrules/rules.json, or ~/.config/... as fallback. + + No hardcoded home directory: both tools must resolve the same path, and + a user with XDG_CONFIG_HOME set expects it honoured. + """ + base = os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config" + return Path(base) / "mailrules" / "rules.json" + + +def load(path=None): + """Read the store. Never raises for a bad file: problems land in + Store.warnings and the offending rule is dropped, so one malformed rule + cannot stop the other nineteen from running.""" + path = Path(path) if path else default_path() + store = Store() + + if not path.exists(): + store.missing = True + return store + + try: + raw = json.loads(path.read_text()) + except (json.JSONDecodeError, OSError) as exc: + store.warnings.append(f"{path}: cannot read: {exc}") + store.failed = True + return store + + if not isinstance(raw, dict): + store.warnings.append(f"{path}: top level is not an object") + store.failed = True + return store + + version = raw.get("version", FORMAT_VERSION) + if version != FORMAT_VERSION: + store.warnings.append( + f"{path}: format version {version} is newer than this tool " + f"understands ({FORMAT_VERSION}); refusing to guess") + store.failed = True + return store + + store.unknown = {k: v for k, v in raw.items() + if k not in ("version", "rules")} + + seen = set() + for index, obj in enumerate(raw.get("rules", [])): + rule = _parse_rule(obj, index, seen, store.warnings) + if rule is not None: + seen.add(rule.id) + store.rules.append(rule) + + return store + + +def scoped_query(rule, scope): + """The rule's query narrowed by `scope`, or the bare query when scope is + empty. + + The parentheses are load-bearing. notmuch's `and` binds tighter than + `or`, so `tag:new and a or b` means `(tag:new and a) or b`: a rule that + is a disjunction of senders would escape its scope and match the whole + corpus. Do not remove them, and do not build this string anywhere else. + """ + if not scope: + return rule.query + return f"{scope} and ({rule.query})" + + +def tag_arguments(rule): + """The +tag/-tag arguments for `notmuch tag`, adds before removes.""" + return [f"+{t}" for t in rule.add] + [f"-{t}" for t in rule.remove] + + +def save(store, path=None): + """Write the store atomically: a temp file in the same directory, then + rename. Rename within a filesystem is atomic, so a concurrent reader sees + either the old file or the new one and never a partial write. + + There is no locking. Last writer wins on a true collision, which is + accepted for a single-user setup; the failure that would actually hurt is + a truncated read by the hook, and rename eliminates it. + """ + path = Path(path) if path else default_path() + path.parent.mkdir(parents=True, exist_ok=True) + + payload = dict(store.unknown) + payload["version"] = FORMAT_VERSION + payload["rules"] = [_rule_to_dict(r) for r in store.rules] + + # delete=False plus an explicit replace: NamedTemporaryFile would unlink + # the file on close, and the rename is the whole point. + handle = tempfile.NamedTemporaryFile( + mode="w", dir=path.parent, prefix=".rules-", suffix=".tmp", + delete=False) + try: + with handle: + json.dump(payload, handle, indent=2, ensure_ascii=False) + handle.write("\n") + handle.flush() + os.fsync(handle.fileno()) + os.replace(handle.name, path) + except BaseException: + # A failed write must not leave the temp file beside the real one. + try: + os.unlink(handle.name) + except OSError: + pass + raise + + +def _rule_to_dict(rule): + """Known fields first in a stable order, then anything this version did + not understand. Stable ordering keeps a diff of this file readable.""" + out = { + "id": rule.id, + "stage": rule.stage, + "enabled": rule.enabled, + "add": list(rule.add), + "remove": list(rule.remove), + "query": rule.query, + "note": rule.note, + } + out.update(rule.unknown) + return out + + +def ordered(rules): + """Enabled rules in execution order: by stage ascending, ties by position. + + `sorted` is stable, so sorting on stage alone preserves file order within + a stage. That is the tie-break the format promises, and it is why this + does not sort on (stage, id): an id-sorted tie would reorder rules a user + deliberately sequenced. + """ + return sorted([r for r in rules if r.enabled], key=lambda r: r.stage) + + +def _parse_rule(obj, index, seen, warnings): + """One rule, or None with a warning appended. `index` names the rule when + it has no usable id of its own.""" + where = f"rule #{index + 1}" + + if not isinstance(obj, dict): + warnings.append(f"{where}: not an object; dropped") + return None + + rule_id = obj.get("id", "") + if not isinstance(rule_id, str) or not ID_RE.match(rule_id): + warnings.append( + f"{where}: id '{rule_id}' is missing or not lowercase " + f"letters, digits and dashes; dropped") + return None + + if rule_id in seen: + warnings.append(f"rule '{rule_id}': duplicate id; keeping the first") + return None + + query = obj.get("query", "") + if not isinstance(query, str) or not query.strip(): + warnings.append(f"rule '{rule_id}': no query; dropped") + return None + + add = [t for t in obj.get("add", []) if isinstance(t, str) and t.strip()] + remove = [t for t in obj.get("remove", []) if isinstance(t, str) and t.strip()] + if not add and not remove: + warnings.append( + f"rule '{rule_id}': adds and removes nothing; dropped") + return None + + try: + stage = int(obj.get("stage", DEFAULT_STAGE)) + except (TypeError, ValueError): + warnings.append( + f"rule '{rule_id}': stage '{obj.get('stage')}' is not a " + f"number; using {DEFAULT_STAGE}") + stage = DEFAULT_STAGE + + return Rule( + id=rule_id, + query=query, + add=add, + remove=remove, + stage=stage, + enabled=bool(obj.get("enabled", True)), + note=obj.get("note", "") if isinstance(obj.get("note", ""), str) else "", + unknown={k: v for k, v in obj.items() if k not in KNOWN_KEYS}, + ) diff --git a/assets/hooks/post-new b/assets/hooks/post-new new file mode 100755 index 0000000..5102103 --- /dev/null +++ b/assets/hooks/post-new @@ -0,0 +1,188 @@ +#!/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. +"""notmuch post-new hook: auto-tag incoming mail from the shared rule store. + +Runs after every `notmuch new`. Reads ~/.config/mailrules/rules.json and +applies each enabled rule scoped to `tag:new`, in stage order, then consumes +the `tag:new` marker. + +Rules may add any tag and remove most, but this hook REFUSES to remove `unread` +or `inbox` unattended and skips any rule that asks: see PROTECTED_REMOVALS +below for why, and for the conditions under which that restriction should be +lifted. It is expected to be relaxed once there is a story for confirming such +a rule before it runs. + +Requires `new` in [new] tags= in ~/.notmuch-config. Without it every scoped +query matches nothing and this silently no-ops. + +Install: copy to <database.path>/.notmuch/hooks/post-new, with mailrules.py +importable (same directory, or on PYTHONPATH). +""" + +import subprocess +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +import mailrules +import qtmaildirconf + +SCOPE = "tag:new" + +# Tags this hook refuses to REMOVE, whatever a rule says. +# +# maildir.synchronize_flags is true, so `unread` is not just an index entry: +# removing it rewrites Maildir filenames and propagates to the server on the +# next mbsync. `inbox` is what keeps mail visible at all. Unattended, on every +# sync, either one silently reorganizes a mailbox in a way that is tedious to +# undo and reaches other clients before anyone notices. +# +# Adding these tags is untouched, and so is removing anything else: a rule may +# still strip `promo` or any tag of its own making. +# +# DELIBERATELY CONSERVATIVE, AND EXPECTED TO BE RELAXED. The rules in use today +# only add tags, so this forbids nothing anyone is doing. It exists because the +# hook runs unattended and a mistake here is expensive, not because removing +# `unread` is wrong in principle: an "archive anything in notify/* older than +# 90 days" rule is a reasonable thing to want and would need this list revised. +# When that day comes, the question to answer first is what confirms the rule +# before it runs, not whether the guard is annoying. +# NOT the same list as mailctl.py's PROTECTED_REMOVALS, and the two must not be +# merged. That one is `{inbox}` and is a GATE: a human can override it with +# --confirm-destructive. This one is `{unread, inbox}` and is a REFUSAL, because +# there is no human present to confirm anything when cron runs a sync. +PROTECTED_REMOVALS = frozenset({"unread", "inbox"}) + + +def log(message): + print(f"post-new: {message}", file=sys.stderr) + + +def strip_inbox_from_sent(run): + """Take `inbox` off mail the user SENT, and nothing else. + + `notmuch new` applies new.tags to every file it indexes, and it cannot + tell an arrival from the copy qtmaildir files into a sent folder after a + send. The result is sent mail carrying `inbox`, which puts it in an inbox + view it never arrived in and in any hand-typed `tag:inbox` search. + + This is NOT a relaxation of PROTECTED_REMOVALS below, and the difference + is the whole reason it can run unattended. That guard is about a RULE + removing `inbox` from mail whose provenance the hook cannot judge. Here + the provenance is the file's own path: a message inside a configured sent + folder is one this system sent, and `inbox` was never true of it. Nothing + the user could act on is being hidden. + + Only `inbox`. `unread` is untouched, because maildir.synchronize_flags is + true and removing it rewrites Maildir filenames, which reaches the server + on the next mbsync. + + Scoped to tag:new like every rule, so a sync never rewrites tags across + the whole corpus. Mail already indexed keeps whatever it has. + """ + folders = qtmaildirconf.sent_folders() + if not folders: + # No config, or no account keeping sent mail locally. Nothing to + # protect, and this must NOT fall through to an empty query: notmuch + # reads that as "match everything", which would strip `inbox` from + # every newly indexed message on the system. + return True + + query = f"{SCOPE} and ({qtmaildirconf.sent_query(folders)})" + if not run(["-inbox"], query): + return False + + log(f"sent-folder carve-out applied over {len(folders)} folder(s)") + return True + + +def protected_removals(rule): + """The protected tags this rule would remove, if any.""" + return sorted(PROTECTED_REMOVALS.intersection(rule.remove)) + + +def run_tag(arguments, query): + result = subprocess.run(["notmuch", "tag"] + arguments + ["--", query], + capture_output=True, text=True) + if result.returncode != 0: + log(f"notmuch tag failed: {result.stderr.strip()}") + return False + return True + + +def main(): + store = mailrules.load() + + # A file that will not load must NOT reach the consumer below. If the + # marker were cleared while the rules did not run, that mail could never + # be tagged by these rules again: the failure is silent, permanent, and + # invisible until someone notices a gap months later. Leaving tag:new in + # place makes the next successful run catch up instead. + if store.failed: + for warning in store.warnings: + log(warning) + log("rules did not load; leaving tag:new in place") + return 1 + + if store.missing: + log("no rules file; nothing to do") + return 0 + + # A dropped rule is not fatal, but it must be visible: this goes to the + # sync log, which is where someone looks when a tag stops appearing. + for warning in store.warnings: + log(warning) + + applied = 0 + for rule in mailrules.ordered(store.rules): + # Skip the rule, do not abort the run. A single over-reaching rule + # must not cost the tagging every other rule would have done, and + # aborting here would also leave tag:new set forever: the rule would + # be refused again on every subsequent sync and the marker would never + # be consumed. + refused = protected_removals(rule) + if refused: + log(f"rule '{rule.id}' would remove {', '.join(refused)}; " + f"skipped, this hook does not remove those unattended") + continue + + query = mailrules.scoped_query(rule, SCOPE) + if not run_tag(mailrules.tag_arguments(rule), query): + log(f"rule '{rule.id}' failed; leaving tag:new in place") + return 1 + applied += 1 + + # AFTER the rules and BEFORE the marker is consumed. After, so a rule can + # still see its own sent mail with `inbox` on it and match the way it + # always did; before, because the marker is what scopes this to newly + # indexed mail and consuming it first would leave nothing to match. + if not strip_inbox_from_sent(run_tag): + log("sent-folder carve-out failed; leaving tag:new in place") + return 1 + + # Only after every rule succeeded. A failure part way through leaves the + # marker set, so re-running the hook is safe and finishes the work. + if not run_tag(["-new"], SCOPE): + return 1 + + log(f"applied {applied} rule(s)") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/assets/hooks/qtmaildirconf.py b/assets/hooks/qtmaildirconf.py new file mode 100755 index 0000000..e709a28 --- /dev/null +++ b/assets/hooks/qtmaildirconf.py @@ -0,0 +1,134 @@ +#!/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. +"""Reads the account layout out of qtmaildir.conf, for the post-new hook. + +Only the sent folders are read, and only so the hook can tell mail the user +SENT from mail that arrived. Everything else in that file belongs to the +application. + +Stdlib only: this is imported by a notmuch hook that runs on every sync. + +The file is written by QSettings rather than by configparser, and the two +disagree in one place that matters here. QSettings treats `/` in a section +name as a group separator, so accounts are `[account.<key>]` with a DOT, and +that key may itself contain dots (`[account.provider.name]`). The account key +is therefore everything after the FIRST dot, never a split on the last one. +""" + +import configparser +from pathlib import Path + +ACCOUNT_PREFIX = "account." + + +def default_path(): + """~/.config/qtmaildir/qtmaildir.conf, honouring XDG_CONFIG_HOME. + + Read through the environment rather than hardcoded so a test can point + at a throwaway config, which is how the hook's own tests reach it. + """ + import os + base = os.environ.get("XDG_CONFIG_HOME") or (Path.home() / ".config") + return Path(base) / "qtmaildir" / "qtmaildir.conf" + + +def _accounts(path): + """Every `[account.*]` section as a dict, or nothing at all. + + A file that will not parse yields NO accounts rather than raising. The + caller is a hook running after `notmuch new` has already indexed the + mail: failing the sync over a malformed application config is worse than + not protecting sent mail for one cycle, and the hook logs the miss. + """ + parser = configparser.ConfigParser( + # QSettings writes `;` comments, and `#` appears inside values (a + # colour is `#2f6fa8`), so `#` must NOT introduce a comment. + comment_prefixes=(";",), + # A value may contain `%` and `$`; neither is an interpolation here. + interpolation=None, + # `[Gmail]/Posta inviata` is a legal value. Nothing in this file + # relies on duplicate keys, but tolerating them beats raising. + strict=False) + try: + # Explicit UTF-8: QSettings writes it, and the C locale would + # otherwise decide. + with open(path, encoding="utf-8") as handle: + parser.read_file(handle) + except (OSError, UnicodeDecodeError, configparser.Error): + return [] + + return [(name[len(ACCOUNT_PREFIX):], parser[name]) + for name in parser.sections() + if name.startswith(ACCOUNT_PREFIX)] + + +# Folders mail does not ARRIVE in: this system put the message there itself. +# +# Trash is deliberately absent. qtmaildir's own Delete leaves `inbox` on a +# trashed message so Restore can put it back where it came from, and stripping +# it here would fight that. +NOT_ARRIVALS = ("sent", "drafts") + + +def sent_folders(path=None): + """Every folder mail does not arrive in, relative to the mail root. + + An account contributes nothing unless it names a maildir: a bare `Sent` + would match every account's folder of that name at once. Each of the keys + in NOT_ARRIVALS is optional on its own, since an account may keep no sent + mail or no drafts locally. + """ + if path is None: + path = default_path() + + folders = [] + for _key, section in _accounts(path): + maildir = section.get("maildir", "").strip() + if not maildir: + continue + for key in NOT_ARRIVALS: + folder = section.get(key, "").strip() + if folder: + folders.append(f"{maildir}/{folder}") + return folders + + +def sent_query(folders): + """A notmuch query matching everything inside the given folders. + + Empty for an empty list, and the caller MUST check: an empty query means + "match everything" to notmuch, so handing this straight to a tag command + would treat the whole corpus as sent mail. + + `path:` is hierarchical, so `<folder>/**` covers `cur/` and `new/` and + any nesting a provider invents underneath. + """ + if not folders: + return "" + + terms = [f'path:"{_quote(folder)}/**"' for folder in folders] + return " or ".join(terms) + + +def _quote(value): + """Escape a folder name for a double-quoted notmuch term. + + Backslashes BEFORE quotes: the other order escapes the backslashes just + added. Same rule as SearchTerm::quote() in the application, and the same + reason. + """ + return value.replace("\\", "\\\\").replace('"', '\\"') diff --git a/assets/hooks/test_mailrules.py b/assets/hooks/test_mailrules.py new file mode 100755 index 0000000..b1f31c9 --- /dev/null +++ b/assets/hooks/test_mailrules.py @@ -0,0 +1,278 @@ +#!/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. +"""Self-checks for mailrules.py, the shared tagging-rule store. + +The risk in this file is the format, not the notmuch calls: a rule that +silently loses a field on save, or one that sorts into the wrong stage, +mis-tags real mail on the next sync and does it quietly. + +Run: ./test_mailrules.py +""" + +import json +import tempfile +from pathlib import Path + +import mailrules + + +def write_rules(tmp, payload): + path = Path(tmp) / "rules.json" + path.write_text(json.dumps(payload)) + return path + + +def test_loads_a_rule(): + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "rules": [ + { + "id": "notify-forge", + "stage": 50, + "enabled": True, + "add": ["notify/forge"], + "remove": [], + "query": "from:notifications@example.com", + "note": "All repositories, not one project.", + } + ], + }) + store = mailrules.load(path) + assert store.warnings == [], store.warnings + assert len(store.rules) == 1 + rule = store.rules[0] + assert rule.id == "notify-forge" + assert rule.stage == 50 + assert rule.enabled is True + assert rule.add == ["notify/forge"] + assert rule.remove == [] + assert rule.query == "from:notifications@example.com" + assert rule.note == "All repositories, not one project." + + +def test_defaults_are_applied(): + """stage, enabled, remove and note are all optional in the file.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "rules": [{"id": "minimal", "add": ["x"], + "query": "from:someone@example.com"}], + }) + store = mailrules.load(path) + assert store.warnings == [], store.warnings + rule = store.rules[0] + assert rule.stage == 50 + assert rule.enabled is True + assert rule.remove == [] + assert rule.note == "" + + +def test_a_bad_rule_is_dropped_and_the_rest_survive(): + """One malformed rule must not stop the others. The hook runs every ten + minutes on real mail; losing all tagging because of one typo is worse + than losing one rule.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "rules": [ + {"id": "good", "add": ["x"], "query": "from:a@example.com"}, + {"id": "no-query", "add": ["y"]}, + {"id": "no-tags", "query": "from:b@example.com"}, + {"id": "bad id!", "add": ["z"], "query": "from:c@example.com"}, + {"add": ["w"], "query": "from:d@example.com"}, + ], + }) + store = mailrules.load(path) + assert [r.id for r in store.rules] == ["good"] + assert len(store.warnings) == 4, store.warnings + joined = " ".join(store.warnings) + assert "no-query" in joined + assert "no-tags" in joined + assert "bad id!" in joined + + +def test_duplicate_ids_keep_the_first(): + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "rules": [ + {"id": "dup", "add": ["first"], "query": "from:a@example.com"}, + {"id": "dup", "add": ["second"], "query": "from:b@example.com"}, + ], + }) + store = mailrules.load(path) + assert len(store.rules) == 1 + assert store.rules[0].add == ["first"] + assert any("dup" in w for w in store.warnings) + + +def test_a_missing_file_is_empty_not_an_error(): + """qtmaildir must open on a machine that has never written this file.""" + with tempfile.TemporaryDirectory() as tmp: + store = mailrules.load(Path(tmp) / "absent.json") + assert store.rules == [] + assert store.warnings == [] + assert store.missing is True + + +def test_unparseable_json_warns_and_yields_no_rules(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "rules.json" + path.write_text("{not json") + store = mailrules.load(path) + assert store.rules == [] + assert len(store.warnings) == 1 + assert store.failed is True + + +def test_a_newer_format_version_is_refused(): + """Guessing at semantics a later version defined is how a rule silently + changes meaning. Refuse instead.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 2, + "rules": [{"id": "x", "add": ["a"], "query": "from:a@example.com"}], + }) + store = mailrules.load(path) + assert store.rules == [] + assert store.failed is True + assert any("version" in w for w in store.warnings) + + +def test_ordered_sorts_by_stage_then_file_position(): + """Account tags must run before topic rules. Ties keep file order, so + the file still reads as a sequence.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "rules": [ + {"id": "topic-b", "stage": 50, "add": ["b"], + "query": "from:b@example.com"}, + {"id": "account", "stage": 10, "add": ["acct"], + "query": "path:\"work/**\""}, + {"id": "topic-a", "stage": 50, "add": ["a"], + "query": "from:a@example.com"}, + ], + }) + store = mailrules.load(path) + assert [r.id for r in mailrules.ordered(store.rules)] == [ + "account", "topic-b", "topic-a"] + + +def test_ordered_skips_disabled_rules(): + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "rules": [ + {"id": "on", "add": ["a"], "query": "from:a@example.com"}, + {"id": "off", "add": ["b"], "query": "from:b@example.com", + "enabled": False}, + ], + }) + store = mailrules.load(path) + assert [r.id for r in mailrules.ordered(store.rules)] == ["on"] + # The disabled rule is still LOADED, so a UI can show and re-enable it. + assert [r.id for r in store.rules] == ["on", "off"] + + +def test_save_round_trips_unknown_fields(): + """The neutrality guarantee. If this tool strips a field qtmaildir + added, the file is this tool's file that qtmaildir may read.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_rules(tmp, { + "version": 1, + "future_top_level": {"set_by": "another tool"}, + "rules": [{ + "id": "keeper", + "add": ["x"], + "query": "from:a@example.com", + "future_field": [1, 2, 3], + }], + }) + store = mailrules.load(path) + assert store.rules[0].unknown == {"future_field": [1, 2, 3]} + + mailrules.save(store, path) + + raw = json.loads(path.read_text()) + assert raw["future_top_level"] == {"set_by": "another tool"} + assert raw["rules"][0]["future_field"] == [1, 2, 3] + assert raw["rules"][0]["id"] == "keeper" + assert raw["version"] == 1 + + +def test_save_is_atomic(): + """A reader must never see a half-written file: the hook runs every ten + minutes and a truncated read would be a failed sync.""" + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "rules.json" + store = mailrules.Store(rules=[ + mailrules.Rule(id="a", query="from:a@example.com", add=["x"])]) + mailrules.save(store, path) + # The temp file the write went through must not be left behind. + assert [p.name for p in Path(tmp).iterdir()] == ["rules.json"] + assert json.loads(path.read_text())["rules"][0]["id"] == "a" + + +def test_save_creates_the_directory(): + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "nested" / "rules.json" + mailrules.save(mailrules.Store(), path) + assert path.exists() + assert json.loads(path.read_text()) == {"version": 1, "rules": []} + + +def test_scoped_query_parenthesises_the_rule(): + """Without the parentheses `tag:new and a or b` binds as + `(tag:new and a) or b`, and the rule matches every message in the corpus + satisfying b rather than only new arrivals. Several real rules are a + disjunction of senders, so this is the difference between tagging four + messages and tagging four thousand.""" + rule = mailrules.Rule( + id="disjunction", + query="from:a@example.com or from:b@example.com", + add=["promo"]) + assert mailrules.scoped_query(rule, "tag:new") == ( + "tag:new and (from:a@example.com or from:b@example.com)") + + +def test_scoped_query_with_no_scope_is_the_bare_query(): + """A dry run counts against the whole corpus, which is what makes the + same rule answer 'what would this tag on arrival' and 'what does this + match in all my mail'.""" + rule = mailrules.Rule(id="x", query="from:a@example.com", add=["y"]) + assert mailrules.scoped_query(rule, None) == "from:a@example.com" + assert mailrules.scoped_query(rule, "") == "from:a@example.com" + + +def test_tag_arguments(): + rule = mailrules.Rule(id="x", query="from:a@example.com", + add=["one", "two"], remove=["three"]) + assert mailrules.tag_arguments(rule) == ["+one", "+two", "-three"] + + +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") diff --git a/assets/hooks/test_post_new.py b/assets/hooks/test_post_new.py new file mode 100755 index 0000000..a0228aa --- /dev/null +++ b/assets/hooks/test_post_new.py @@ -0,0 +1,340 @@ +#!/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 setup_accounts(tmp, sent_config=True): + """A maildir laid out as qtmaildir configures it: two accounts, each with + an Inbox and a Sent folder, one message in each. + + Separate from setup_database() because the sent carve-out is the only + thing that cares where a file sits. The folder names are the awkward + ones deliberately: a bracketed, spaced provider folder is what the real + config carries, and a flat `Sent` is what the other half carries. + """ + root = Path(tmp) / "Mail" + folders = { + "one": ("acct-one/Inbox", "acct-one/Sent"), + "two": ("acct-two/[Provider]/Posta inviata", + "acct-two/[Provider]/Posta inviata"), + } + for sub in ("acct-one/Inbox", "acct-one/Sent", + "acct-two/Inbox", "acct-two/[Provider]/Posta inviata"): + for part in ("new", "cur", "tmp"): + (root / sub / part).mkdir(parents=True) + + make_message(root / "acct-one/Inbox", "arrived-one", + "friend@example.org", "an arrival") + make_message(root / "acct-one/Sent", "sent-one", + "you@example.org", "something sent") + make_message(root / "acct-two/Inbox", "arrived-two", + "friend@example.org", "another arrival") + make_message(root / "acct-two/[Provider]/Posta inviata", "sent-two", + "you@example.org", "something else sent") + + config = Path(tmp) / "notmuch-config" + config.write_text( + f"[database]\npath={root}\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") + + if sent_config: + conf = Path(env["XDG_CONFIG_HOME"]) / "qtmaildir" / "qtmaildir.conf" + conf.parent.mkdir(parents=True, exist_ok=True) + conf.write_text( + "[account.one]\nmaildir = acct-one\nsent = Sent\n" + "[account.two]\nmaildir = acct-two\n" + "sent = [Provider]/Posta inviata\n") + + subprocess.run(["notmuch", "new"], env=env, capture_output=True, + check=True) + return env + + +def test_sent_mail_does_not_keep_the_inbox_tag(): + """The carve-out. notmuch's new.tags applies `inbox` to every file it + indexes, including the copy the composer files into a sent folder, so + mail the user SENT shows up in an inbox view it never arrived in. + + Both accounts are asserted, because the folder shapes differ and a + reader that mishandles the bracketed, spaced one would still pass on the + flat `Sent`. + """ + with tempfile.TemporaryDirectory() as tmp: + env = setup_accounts(tmp) + write_rules(env, []) + assert count(env, "tag:inbox") == 4 + + result = subprocess.run([str(HOOK)], env=env, capture_output=True, + text=True) + assert result.returncode == 0, result.stderr + + # The two sent copies lose it... + assert count(env, 'tag:inbox and path:"acct-one/Sent/**"') == 0 + assert count( + env, + 'tag:inbox and path:"acct-two/[Provider]/Posta inviata/**"') == 0 + # ...and the two arrivals keep it. This is the half that fails if the + # query is unscoped, which is the expensive mistake here. + assert count(env, "tag:inbox") == 2 + assert count(env, 'tag:inbox and path:"acct-one/Inbox/**"') == 1 + assert count(env, 'tag:inbox and path:"acct-two/Inbox/**"') == 1 + + +def test_sent_mail_keeps_every_other_tag(): + """Only `inbox` is stripped. `unread` in particular must survive: + maildir.synchronize_flags is true, so removing it rewrites Maildir + filenames and reaches the server on the next mbsync. + """ + with tempfile.TemporaryDirectory() as tmp: + env = setup_accounts(tmp) + write_rules(env, []) + subprocess.run([str(HOOK)], env=env, capture_output=True, check=True) + assert count(env, 'tag:unread and path:"acct-one/Sent/**"') == 1 + + +def test_the_carve_out_only_touches_newly_indexed_mail(): + """Scoped to tag:new like every rule, so the hook never rewrites tags + across the whole corpus on a sync. A sent message whose `inbox` tag was + put back by hand stays that way until it is reindexed. + """ + with tempfile.TemporaryDirectory() as tmp: + env = setup_accounts(tmp) + write_rules(env, []) + subprocess.run([str(HOOK)], env=env, capture_output=True, check=True) + assert count(env, 'tag:inbox and path:"acct-one/Sent/**"') == 0 + + subprocess.run(["notmuch", "tag", "+inbox", "--", + 'path:"acct-one/Sent/**"'], env=env, check=True) + subprocess.run([str(HOOK)], env=env, capture_output=True, check=True) + assert count(env, 'tag:inbox and path:"acct-one/Sent/**"') == 1 + + +def test_no_qtmaildir_config_leaves_every_tag_alone(): + """The hook must run on a system with no qtmaildir config: it then + protects nothing rather than failing the sync, and above all does not + treat an empty folder list as "every path", which is what an empty + notmuch query means. + """ + with tempfile.TemporaryDirectory() as tmp: + env = setup_accounts(tmp, sent_config=False) + write_rules(env, []) + result = subprocess.run([str(HOOK)], env=env, capture_output=True, + text=True) + assert result.returncode == 0, result.stderr + assert count(env, "tag:inbox") == 4 + + +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") diff --git a/assets/hooks/test_qtmaildirconf.py b/assets/hooks/test_qtmaildirconf.py new file mode 100755 index 0000000..c8aa78d --- /dev/null +++ b/assets/hooks/test_qtmaildirconf.py @@ -0,0 +1,179 @@ +#!/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. +"""Unit checks for the qtmaildir.conf reader the post-new hook uses to find +the sent folders. + +The file is written by QSettings, not by configparser, so the cases that +matter are the ones where the two disagree: a section name carrying a dot, a +comment introduced by `;`, and a key present but empty. + +Run: ./test_qtmaildirconf.py +""" + +import tempfile +from pathlib import Path + +import qtmaildirconf + + +def write_config(tmp, text): + path = Path(tmp) / "qtmaildir" / "qtmaildir.conf" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text) + return path + + +def test_sent_folders_are_read_per_account(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.work]\n" + "maildir = work\n" + "sent = Sent\n" + "trash = Trash\n") + assert qtmaildirconf.sent_folders(path) == ["work/Sent"] + + +def test_drafts_are_excluded_alongside_sent(): + """A draft never arrived either, so it must not carry `inbox`. Both keys + feed one list: the hook asks a single question, "is this a folder mail + arrives in", and sent and drafts answer it the same way. + + Trash is deliberately NOT here. qtmaildir's own Delete leaves `inbox` on + a trashed message so Restore can put it back where it came from, and + stripping it here would fight that. + """ + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.work]\n" + "maildir = work\n" + "sent = Sent\n" + "drafts = Drafts\n" + "trash = Trash\n") + assert qtmaildirconf.sent_folders(path) == ["work/Sent", "work/Drafts"] + + +def test_an_account_with_only_drafts_still_contributes(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.a]\nmaildir = a\ndrafts = Drafts\n") + assert qtmaildirconf.sent_folders(path) == ["a/Drafts"] + + +def test_an_account_section_may_carry_a_dot(): + """QSettings writes `[account.a.b]` for the key `a.b`, and the account + key is everything after the first dot. Splitting on the LAST dot names + an account that does not exist and finds no folder.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.provider.name]\n" + "maildir = provider-name\n" + "sent = Sent\n") + assert qtmaildirconf.sent_folders(path) == ["provider-name/Sent"] + + +def test_a_folder_may_contain_spaces_and_brackets(): + """`[Gmail]/Posta inviata` is a real folder name here. The brackets are + the provider's, not INI syntax, because they are in a VALUE.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.g]\n" + "maildir = gmail\n" + "sent = [Gmail]/Posta inviata\n") + assert qtmaildirconf.sent_folders(path) == [ + "gmail/[Gmail]/Posta inviata"] + + +def test_an_account_without_a_sent_key_contributes_nothing(): + """`sent` is optional: an account may keep no sent mail locally. It must + not contribute an entry, since a bare `maildir/` prefix would match the + whole account.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.a]\nmaildir = a\ntrash = Trash\n" + "[account.b]\nmaildir = b\nsent = Sent\n") + assert qtmaildirconf.sent_folders(path) == ["b/Sent"] + + +def test_an_empty_sent_value_contributes_nothing(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.a]\nmaildir = a\nsent =\n") + assert qtmaildirconf.sent_folders(path) == [] + + +def test_an_account_without_a_maildir_contributes_nothing(): + """Without the account's own subdirectory the folder cannot be located, + and a bare `Sent` would match every account's sent folder at once.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "[account.a]\nsent = Sent\n") + assert qtmaildirconf.sent_folders(path) == [] + + +def test_comments_and_other_sections_are_ignored(): + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "; a comment\n" + "[general]\n" + "language = it\n" + "[sync]\n" + "command = /bin/true\n" + "[account.a]\n" + "; another comment\n" + "maildir = a\n" + "sent = Sent\n") + assert qtmaildirconf.sent_folders(path) == ["a/Sent"] + + +def test_a_missing_file_yields_no_folders(): + """The hook must run on a system with no qtmaildir config at all: it + then protects nothing, rather than failing the sync.""" + with tempfile.TemporaryDirectory() as tmp: + assert qtmaildirconf.sent_folders(Path(tmp) / "absent.conf") == [] + + +def test_an_unreadable_file_yields_no_folders(): + """A malformed config must not fail the sync. notmuch new has already + run at this point; refusing to tag is worse than not protecting sent + mail for one cycle.""" + with tempfile.TemporaryDirectory() as tmp: + path = write_config(tmp, "this is not an ini file\n[[[\n") + assert qtmaildirconf.sent_folders(path) == [] + + +def test_the_query_scopes_every_folder(): + folders = ["a/Sent", "g/[Gmail]/Posta inviata"] + query = qtmaildirconf.sent_query(folders) + assert query == ('path:"a/Sent/**" or path:"g/[Gmail]/Posta inviata/**"') + + +def test_the_query_is_empty_when_no_folder_is_configured(): + """An empty query means "match everything" to notmuch, so the caller + must be able to tell "nothing to protect" from "protect the world".""" + assert qtmaildirconf.sent_query([]) == "" + + +def test_a_folder_containing_a_quote_cannot_break_out_of_the_query(): + """The folder name reaches a notmuch query as a quoted string. A stray + double quote would end the term and let the rest be read as syntax.""" + query = qtmaildirconf.sent_query(['a/He said "hi"']) + assert query.count('"') % 2 == 0 + assert "\\\"" in query or '""' in query + + +def main(): + tests = [value for name, value in sorted(globals().items()) + if name.startswith("test_") and callable(value)] + for test in tests: + test() + print(f"ok {test.__name__}") + print(f"\n{len(tests)} passed") + + +if __name__ == "__main__": + main() |
