#!/usr/bin/env python3 # # Copyright (C) 2026 Danilo M. # # 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 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")