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

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()

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

    assert created is None  # nothing new to notify an admin about
    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