siftlode/backend/tests/test_mail_language.py

"""E9.5 S7 — every outbound mail is written in the reader's language.

Two halves. The first is the catalog invariant: HU/EN parity with identical placeholders, enforced
here because a mail is the one user-facing surface nobody can re-render by switching the UI language
— it arrives once, in whatever language we picked. The second is the resolution chain: which
language a given recipient actually gets, and where that signal was captured, since for an address
with no account it exists only at the moment they asked us.
"""

import re
import string

import pytest
from sqlalchemy import select

from app import mailtext
from app.models import Invite, User
from app.routes import auth as auth_routes

_PW = "correct horse battery"


def _placeholders(template: str) -> set[str]:
    return {name for _, name, _, _ in string.Formatter().parse(template) if name}


# --- the catalog itself ---------------------------------------------------------------------------


@pytest.mark.parametrize("key", sorted(mailtext.MAIL_TEXT))
def test_every_mail_string_exists_in_both_languages(key):
    assert set(mailtext.MAIL_TEXT[key]) == set(mailtext.SUPPORTED)


@pytest.mark.parametrize("key", sorted(mailtext.MAIL_TEXT))
def test_every_translation_takes_the_same_placeholders(key):
    # The parity that actually bites: a translation that drops `{link}` renders a mail with no link
    # in it, and a translation that invents one raises KeyError inside a background task nobody is
    # watching. Both are silent in production and loud here.
    variants = mailtext.MAIL_TEXT[key]
    expected = _placeholders(variants[mailtext.DEFAULT])
    for lang, template in variants.items():
        assert _placeholders(template) == expected, f"{key} [{lang}]"


def _app_sources_without_the_catalog() -> str:
    """Every app source, with the catalog LITERAL cut out of mailtext.py.

    Cutting it is the whole point: the dict spells each key as a quoted literal, so a scan that
    included it reported every key as used no matter what the catalog held. Cutting the whole FILE
    is too much, though — `link_expiry` legitimately asks for `duration.*`, and those keys would
    then look orphaned."""
    from pathlib import Path

    app_dir = Path(__file__).resolve().parents[1] / "app"
    out = []
    for f in app_dir.rglob("*.py"):
        src = f.read_text(encoding="utf-8")
        if f.name == "mailtext.py":
            src = src[: src.index("MAIL_TEXT: dict")]
        out.append(src)
    return "\n".join(out)


def test_no_catalog_entry_is_orphaned():
    """Every key is referenced from the app, so a mail that stopped being sent can't leave its text
    behind to be 'kept up to date' by the next person who greps for it."""
    sources = _app_sources_without_the_catalog()
    assert [k for k in mailtext.MAIL_TEXT if f'"{k}"' not in sources] == []


def test_every_key_the_app_asks_for_exists():
    """The other direction, and the one that raises in production: `text()` looks the key up in a
    dict, so a renamed or mistyped key is a KeyError inside a background task with no user in sight."""
    asked = set(re.findall(r'text\(\s*"([a-z_]+\.[a-z_]+)"', _app_sources_without_the_catalog()))
    assert asked, "the scan found no mailtext.text(...) call sites — the pattern has rotted"
    assert sorted(asked - set(mailtext.MAIL_TEXT)) == []


def test_an_unknown_language_falls_back_to_english_instead_of_raising():
    # `preferences.language` is free-form JSON written by older builds and by locales we may stop
    # shipping. A send that raised on one would take out a background task with no user in sight.
    assert mailtext.text("test.subject", "de") == mailtext.text("test.subject", "en")
    assert mailtext.text("test.subject", None) == mailtext.text("test.subject", "en")


# --- who gets which language ----------------------------------------------------------------------


def test_a_users_own_preference_decides_their_language(make_user):
    hu = make_user(email="hu@example.com", preferences={"language": "hu"})
    en = make_user(email="en@example.com", preferences={"language": "en"})
    none = make_user(email="none@example.com", preferences=None)

    assert mailtext.lang_for_user(hu) == "hu"
    assert mailtext.lang_for_user(en) == "en"
    assert mailtext.lang_for_user(none) == "en"


def test_an_address_with_no_account_falls_back_to_the_language_it_asked_in(db):
    db.add(Invite(email="asked@example.com", status="pending", lang="hu"))
    db.commit()

    assert mailtext.lang_for_email(db, "asked@example.com") == "hu"


def test_the_account_wins_over_the_invite_row(db, make_user):
    make_user(email="both@example.com", preferences={"language": "en"})
    db.add(Invite(email="both@example.com", status="approved", lang="hu"))
    db.commit()

    # The invite records how they asked once; the account records what they have since chosen.
    assert mailtext.lang_for_email(db, "both@example.com") == "en"


def test_a_row_that_never_recorded_a_language_is_walked_past(db, make_user):
    # Prod carries TWO accounts for one Gmail mailbox (the 2026-09-03 pair). Stopping at the first
    # row would answer English for a person whose other account says Hungarian.
    make_user(email="dup.one@gmail.com", preferences={})
    make_user(email="dupone@gmail.com", preferences={"language": "hu"})

    assert mailtext.lang_for_email(db, "dup.one@gmail.com") == "hu"


def test_the_lookup_is_canonical_like_every_other_identity_match(db, make_user):
    make_user(email="canon.match@gmail.com", preferences={"language": "hu"})

    assert mailtext.lang_for_email(db, "canonmatch+tag@gmail.com") == "hu"


# --- capturing the signal at the one moment it exists ---------------------------------------------


def _ask(db, lang: str) -> None:
    """A pending request carrying the language it was made in — what a refused Google sign-in records
    (with the locale Google reported), through the one function every entry point shares."""
    auth_routes.upsert_pending_invite(db, "asking@example.com", lang)


