siftlode/backend/tests/test_email_canonical.py

"""One mailbox, many spellings — the canonical-email identity rule (E9.5 S3).

Gmail delivers `x.y@gmail.com`, `xy@gmail.com` and `xy+tag@gmail.com` to ONE inbox. The invite gate
used to compare literal strings, so on 2026-09-03 an approved user's Google sign-in became a silent
SECOND access request: they had registered as `npeter83.temu@gmail.com`, and OAuth returned the
account's stored primary `npeter83temu@gmail.com`. The `+tag` case fails every time — a Google account
cannot contain `+`, so OAuth can only ever answer with the base address.

The rule is narrow on purpose: dots are significant on most providers, and the Gmail dot rule does not
hold for Google Workspace custom domains unless their admin enabled it.
"""

import pytest
from sqlalchemy import select, text

_UAT_PW = "correct horse battery"

from app.auth import is_allowed
from app.models import Invite, User
from app.utils import canonical_email

# (input, expected canonical) — the matrix the sprint promised, plus the shapes that must NOT fold.
CASES = [
    # Gmail: dots and +tags are noise…
    ("npeter83.temu@gmail.com", "npeter83temu@gmail.com"),
    ("npeter83temu@gmail.com", "npeter83temu@gmail.com"),
    ("n.p.e.t.e.r@gmail.com", "npeter@gmail.com"),
    ("user+siftlode@gmail.com", "user@gmail.com"),
    ("u.ser+a+b@gmail.com", "user@gmail.com"),
    ("USer.Name@GMail.COM", "username@gmail.com"),
    # …and googlemail is the same mailbox as gmail.
    ("user.name@googlemail.com", "username@gmail.com"),
    # Everyone else: lowercase ONLY. Dots stay significant.
    ("first.last@example.com", "first.last@example.com"),
    ("First.Last@Example.COM", "first.last@example.com"),
    ("first+tag@example.com", "first+tag@example.com"),
    # A Workspace domain is not Gmail, however Gmail-ish it looks.
    ("first.last@gmail.company.com", "first.last@gmail.company.com"),
    # Degenerate input must not explode or invent an address.
    ("", ""),
    (None, ""),
    ("not-an-email", "not-an-email"),
    (".@gmail.com", ".@gmail.com"),
]


@pytest.mark.parametrize("raw,expected", CASES)
def test_canonical_email(raw, expected):
    assert canonical_email(raw) == expected


def test_the_migration_sql_agrees_with_the_helper(db):
    """The 0079 backfill reimplements canonical_email in SQL (a migration must not import app code
    that keeps changing). That duplication is only safe while the two agree, so pin them together —
    a drift here means rows backfilled into a form the application will never look up."""
    import importlib.util
    from pathlib import Path

    path = Path(__file__).resolve().parents[1] / "alembic" / "versions" / "0079_email_canonical.py"
    spec = importlib.util.spec_from_file_location("m0079", path)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)

    sql = module._CANON_SQL.format(col=":addr")
    for raw, expected in CASES:
        if raw is None:
            continue  # the column is NOT NULL; the backfill skips NULL emails explicitly
        got = db.execute(text(f"SELECT {sql}"), {"addr": raw}).scalar_one()
        assert got == expected, f"SQL and helper disagree on {raw!r}: {got!r} != {expected!r}"


def test_the_column_is_kept_in_sync_by_writing_the_email(db):
    """No write site has to remember: the model validator derives it. If this breaks, every lookup
    below silently stops matching."""
    user = User(email="Dotted.Name+tag@GoogleMail.com")
    db.add(user)
    db.commit()
    assert user.email == "Dotted.Name+tag@GoogleMail.com"  # stored AS TYPED — mail goes here
    assert user.email_canonical == "dottedname@gmail.com"  # …matched on this

    user.email = "someone.else@example.com"
    db.commit()
    assert user.email_canonical == "someone.else@example.com"


def test_an_approval_covers_every_spelling_of_the_same_gmail_mailbox(db):
    # THE 2026-09-03 lockout, in one assertion: approved under the dotted spelling, signing in with
    # the primary Google returns.
    db.add(Invite(email="npeter83.temu@gmail.com", status="approved"))
    db.commit()

    assert is_allowed(db, "npeter83temu@gmail.com") is True
    assert is_allowed(db, "npeter83.temu@gmail.com") is True
    assert is_allowed(db, "npeter83.temu+throwaway@gmail.com") is True


def test_a_plus_tagged_registration_is_not_locked_out_after_approval(db):
    # The case that ALWAYS broke: a Google account cannot contain `+`, so OAuth can only return the
    # base address, and a literal match could never find the invite.
    db.add(Invite(email="someone+siftlode@gmail.com", status="approved"))
    db.commit()

    assert is_allowed(db, "someone@gmail.com") is True


