path: root/test_mailctl.py

#!/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 the bits of mailctl with real logic in them:

  - the senders address merge: notmuch dedupes on name-addr, so one address
    appears once per display name it ever used, and we merge on the address.
  - subject_terms: tokenizing/stopword-filtering subject lines, counting each
    term once per subject.
  - validate_accounts: the config gate. Every rejection case here is a typo
    that would otherwise silently produce a query matching nothing, or a
    draft written under the wrong identity.

Run: ./test_mailctl.py
"""

import io
import json
import os
import tempfile
from contextlib import redirect_stdout
from pathlib import Path
from unittest.mock import patch

# mailctl validates its account config at import time, so a valid one has to
# exist before the import below. Point it at a throwaway maildir tree.
_tmp = tempfile.TemporaryDirectory()
FIXTURE_ROOT = Path(_tmp.name)
(FIXTURE_ROOT / "mail" / "acct-a" / "Drafts").mkdir(parents=True)
(FIXTURE_ROOT / "mail" / "acct-b").mkdir(parents=True)
FIXTURE_CONFIG = FIXTURE_ROOT / "accounts.json"
FIXTURE_CONFIG.write_text(json.dumps({"accounts": {
    "acct-a": {"maildir": "acct-a", "address": "a@example.org",
               "drafts": "Drafts"},
    "acct-b": {"maildir": "acct-b", "address": "b@example.org",
               "drafts": None},
}}))
os.environ["MAILCTL_CONFIG"] = str(FIXTURE_CONFIG)
os.environ["MAILCTL_MAIL_ROOT"] = str(FIXTURE_ROOT / "mail")

import mailctl  # noqa: E402  (must follow the env vars above)

MAIL = FIXTURE_ROOT / "mail"


def senders_json(raw, **overrides):
    """Run cmd_senders against a canned notmuch reply, return parsed JSON."""
    opts = {"query": "*", "account": None, "top": None, "json": True}
    opts.update(overrides)
    args = type("Args", (), opts)()
    buf = io.StringIO()
    with patch.object(mailctl, "run_notmuch", return_value=json.dumps(raw)), \
            redirect_stdout(buf):
        mailctl.cmd_senders(args)
    return json.loads(buf.getvalue())


def test_merges_on_address_keeping_longest_name():
    got = senders_json([
        {"name": "", "address": "a@x", "count": 112},
        {"name": "DPReview", "address": "a@x", "count": 353},
        {"name": "B", "address": "b@x", "count": 5},
    ])
    assert got == [
        {"address": "a@x", "name": "DPReview", "count": 465},
        {"address": "b@x", "name": "B", "count": 5},
    ], got


def test_top_truncates_after_sorting():
    got = senders_json([
        {"name": "small", "address": "s@x", "count": 1},
        {"name": "big", "address": "b@x", "count": 99},
    ], top=1)
    assert [a["address"] for a in got] == ["b@x"], got


def test_empty():
    assert senders_json([]) == []


def test_subject_terms_counts_each_term_once_per_subject():
    # "promo" three times in one subject must not outrank "sconto" in two
    got = mailctl.subject_terms(["Promo promo PROMO", "sconto", "Sconto!"])
    assert got["promo"] == 1, got
    assert got["sconto"] == 2, got


def test_subject_terms_drops_stopwords_digits_and_emoji():
    got = mailctl.subject_terms(["🔥 Le offerte di oggi for you 2024 ⏳"])
    assert set(got) == {"offerte"}, got


def test_subject_terms_keeps_accented_words():
    got = mailctl.subject_terms(["Località e novità"])
    assert set(got) == {"località", "novità"}, got


def test_subject_terms_empty():
    assert mailctl.subject_terms([]) == {}
    assert mailctl.subject_terms(["", "🔥"]) == {}


def rejects(accounts_value, expect_in_message):
    """Assert a config is refused, and that the message names the problem.

    Checking the message matters as much as the exit: these fire on the user's
    own typo, and a rejection that doesn't say which account and which field
    is barely better than a silent wrong answer.
    """
    buf = io.StringIO()
    try:
        with patch("sys.stderr", buf):
            mailctl.validate_accounts(accounts_value, MAIL)
    except SystemExit as e:
        assert e.code == 2, f"expected exit 2, got {e.code}"
        msg = buf.getvalue()
        assert expect_in_message in msg, f"want {expect_in_message!r} in:\n{msg}"
        return
    raise AssertionError(f"config was accepted but should not be: {accounts_value}")


def good(**overrides):
    spec = {"maildir": "acct-a", "address": "a@example.org", "drafts": "Drafts"}
    spec.update(overrides)
    return {"accounts": {"acct-a": spec}}


def test_valid_config_returns_both_maps():
    accounts, drafts = mailctl.validate_accounts(json.loads(
        FIXTURE_CONFIG.read_text()), MAIL)
    assert accounts == {"acct-a": ("acct-a", "a@example.org"),
                        "acct-b": ("acct-b", "b@example.org")}, accounts
    assert drafts == {"acct-a": "Drafts", "acct-b": None}, drafts


def test_rejects_misspelled_maildir():
    # the typo this whole gate exists for: scoping would match zero mail
    rejects(good(maildir="acct-A"), "does not exist")


def test_rejects_misspelled_drafts_dir():
    # would write a draft into a folder mbsync never syncs back
    rejects(good(drafts="Bozze"), "does not exist")


def test_rejects_absolute_and_traversing_maildir():
    rejects(good(maildir="/etc"), "plain subdirectory")
    rejects(good(maildir="../../etc"), "plain subdirectory")


def test_rejects_bad_address():
    rejects(good(address="a@example"), "not a valid email address")
    rejects(good(address="not-an-address"), "not a valid email address")


def test_rejects_unknown_field():
    # catches "addresss"/"maildirs" style typos instead of ignoring them
    rejects(good(adress="a@example.org"), "unknown field")


def test_rejects_missing_required_field():
    spec = good()
    del spec["accounts"]["acct-a"]["address"]
    rejects(spec, 'missing required field "address"')


def test_rejects_duplicate_maildir():
    rejects({"accounts": {
        "one": {"maildir": "acct-a", "address": "a@example.org", "drafts": None},
        "two": {"maildir": "acct-a", "address": "b@example.org", "drafts": None},
    }}, "is used by 2 accounts")


def test_rejects_structural_problems():
    rejects([], "must be a JSON object")
    rejects({}, 'missing top-level "accounts" key')
    rejects({"accounts": {}}, "non-empty")
    rejects({"accounts": {"acct-a": "acct-a"}}, "must be an object")
    rejects({"accounts": {"Acct A": good()["accounts"]["acct-a"]}},
            "invalid name")


def test_message_names_the_offending_account():
    buf = io.StringIO()
    try:
        with patch("sys.stderr", buf):
            mailctl.validate_accounts({"accounts": {
                "fine": {"maildir": "acct-b", "address": "b@example.org"},
                "broken": {"maildir": "nope", "address": "c@example.org"},
            }}, MAIL)
    except SystemExit:
        pass
    assert '"broken"' in buf.getvalue(), buf.getvalue()


def test_drafts_null_is_allowed():
    accounts, drafts = mailctl.validate_accounts(good(drafts=None), MAIL)
    assert drafts == {"acct-a": None}, drafts


def test_drafts_field_is_optional():
    spec = good()
    del spec["accounts"]["acct-a"]["drafts"]
    _, drafts = mailctl.validate_accounts(spec, MAIL)
    assert drafts == {"acct-a": None}, drafts


def test_load_accounts_rejects_bad_json():
    bad = FIXTURE_ROOT / "bad.json"
    bad.write_text("{not json")
    buf = io.StringIO()
    try:
        with patch("sys.stderr", buf):
            mailctl.load_accounts(bad, MAIL)
    except SystemExit as e:
        assert e.code == 2
        assert "not valid JSON" in buf.getvalue(), buf.getvalue()
        return
    raise AssertionError("bad JSON was accepted")


def test_load_accounts_missing_file_explains_how_to_create_it():
    buf = io.StringIO()
    try:
        with patch("sys.stderr", buf):
            mailctl.load_accounts(FIXTURE_ROOT / "absent.json", MAIL)
    except SystemExit as e:
        assert e.code == 2
        assert "no config at" in buf.getvalue(), buf.getvalue()
        return
    raise AssertionError("missing config was accepted")


if __name__ == "__main__":
    for name, fn in sorted(globals().items()):
        if name.startswith("test_") and callable(fn):
            fn()
    print("ok")