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

"""The platform hierarchy: one owner, owner-created managers, invited tenants.

This is the security core of the identity model, so every rule gets both sides
tested — the allow and the deny:

1. **Exactly one owner.** The database refuses a second owner row outright;
   the API refuses to mint or demote one. The seat moves only through a
   transactional transfer that demotes the previous owner in the same step.
2. **Only the owner and managers create organizations.** A plain user may not.
3. **Only the owner creates managers.** No manager can mint another manager,
   and no manager can enter the admin console.
4. **A departing manager's organizations move to the owner** — never orphaned,
   never deleted with their data.
"""

from __future__ import annotations

import pytest
from httpx import AsyncClient
from sqlalchemy import select

from app.domain.identity.models import Role
from app.infrastructure.db.models import MembershipRow, OrganizationRow, UserRow
from tests.conftest import auth_header


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


async def _login(client: AsyncClient, email: str, password: str) -> str:
    response = await client.post(
        "/api/auth/login", json={"email": email, "password": password}
    )
    assert response.status_code == 200, response.text
    return response.json()["access_token"]


async def _owner_creates_user(
    client: AsyncClient,
    owner_token: str,
    email: str,
    *,
    platform_role: str = "manager",
    name: str = "",
) -> dict:
    response = await client.post(
        "/api/admin/users",
        json={"email": email, "name": name, "platform_role": platform_role},
        headers=auth_header(owner_token),
    )
    assert response.status_code == 201, response.text
    return response.json()


# ---------------------------------------------------------------------------
# Exactly one owner
# ---------------------------------------------------------------------------


async def test_only_one_platform_owner_may_exist(
    client: AsyncClient, platform_owner_token: str, session_factory
):
    """The invariant is enforced by the database, not by habit.

    Even a direct INSERT — the kind a future code path or a migration would
    perform — cannot produce a second owner row.
    """
    from sqlalchemy.exc import IntegrityError as _IntegrityError

    async with session_factory() as session:
        session.add(
            UserRow(
                id="second-owner-row",
                email="second-owner@example.com",
                password_hash="x",
                platform_role="owner",
            )
        )
        with pytest.raises(_IntegrityError):
            await session.flush()
        await session.rollback()


