aboutsummaryrefslogtreecommitdiffstats
path: root/test_mailrules.py
blob: f349f05ba0afa261543df991a5c0759f69402463 (plain)
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
#!/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 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")