smart-org/apps/api/tests/test_invites.py

"""Invite redemption: the link an admin hands out must actually work.

Covers the full loop — create the invite, preview it as the invitee, redeem it,
and the ways redemption can legitimately fail. Also pins the absolute invite
URL, since a bare path is not clickable for the person receiving it.

Because signup is closed, the invite link doubles as the account-creation
path: a brand-new invitee redeems it with a password and gets an account bound
to the invited email. The token can never reset an existing account's password
— that is what the "account already exists" rejection guards.
"""

from __future__ import annotations

from httpx import AsyncClient

from tests.conftest import auth_header, signup


async def _owner_with_org(client: AsyncClient, name: str = "Acme") -> tuple[str, str]:
    """``(token, org_id)`` for a freshly created manager and their organization."""
    body = await signup(client, "owner@example.com", name="Owner")
    headers = auth_header(body["access_token"])
    org = await client.post("/api/organizations", json={"name": name}, headers=headers)
    assert org.status_code == 201, org.text
    return body["access_token"], org.json()["id"]


async def _invite(client: AsyncClient, token: str, org_id: str, email: str, role: str) -> dict:
    response = await client.post(
        f"/api/organizations/{org_id}/invites",
        json={"email": email, "role": role},
        headers=auth_header(token),
    )
    assert response.status_code == 201, response.text
    return response.json()


async def test_invite_url_is_absolute_and_points_at_the_accept_screen(client: AsyncClient):
    token, org_id = await _owner_with_org(client)
    invite = await _invite(client, token, org_id, "newbie@example.com", "member")

    # Relative would resolve against whatever origin the link is pasted on.
    assert invite["invite_url"].startswith("http")
    assert invite["invite_url"].endswith(f"/accept-invite?token={invite['invite_token']}")


async def test_invitee_can_preview_then_accept(client: AsyncClient):
    owner_token, org_id = await _owner_with_org(client)
    invite = await _invite(client, owner_token, org_id, "newbie@example.com", "manager")
    raw = invite["invite_token"]

    invitee = await signup(client, "newbie@example.com", name="Newbie")
    headers = auth_header(invitee["access_token"])

    preview = await client.get(f"/api/invites/{raw}", headers=headers)
    assert preview.status_code == 200, preview.text
    assert preview.json()["organization_name"] == "Acme"
    assert preview.json()["role"] == "manager"
    assert preview.json()["already_member"] is False

    accepted = await client.post("/api/invites/accept", json={"token": raw}, headers=headers)
    assert accepted.status_code == 200, accepted.text
    assert accepted.json()["organization_id"] == org_id

    me = await client.get("/api/auth/me", headers=headers)
    joined = [o for o in me.json()["organizations"] if o["id"] == org_id]
    assert joined and joined[0]["role"] == "manager"


async def test_invite_token_is_single_use(client: AsyncClient):
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
        "invite_token"
    ]

    invitee = await signup(client, "newbie@example.com")
    headers = auth_header(invitee["access_token"])
    first = await client.post("/api/invites/accept", json={"token": raw}, headers=headers)
    assert first.status_code == 200

    second = await client.post("/api/invites/accept", json={"token": raw}, headers=headers)
    assert second.status_code == 409
    assert second.json()["error"]["code"] == "conflict"


async def test_invite_cannot_be_redeemed_by_a_different_account(client: AsyncClient):
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
        "invite_token"
    ]

    intruder = await signup(client, "intruder@example.com")
    headers = auth_header(intruder["access_token"])

    response = await client.get(f"/api/invites/{raw}", headers=headers)
    assert response.status_code == 403
    assert response.json()["error"]["details"]["invite_email"] == "newbie@example.com"

    # The token is not burned by the failed attempt — the real invitee can
    # still use it.
    invitee = await signup(client, "newbie@example.com")
    ok = await client.post(
        "/api/invites/accept",
        json={"token": raw},
        headers=auth_header(invitee["access_token"]),
    )
    assert ok.status_code == 200