async def test_owner_cannot_be_created_through_the_api(
    client: AsyncClient, platform_owner_token: str
):
    response = await client.post(
        "/api/admin/users",
        json={"email": "wannabe@example.com", "platform_role": "owner"},
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 422
    assert "مالکیت" in response.json()["error"]["message"]


async def test_owner_account_is_immutable_from_the_api(
    client: AsyncClient, platform_owner_token: str
):
    """No API path may demote, deactivate or delete the single owner seat."""
    me = await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
    owner_id = me.json()["user"]["id"]

    demote = await client.patch(
        f"/api/admin/users/{owner_id}",
        json={"platform_role": "manager"},
        headers=auth_header(platform_owner_token),
    )
    assert demote.status_code == 403

    deactivate = await client.patch(
        f"/api/admin/users/{owner_id}",
        json={"is_active": False},
        headers=auth_header(platform_owner_token),
    )
    assert deactivate.status_code == 403

    delete = await client.delete(
        f"/api/admin/users/{owner_id}", headers=auth_header(platform_owner_token)
    )
    assert delete.status_code == 403


async def test_transfer_ownership_moves_the_seat_and_keeps_exactly_one(
    client: AsyncClient, platform_owner_token: str, session_factory
):
    """The safety valve for a lost owner account — transactional by design."""
    manager = await _owner_creates_user(
        client, platform_owner_token, "successor@example.com", name="جانشین"
    )
    successor_token = await _login(
        client, "successor@example.com", manager["temp_password"]
    )

    response = await client.post(
        "/api/admin/transfer-ownership",
        json={"user_id": manager["user"]["id"]},
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 200, response.text
    assert response.json()["platform_role"] == "owner"

    async with session_factory() as session:
        owners = (
            (await session.execute(select(UserRow).where(UserRow.platform_role == "owner")))
            .scalars()
            .all()
        )
    assert [o.email for o in owners] == ["successor@example.com"]

    # The previous owner is demoted to manager in the same step — a manager,
    # not a powerless user, so they keep the ability to run organizations.
    previous = await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
    assert previous.json()["user"]["platform_role"] == "manager"

    # And the successor really is the owner: the console opens for them.
    console = await client.get("/api/admin/users", headers=auth_header(successor_token))
    assert console.status_code == 200


# ---------------------------------------------------------------------------
# Organization creation is gated by platform role
# ---------------------------------------------------------------------------


async def test_plain_user_cannot_create_an_organization(
    client: AsyncClient, platform_owner_token: str, plain_user_token: str
):
    response = await client.post(
        "/api/organizations",
        json={"name": "سازمان ممنوعه"},
        headers=auth_header(plain_user_token),
    )
    assert response.status_code == 403
    assert "دعوت" in response.json()["error"]["message"]


async def test_manager_can_create_an_organization(client: AsyncClient, manager_token: str):
    org = await _create_org(client, manager_token, "سازمان مدیر")
    assert org["owner_user_id"] == (
        await client.get("/api/auth/me", headers=auth_header(manager_token))
    ).json()["user"]["id"]


async def test_owner_can_create_an_organization(client: AsyncClient, platform_owner_token: str):
    org = await _create_org(client, platform_owner_token, "سازمان مالک")
    assert org["name"] == "سازمان مالک"


# ---------------------------------------------------------------------------
# Only the owner creates managers
# ---------------------------------------------------------------------------


async def test_manager_cannot_create_accounts(client: AsyncClient, manager_token: str):
    response = await client.post(
        "/api/admin/users",
        json={"email": "sneaky@example.com", "platform_role": "manager"},
        headers=auth_header(manager_token),
    )
    assert response.status_code == 403


async def test_plain_user_cannot_create_accounts(client: AsyncClient, plain_user_token: str):
    response = await client.post(
        "/api/admin/users",
        json={"email": "sneaky2@example.com", "platform_role": "manager"},
        headers=auth_header(plain_user_token),
    )
    assert response.status_code == 403


async def test_manager_cannot_reach_the_admin_console(client: AsyncClient, manager_token: str):
    for path in ("/api/admin/users", "/api/admin/models", "/api/admin/organizations"):
        response = await client.get(path, headers=auth_header(manager_token))
        assert response.status_code == 403, path


async def test_owner_creates_a_manager_whose_temp_password_works(
    client: AsyncClient, platform_owner_token: str
):
    created = await _owner_creates_user(
        client, platform_owner_token, "manager1@example.com", name="مدیر یک"
    )
    assert created["user"]["platform_role"] == "manager"
    assert created["user"]["is_manager"] is True
    assert created["user"]["owned_organizations"] == 0
    temp = created["temp_password"]
    assert temp

    login = await client.post(
        "/api/auth/login",
        json={"email": "manager1@example.com", "password": temp},
    )
    assert login.status_code == 200, login.text
    assert login.json()["user"]["platform_role"] == "manager"


async def test_duplicate_email_is_rejected(client: AsyncClient, platform_owner_token: str):
    await _owner_creates_user(client, platform_owner_token, "twin@example.com")
    response = await client.post(
        "/api/admin/users",
        json={"email": "twin@example.com", "platform_role": "manager"},
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 409


async def test_owner_can_create_plain_users(client: AsyncClient, platform_owner_token: str):
    created = await _owner_creates_user(
        client,
        platform_owner_token,
        "plain@example.com",
        platform_role="user",
        name="کاربر عادی",
    )
    assert created["user"]["platform_role"] == "user"
    assert created["user"]["is_manager"] is False


async def test_owner_resets_a_forgotten_temp_password(
    client: AsyncClient, platform_owner_token: str
):
    created = await _owner_creates_user(
        client, platform_owner_token, "forgetful@example.com", name="فراموشکار"
    )
    first = created["temp_password"]

    reset = await client.post(
        f"/api/admin/users/{created['user']['id']}/reset-password",
        headers=auth_header(platform_owner_token),
    )
    assert reset.status_code == 200, reset.text
    second = reset.json()["temp_password"]
    assert second and second != first

    stale = await client.post(
        "/api/auth/login", json={"email": "forgetful@example.com", "password": first}
    )
    assert stale.status_code == 401
    fresh = await client.post(
        "/api/auth/login", json={"email": "forgetful@example.com", "password": second}
    )
    assert fresh.status_code == 200


# ---------------------------------------------------------------------------
# A departing manager's organizations move to the owner
# ---------------------------------------------------------------------------


async def test_demoting_a_manager_transfers_organizations_to_the_owner(
    client: AsyncClient, platform_owner_token: str, session_factory
):
    manager = await _owner_creates_user(
        client, platform_owner_token, "leaving@example.com", name="خداحافظ"
    )
    manager_token = await _login(
        client, "leaving@example.com", manager["temp_password"]
    )
    org = await _create_org(client, manager_token, "سازمان خداحافظ")

    demote = await client.patch(
        f"/api/admin/users/{manager['user']['id']}",
        json={"platform_role": "user"},
        headers=auth_header(platform_owner_token),
    )
    assert demote.status_code == 200, demote.text
    assert demote.json()["platform_role"] == "user"
    assert demote.json()["owned_organizations"] == 0

    async with session_factory() as session:
        row = await session.get(OrganizationRow, org["id"])
        assert row is not None
        owner_id = (
            await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
        ).json()["user"]["id"]
        assert row.owner_user_id == owner_id

        memberships = (
            (
                await session.execute(
                    select(MembershipRow).where(MembershipRow.organization_id == org["id"])
                )
            )
            .scalars()
            .all()
        )
        by_user = {m.user_id: m for m in memberships}
        # The new owner can actually see the tenant...
        assert by_user[owner_id].role == Role.OWNER.value
        # ...and the demoted manager keeps working access without ownership.
        assert by_user[manager["user"]["id"]].role == Role.ADMIN.value


async def test_deleting_a_manager_transfers_organizations_and_memberships(
    client: AsyncClient, platform_owner_token: str, session_factory
):
    manager = await _owner_creates_user(
        client, platform_owner_token, "doomed@example.com", name="حذف‌شونده"
    )
    manager_token = await _login(client, "doomed@example.com", manager["temp_password"])
    org = await _create_org(client, manager_token, "سازمان حذف‌شونده")

    response = await client.delete(
        f"/api/admin/users/{manager['user']['id']}", headers=auth_header(platform_owner_token)
    )
    assert response.status_code == 200, response.text
    body = response.json()
    assert body["deleted_user_id"] == manager["user"]["id"]
    assert [t["organization_id"] for t in body["transferred_organizations"]] == [org["id"]]

    async with session_factory() as session:
        assert await session.get(UserRow, manager["user"]["id"]) is None
        row = await session.get(OrganizationRow, org["id"])
        assert row is not None, "the organization must survive its owner's deletion"
        owner_id = (
            await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
        ).json()["user"]["id"]
        assert row.owner_user_id == owner_id
        stale = (
            await session.execute(
                select(MembershipRow).where(
                    MembershipRow.user_id == manager["user"]["id"]
                )
            )
        ).scalars().all()
        assert stale == [], "the deleted account's memberships must be gone"


# ---------------------------------------------------------------------------
# Managers create and join organizations
# ---------------------------------------------------------------------------


async def test_manager_can_join_another_managers_organization_by_invite(
    client: AsyncClient, platform_owner_token: str
):
    """The invited manager keeps their own organization and gains a second."""
    first = await _owner_creates_user(
        client, platform_owner_token, "first@example.com", name="مدیر یکم"
    )
    second = await _owner_creates_user(
        client, platform_owner_token, "second@example.com", name="مدیر دوم"
    )
    first_token = await _login(client, "first@example.com", first["temp_password"])
    second_token = await _login(client, "second@example.com", second["temp_password"])

    own_org = await _create_org(client, first_token, "سازمان مدیر یکم")
    # The second manager has a tenant of their own before being invited.
    second_own_org = await _create_org(client, second_token, "سازمان مدیر دوم")

    invite = await client.post(
        f"/api/organizations/{own_org['id']}/invites",
        json={"email": "second@example.com", "role": "admin"},
        headers=auth_header(first_token),
    )
    assert invite.status_code == 201, invite.text

    accepted = await client.post(
        "/api/invites/accept",
        json={"token": invite.json()["invite_token"]},
        headers=auth_header(second_token),
    )
    assert accepted.status_code == 200, accepted.text

    me = await client.get("/api/auth/me", headers=auth_header(second_token))
    roles = {o["id"]: o["role"] for o in me.json()["organizations"]}
    # Joined the other manager's organization with the invited role...
    assert roles[own_org["id"]] == "admin"
    # ...and still owns their own, untouched.
    assert roles[second_own_org["id"]] == "owner"
    assert len(me.json()["organizations"]) == 2


async def test_promoting_a_plain_user_to_manager_is_possible(
    client: AsyncClient, platform_owner_token: str
):
    created = await _owner_creates_user(
        client, platform_owner_token, "promoted@example.com", platform_role="user"
    )
    promoted = await client.patch(
        f"/api/admin/users/{created['user']['id']}",
        json={"platform_role": "manager"},
        headers=auth_header(platform_owner_token),
    )
    assert promoted.status_code == 200, promoted.text
    assert promoted.json()["platform_role"] == "manager"

    # The new tier takes effect immediately: organizations are now allowed.
    promoted_token = await _login(
        client, "promoted@example.com", created["temp_password"]
    )
    org = await _create_org(client, promoted_token, "سازمان ارتقایافته")
    assert org["name"] == "سازمان ارتقایافته"


async def test_deactivating_an_account_blocks_login(
    client: AsyncClient, platform_owner_token: str
):
    created = await _owner_creates_user(
        client, platform_owner_token, "frozen@example.com", name="مسدود"
    )
    response = await client.patch(
        f"/api/admin/users/{created['user']['id']}",
        json={"is_active": False},
        headers=auth_header(platform_owner_token),
    )
    assert response.status_code == 200, response.text
    assert response.json()["is_active"] is False

    login = await client.post(
        "/api/auth/login",
        json={"email": "frozen@example.com", "password": created["temp_password"]},
    )
    assert login.status_code == 401


async def test_user_list_reports_roles_and_owned_organization_counts(
    client: AsyncClient, platform_owner_token: str
):
    manager = await _owner_creates_user(
        client, platform_owner_token, "listed@example.com", name="فهرست"
    )
    manager_token = await _login(client, "listed@example.com", manager["temp_password"])
    await _create_org(client, manager_token, "سازمان فهرست")

    listing = await client.get("/api/admin/users", headers=auth_header(platform_owner_token))
    assert listing.status_code == 200
    rows = {u["email"]: u for u in listing.json()}
    assert rows["listed@example.com"]["platform_role"] == "manager"
    assert rows["listed@example.com"]["owned_organizations"] == 1
    me = await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
    owner_email = me.json()["user"]["email"]
    assert rows[owner_email]["platform_role"] == "owner"
    assert rows[owner_email]["is_platform_owner"] is True


async def test_admin_user_endpoints_require_authentication(client: AsyncClient):
    assert (await client.get("/api/admin/users")).status_code == 401
    assert (
        await client.post("/api/admin/users", json={"email": "x@example.com"})
    ).status_code == 401


async def test_every_seat_move_is_audited(
    client: AsyncClient, platform_owner_token: str, session_factory
):
    """The transfer and the deletion are the two irreversible moves; both must
    leave a trail that names the actor and the moved organizations."""
    from app.domain.audit.models import AuditAction

    manager = await _owner_creates_user(
        client, platform_owner_token, "audited@example.com", name="ممیزی"
    )
    manager_token = await _login(client, "audited@example.com", manager["temp_password"])
    org = await _create_org(client, manager_token, "سازمان ممیزی")

    response = await client.delete(
        f"/api/admin/users/{manager['user']['id']}", headers=auth_header(platform_owner_token)
    )
    assert response.status_code == 200

    from app.infrastructure.repositories.audit import AuditRepository

    async with session_factory() as session:
        events = await AuditRepository(session).list_platform(limit=100)
    actions = {e.action: e for e in events}
    assert AuditAction.USER_CREATE in actions
    assert AuditAction.USER_DELETE in actions
    delete_event = actions[AuditAction.USER_DELETE]
    assert delete_event.metadata_json["transferred_organizations"] == [org["id"]]
    assert delete_event.actor_user_id == (
        await client.get("/api/auth/me", headers=auth_header(platform_owner_token))
    ).json()["user"]["id"]