def test_dots_stay_significant_outside_gmail(db):
    db.add(Invite(email="first.last@example.com", status="approved"))
    db.commit()

    assert is_allowed(db, "first.last@example.com") is True
    assert is_allowed(db, "firstlast@example.com") is False  # a DIFFERENT mailbox elsewhere


def test_an_approved_row_wins_over_a_denied_sibling(db):
    # Prod carries approved pairs for one mailbox (oldest 2026-06-16). The API cannot flip an approved
    # invite to denied, so approved-beside-denied means the admin approved that mailbox under another
    # spelling — not that they revoked it. Revoking is deleting the row or suspending the account.
    db.add(Invite(email="two.spellings@gmail.com", status="denied"))
    db.add(Invite(email="twospellings@gmail.com", status="approved"))
    db.commit()

    assert is_allowed(db, "twospellings@gmail.com") is True


def test_a_denied_mailbox_stays_denied(db):
    db.add(Invite(email="no.thanks@gmail.com", status="denied"))
    db.commit()

    assert is_allowed(db, "nothanks@gmail.com") is False


def test_a_second_spelling_does_not_open_a_second_access_request(db):
    """The other half of the 2026-09-03 report: the admin queue got a duplicate request for a user
    who was already approved."""
    from app.routes.auth import upsert_pending_invite

    db.add(Invite(email="already.approved@gmail.com", status="approved"))
    db.commit()

    req = upsert_pending_invite(db, "alreadyapproved@gmail.com")

    assert req.is_new is False  # nothing new to notify an admin about
    assert req.invite is not None and req.invite.email == "already.approved@gmail.com"
    assert len(db.execute(select(Invite)).scalars().all()) == 1


def test_manually_whitelisting_another_spelling_reuses_the_row(db):
    """How prod ended up with hartmann.peter1986@ AND hartmannpeter1986@ both approved on the same
    day: the admin typed the other spelling and got a second row instead of a no-op."""
    from app.routes.admin import add_invite, AddInviteIn

    admin = User(email="admin@example.com", role="admin", is_active=True, email_verified=True)
    db.add(admin)
    db.add(Invite(email="hartmann.peter1986@gmail.com", status="approved"))
    db.commit()

    add_invite(AddInviteIn(email="hartmannpeter1986@gmail.com"), admin=admin, db=db)

    rows = db.execute(select(Invite)).scalars().all()
    assert len(rows) == 1
    assert rows[0].email == "hartmann.peter1986@gmail.com"  # the original spelling is preserved


def _capture_reset_links(monkeypatch) -> list[str]:
    """Patch the reset mailer and collect the recipients it is asked to write to."""
    from app.routes import auth as auth_mod

    sent: list[str] = []
    monkeypatch.setattr(auth_mod.email_mod, "send_password_reset", lambda to, link: sent.append(to))
    return sent


# --- signing in and recovering an account, whichever spelling of it the person types -------------
#
# The three flows that look an account up BY TYPED ADDRESS used to match literally, so a Gmail user
# who registered `x.y@` and typed `xy@` met "Invalid email or password" at sign-in — and, worse,
# total silence from password-reset and verify-resend, which answer uniformly by design.


def test_the_password_works_with_any_spelling_of_the_same_mailbox(db):
    from app.auth import find_user_by_email

    make = User(
        email="t.e.st+tag@gmail.com", email_verified=True, is_active=True, password_hash="x"
    )
    db.add(make)
    db.commit()

    for typed in (
        "t.e.st+tag@gmail.com",  # as registered
        "test@gmail.com",  # the mailbox itself
        "t.e.st@gmail.com",  # dots only
        "TEST@googlemail.com",  # the alias domain
    ):
        assert find_user_by_email(db, typed) is make, typed


def test_an_unrelated_address_still_finds_nothing(db):
    from app.auth import find_user_by_email

    db.add(User(email="someone@gmail.com", email_verified=True, is_active=True))
    db.commit()

    assert find_user_by_email(db, "someoneelse@gmail.com") is None
    assert find_user_by_email(db, "some.one@example.com") is None  # dots matter off Gmail


def test_the_exact_spelling_wins_when_two_accounts_share_a_mailbox(db):
    # Prod carries such a pair (ids 13 and 14). Whoever types their own address in full must land on
    # their own account, not on whichever row happens to come first.
    from app.auth import find_user_by_email

    dotted = User(email="npeter83.temu@gmail.com", email_verified=True, is_active=True)
    plain = User(email="npeter83temu@gmail.com", email_verified=True, is_active=True)
    db.add_all([dotted, plain])
    db.commit()

    assert find_user_by_email(db, "npeter83.temu@gmail.com") is dotted
    assert find_user_by_email(db, "npeter83temu@gmail.com") is plain