async def test_unknown_invite_token_is_not_found(client: AsyncClient):
    response = await client.get("/api/invites/definitely-not-a-real-token")
    assert response.status_code == 404


async def test_invite_preview_works_without_a_session(client: AsyncClient):
    """A brand-new invitee must see what they were invited to *before* they can
    log in — otherwise the login button on the accept screen is a dead end."""
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
        "invite_token"
    ]

    preview = await client.get(f"/api/invites/{raw}")
    assert preview.status_code == 200, preview.text
    body = preview.json()
    assert body["organization_name"] == "Acme"
    assert body["email"] == "newbie@example.com"
    assert body["role"] == "member"
    # No account exists for the invited email yet, so the UI offers to create
    # one instead of demanding a login.
    assert body["has_account"] is False
    assert body["already_member"] is False


async def test_invite_preview_reports_an_existing_account(client: AsyncClient):
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "existing@example.com", "member"))[
        "invite_token"
    ]
    await signup(client, "existing@example.com", name="Existing")

    preview = await client.get(f"/api/invites/{raw}")
    assert preview.status_code == 200, preview.text
    assert preview.json()["has_account"] is True


async def test_anonymous_redemption_creates_the_account_and_joins(client: AsyncClient):
    """The closed-signup flow: the link is the capability, the password makes
    the account private, and the org membership is created in one step."""
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
        "invite_token"
    ]

    accepted = await client.post(
        "/api/invites/accept", json={"token": raw, "password": "password123"}
    )
    assert accepted.status_code == 200, accepted.text
    assert accepted.json()["organization_id"] == org_id
    assert accepted.json()["role"] == "member"

    # The account exists for exactly the invited email, with that password, as
    # a plain platform user — an invite must never mint a manager.
    login = await client.post(
        "/api/auth/login", json={"email": "newbie@example.com", "password": "password123"}
    )
    assert login.status_code == 200, login.text
    token = login.json()["access_token"]
    assert login.json()["user"]["platform_role"] == "user"

    me = await client.get("/api/auth/me", headers=auth_header(token))
    joined = [o for o in me.json()["organizations"] if o["id"] == org_id]
    assert joined and joined[0]["role"] == "member"


async def test_anonymous_redemption_without_a_password_is_rejected(client: AsyncClient):
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
        "invite_token"
    ]

    response = await client.post("/api/invites/accept", json={"token": raw})
    assert response.status_code == 422
    assert response.json()["error"]["code"] == "validation_error"


async def test_anonymous_redemption_rejects_a_short_password(client: AsyncClient):
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "newbie@example.com", "member"))[
        "invite_token"
    ]

    response = await client.post(
        "/api/invites/accept", json={"token": raw, "password": "123"}
    )
    assert response.status_code == 422


async def test_a_link_cannot_reset_an_existing_account(client: AsyncClient):
    """A leaked invite must not become a password reset for the real account.

    Whoever holds the link can create an account *only if none exists* for the
    invited email; otherwise they must authenticate first.
    """
    owner_token, org_id = await _owner_with_org(client)
    raw = (await _invite(client, owner_token, org_id, "taken@example.com", "member"))[
        "invite_token"
    ]
    await signup(client, "taken@example.com", password="password123")

    response = await client.post(
        "/api/invites/accept", json={"token": raw, "password": "attacker-password"}
    )
    assert response.status_code == 403
    assert "وارد شوید" in response.json()["error"]["message"]

    # The real account's password is untouched, and the invite is not burned:
    # the invitee can still redeem it after signing in.
    login = await client.post(
        "/api/auth/login", json={"email": "taken@example.com", "password": "password123"}
    )
    assert login.status_code == 200
    accepted = await client.post(
        "/api/invites/accept",
        json={"token": raw},
        headers=auth_header(login.json()["access_token"]),
    )
    assert accepted.status_code == 200, accepted.text