diff options
| -rwxr-xr-x | mailrules.py | 108 | ||||
| -rwxr-xr-x | test_mailrules.py | 72 |
2 files changed, 167 insertions, 13 deletions
diff --git a/mailrules.py b/mailrules.py index c529348..26ed698 100755 --- a/mailrules.py +++ b/mailrules.py @@ -30,6 +30,7 @@ runs on every sync, and mailctl has no dependencies to inherit. import json import os +import re from dataclasses import dataclass, field from pathlib import Path @@ -41,6 +42,10 @@ DEFAULT_STAGE = 50 # 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: @@ -59,6 +64,11 @@ 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(): @@ -78,18 +88,90 @@ def load(path=None): path = Path(path) if path else default_path() store = Store() - raw = json.loads(path.read_text()) - - for obj in raw.get("rules", []): - store.rules.append(Rule( - id=obj["id"], - query=obj["query"], - add=list(obj.get("add", [])), - remove=list(obj.get("remove", [])), - stage=int(obj.get("stage", DEFAULT_STAGE)), - enabled=bool(obj.get("enabled", True)), - note=obj.get("note", ""), - unknown={k: v for k, v in obj.items() if k not in KNOWN_KEYS}, - )) + 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 _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/test_mailrules.py b/test_mailrules.py index dd491e9..f349f05 100755 --- a/test_mailrules.py +++ b/test_mailrules.py @@ -82,6 +82,78 @@ def test_defaults_are_applied(): 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): |