def test_registration_records_the_language_of_the_sign_up_page(db, client):
    r = client.post(
        "/auth/register", json={"email": "signup@example.com", "password": _PW, "lang": "hu"}
    )

    assert r.status_code == 200
    user = db.execute(select(User).where(User.email == "signup@example.com")).scalar_one()
    assert (user.preferences or {}).get("language") == "hu"


def test_an_access_request_records_the_language_it_was_made_in(db):
    _ask(db, "hu")

    inv = db.execute(select(Invite).where(Invite.email == "asking@example.com")).scalar_one()
    assert inv.lang == "hu"


def test_asking_again_from_another_language_updates_the_row(db):
    _ask(db, "en")
    _ask(db, "hu")

    db.expire_all()
    inv = db.execute(select(Invite).where(Invite.email == "asking@example.com")).scalar_one()
    # The newest thing they told us wins — and there is still exactly one row (S3's canonical upsert).
    assert inv.lang == "hu"
    assert db.execute(select(Invite)).scalars().all() == [inv]


def test_an_unsupported_language_is_ignored_rather_than_stored(db):
    _ask(db, "../../etc")

    inv = db.execute(select(Invite).where(Invite.email == "asking@example.com")).scalar_one()
    assert inv.lang is None


# --- the mails themselves -------------------------------------------------------------------------


def _bodies(monkeypatch):
    """Capture what `_send` would have put on the wire, rather than the arguments a caller passed —
    the language only matters if it reaches the body."""
    from app import email as email_mod

    out: list[tuple[list[str], str, str]] = []
    monkeypatch.setattr(
        email_mod,
        "_send",
        lambda to, subject, body, reply_to=None: out.append((to, subject, body)) is None,
    )
    return out


def test_the_confirmation_mail_is_written_in_the_accounts_language(db, client, monkeypatch):
    from app import email as email_mod

    monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)
    sent = _bodies(monkeypatch)

    assert (
        client.post(
            "/auth/register", json={"email": "hu@example.com", "password": _PW, "lang": "hu"}
        ).status_code
        == 200
    )

    (_, subject, body) = sent[0]
    assert subject == mailtext.MAIL_TEXT["verify_email.subject"]["hu"]
    assert "#verify=" in body


def test_the_approval_mail_answers_in_the_language_the_request_was_made_in(
    db, make_user, as_user, monkeypatch
):
    from app import email as email_mod

    monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)
    sent = _bodies(monkeypatch)
    admin = make_user(email="admin@example.com", role="admin")
    _ask(db, "hu")
    inv = db.execute(select(Invite).where(Invite.email == "asking@example.com")).scalar_one()

    r = as_user(admin).post(f"/api/admin/invites/{inv.id}/approve")

    assert r.status_code == 200
    (to, subject, _) = sent[0]
    assert to == ["asking@example.com"]
    assert subject == mailtext.MAIL_TEXT["access_approved.subject"]["hu"]


def test_a_denial_answers_in_that_language_too(db, make_user, as_user, monkeypatch):
    from app import email as email_mod

    monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)
    sent = _bodies(monkeypatch)
    admin = make_user(email="admin@example.com", role="admin")
    _ask(db, "hu")
    inv = db.execute(select(Invite).where(Invite.email == "asking@example.com")).scalar_one()
    sent.clear()

    assert as_user(admin).post(f"/api/admin/invites/{inv.id}/deny").status_code == 200

    assert sent[0][1] == mailtext.MAIL_TEXT["access_denied.subject"]["hu"]


def test_the_admins_pick_decides_an_invitation_and_is_remembered(
    db, make_user, as_user, monkeypatch
):
    from app import email as email_mod

    monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)
    sent = _bodies(monkeypatch)
    admin = make_user(email="admin@example.com", role="admin")
    # Hand-whitelisted: nobody has ever told us anything about this mailbox, which is exactly why
    # the send button asks.
    inv = Invite(email="stranger@example.com", status="approved")
    db.add(inv)
    db.commit()

    r = as_user(admin).post(f"/api/admin/invites/{inv.id}/invite", json={"lang": "hu"})

    assert r.status_code == 200
    assert sent[0][1] == mailtext.MAIL_TEXT["access_invitation.subject"]["hu"]
    db.expire_all()
    # Recorded, so a later mail to this address matches the one the admin already sent.
    assert db.get(Invite, inv.id).lang == "hu"


def test_an_invitation_without_a_pick_still_goes_out(db, make_user, as_user, monkeypatch):
    from app import email as email_mod

    monkeypatch.setattr(email_mod, "email_enabled", lambda db=None: True)
    sent = _bodies(monkeypatch)
    admin = make_user(email="admin@example.com", role="admin")
    inv = Invite(email="stranger@example.com", status="approved")
    db.add(inv)
    db.commit()

    r = as_user(admin).post(f"/api/admin/invites/{inv.id}/invite")

    assert r.status_code == 200
    assert sent[0][1] == mailtext.MAIL_TEXT["access_invitation.subject"]["en"]
    db.expire_all()
    # Nothing was picked, so nothing is remembered: the column means "what this person told us",
    # not "what we last defaulted to".
    assert db.get(Invite, inv.id).lang is None


def test_admins_are_notified_in_their_own_languages(db, make_user):
    make_user(email="hu.admin@example.com", role="admin", preferences={"language": "hu"})
    make_user(email="en.admin@example.com", role="admin", preferences={"language": "en"})

    groups = auth_routes.admin_notify_groups(db)

    # One mail per language, not one blast that has to be wrong for somebody.
    assert groups == {"hu": ["hu.admin@example.com"], "en": ["en.admin@example.com"]}