def test_an_ambiguous_spelling_refuses_rather_than_guessing(db):
    # Neither spelling is exact, and both are the same mailbox. Handing a password sign-in one of
    # them at random would let someone into the wrong account, so this answers "no such account" —
    # the same uniform reply the caller already gives.
    from app.auth import find_user_by_email

    db.add(User(email="npeter83.temu@gmail.com", email_verified=True, is_active=True))
    db.add(User(email="npeter83temu@gmail.com", email_verified=True, is_active=True))
    db.commit()

    assert find_user_by_email(db, "n.peter83.temu+x@gmail.com") is None


def test_a_reset_link_goes_to_the_address_on_the_account(db, monkeypatch):
    """Typing another spelling must still reach the owner — and the mail must go to the address they
    registered with, tag and all, not to the one typed into the form."""
    from app.routes.auth import _send_reset_if_eligible

    db.add(
        User(
            email="me+siftlode@gmail.com",
            email_verified=True,
            is_active=True,
            password_hash="x",
        )
    )
    db.commit()
    sent = _capture_reset_links(monkeypatch)

    _send_reset_if_eligible("me@gmail.com")

    assert sent == ["me+siftlode@gmail.com"]


# --- a Google sign-in must not rewrite the address its owner chose ------------------------------


def test_a_google_sign_in_keeps_a_plus_tagged_address(db):
    """People register as `me+siftlode@gmail.com` so a later leak from this site is identifiable.
    Google returns the account's primary address, which can never contain `+`, so following it
    blindly destroyed the tag on the first Google sign-in — and, since sign-in matches the stored
    address first, stopped them signing in with the address they had chosen."""
    from app.routes.auth import _sync_google_email

    user = User(email="me+siftlode@gmail.com", email_verified=True, is_active=True)
    db.add(user)
    db.commit()

    _sync_google_email(db, user, "me@gmail.com")

    assert user.email == "me+siftlode@gmail.com"


def test_a_google_sign_in_follows_a_real_mailbox_change(db):
    # The case the sync exists for: the Google account genuinely moved to a different mailbox.
    from app.routes.auth import _sync_google_email

    user = User(email="old.address@gmail.com", email_verified=True, is_active=True)
    db.add(user)
    db.commit()

    _sync_google_email(db, user, "brand-new@example.com")

    assert user.email == "brand-new@example.com"


def test_a_mailbox_change_is_refused_when_another_account_owns_it(db):
    from app.routes.auth import _sync_google_email

    db.add(User(email="taken@example.com", email_verified=True, is_active=True))
    user = User(email="mover@gmail.com", email_verified=True, is_active=True)
    db.add(user)
    db.commit()

    _sync_google_email(db, user, "taken@example.com")

    assert user.email == "mover@gmail.com"  # unique constraint would have 500'd the sign-in


# --- showing BOTH addresses, once there are two -------------------------------------------------


def test_the_google_address_is_recorded_beside_the_registered_one(db):
    """Since the sign-in no longer overwrites the registered address, the account has two true
    addresses and only one was stored. This is the other half, for the UI to name."""
    from app.routes.auth import _sync_google_email

    user = User(email="me+siftlode@gmail.com", email_verified=True, is_active=True)
    db.add(user)
    db.commit()

    user.google_email = "me@gmail.com"
    _sync_google_email(db, user, "me@gmail.com")
    db.commit()

    assert user.email == "me+siftlode@gmail.com"  # still theirs
    assert user.google_email == "me@gmail.com"  # …and now we know the other one


def test_me_reports_the_google_address_only_when_it_differs(client, db, make_user):
    """The comparison lives in one place so no surface has to repeat it — and so the account menu
    stays exactly as it is for the majority, whose two addresses are the same string."""
    same = make_user(email="plain@gmail.com", password=_UAT_PW)
    same.google_email = "plain@gmail.com"
    db.commit()
    assert (
        client.post(
            "/auth/password-login", json={"email": "plain@gmail.com", "password": _UAT_PW}
        ).status_code
        == 200
    )
    assert client.get("/api/me").json()["google_email"] is None

    client.post("/auth/logout")
    tagged = make_user(email="me+siftlode@gmail.com", password=_UAT_PW)
    tagged.google_email = "me@gmail.com"
    db.commit()
    assert (
        client.post(
            "/auth/password-login", json={"email": "me+siftlode@gmail.com", "password": _UAT_PW}
        ).status_code
        == 200
    )
    assert client.get("/api/me").json()["google_email"] == "me@gmail.com"