The 7 baselined Optional accesses in plex/stream.py were one fact the type did not carry: the initial spawn (replace_only=None) always supersedes, so only the fallback respawn can get None. An assert at the call site says so; the pyright baseline drops 195 -> 188. test_abuse_limits: the public-URL and YouTube enqueue tests stopped at the profile lookup (400 no_profile) and passed on that unrelated refusal. They now send an inline spec and assert the job is queued. The same shape — a request whose response is discarded — sat at 53 more places across the suite (all returned what they should, measured with a TestClient probe); each now asserts its status, and test_response_checked keeps it that way (seen red).
320 lines
12 KiB
Python
320 lines
12 KiB
Python
"""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 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, client):
|
|
assert (
|
|
client.post(
|
|
"/auth/request-access", json={"email": "asking@example.com", "lang": "hu"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
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, client):
|
|
assert (
|
|
client.post(
|
|
"/auth/request-access", json={"email": "asking@example.com", "lang": "en"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
assert (
|
|
client.post(
|
|
"/auth/request-access", json={"email": "asking@example.com", "lang": "hu"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
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, client):
|
|
assert (
|
|
client.post(
|
|
"/auth/request-access", json={"email": "asking@example.com", "lang": "../../etc"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
|
|
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, client, 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")
|
|
assert (
|
|
client.post(
|
|
"/auth/request-access", json={"email": "asking@example.com", "lang": "hu"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
inv = db.execute(select(Invite).where(Invite.email == "asking@example.com")).scalar_one()
|
|
sent.clear() # drop the admin notification; this test is about the reply to the requester
|
|
|
|
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, client, 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")
|
|
assert (
|
|
client.post(
|
|
"/auth/request-access", json={"email": "asking@example.com", "lang": "hu"}
|
|
).status_code
|
|
== 200
|
|
)
|
|
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"]}
|