bdlg2023-sms-back/app/tests/test_organizations.py

import uuid

import pytest
from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.orm import Session

from app.core import security
from app.main import app
from app.models import GlobalRole, Organization, OrgRole, User, UserOrganization
from app.tests.conftest import default_organization_id, default_organization_name

pytestmark = pytest.mark.asyncio

ALL_ORG_ROLES = [OrgRole.ORG_ADMIN, OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION]


class TestListOrganizations:
    async def test_requires_auth(self, client: AsyncClient):
        response = await client.get(app.url_path_for("list_organizations"))
        assert response.status_code == 401

    @pytest.mark.parametrize("role", ALL_ORG_ROLES)
    async def test_member_sees_own_org(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(app.url_path_for("list_organizations"), headers=headers)
        assert response.status_code == 200
        data = response.json()
        assert len(data) == 1
        assert data[0]["id"] == default_organization_id

    async def test_non_member_sees_nothing(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        _, headers = make_org_user(role=None)
        response = await client.get(app.url_path_for("list_organizations"), headers=headers)
        assert response.status_code == 200
        assert response.json() == []

    async def test_super_admin_sees_all_orgs(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        other_org = Organization(id=str(uuid.uuid4()), name="Other Org")
        session.add(other_org)
        session.commit()

        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.get(app.url_path_for("list_organizations"), headers=headers)
        assert response.status_code == 200
        ids = [o["id"] for o in response.json()]
        assert default_organization_id in ids
        assert other_org.id in ids


class TestCreateOrganization:
    async def test_requires_auth(self, client: AsyncClient):
        response = await client.post(
            app.url_path_for("create_organization"), json={"name": "New Org"}
        )
        assert response.status_code == 401

    @pytest.mark.parametrize("role", ALL_ORG_ROLES + [None])
    async def test_non_super_admin_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("create_organization"), headers=headers, json={"name": "New Org"}
        )
        assert response.status_code == 403

    async def test_super_admin_creates_organization(
        self, client: AsyncClient, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("create_organization"), headers=headers, json={"name": "New Org"}
        )
        assert response.status_code == 200
        data = response.json()
        assert data["name"] == "New Org"

        org = (
            session.execute(select(Organization).where(Organization.name == "New Org"))
            .scalars()
            .first()
        )
        assert org is not None

    async def test_duplicate_name_rejected(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("create_organization"),
            headers=headers,
            json={"name": default_organization_name},
        )
        assert response.status_code == 400

    async def test_missing_name_validation_error(self, client: AsyncClient, make_org_user):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("create_organization"), headers=headers, json={}
        )
        assert response.status_code == 422


class TestGetOrganization:
    async def test_requires_auth(self, client: AsyncClient, default_organization: Organization):
        response = await client.get(
            app.url_path_for("get_organization", organization_id=default_organization_id)
        )
        assert response.status_code == 401

    @pytest.mark.parametrize("role", ALL_ORG_ROLES)
    async def test_member_can_view(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.get(
            app.url_path_for("get_organization", organization_id=default_organization_id),
            headers=headers,
        )
        assert response.status_code == 200
        assert response.json()["id"] == default_organization_id

    async def test_non_member_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        _, headers = make_org_user(role=None)
        response = await client.get(
            app.url_path_for("get_organization", organization_id=default_organization_id),
            headers=headers,
        )
        assert response.status_code == 403

    async def test_super_admin_bypasses_membership(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.get(
            app.url_path_for("get_organization", organization_id=default_organization_id),
            headers=headers,
        )
        assert response.status_code == 200

    async def test_includes_memberships(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        _, headers = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        response = await client.get(
            app.url_path_for("get_organization", organization_id=default_organization_id),
            headers=headers,
        )
        assert response.status_code == 200
        memberships = response.json()["memberships"]
        assert len(memberships) == 1
        assert memberships[0]["role"] == OrgRole.RESPO_BENEVOLE.value

    async def test_not_found(self, client: AsyncClient, make_org_user):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.get(
            app.url_path_for("get_organization", organization_id=uuid.uuid4()), headers=headers
        )
        assert response.status_code == 404


class TestUpdateOrganization:
    @pytest.mark.parametrize("role", ALL_ORG_ROLES + [None])
    async def test_non_super_admin_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("update_organization", organization_id=default_organization_id),
            headers=headers,
            json={"name": "Renamed"},
        )
        assert response.status_code == 403

    async def test_super_admin_renames(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("update_organization", organization_id=default_organization_id),
            headers=headers,
            json={"name": "Renamed"},
        )
        assert response.status_code == 200
        org = session.get(Organization, default_organization_id)
        assert org.name == "Renamed"

    async def test_not_found(self, client: AsyncClient, make_org_user):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("update_organization", organization_id=uuid.uuid4()),
            headers=headers,
            json={"name": "Nope"},
        )
        assert response.status_code == 404


class TestDeleteOrganization:
    @pytest.mark.parametrize("role", ALL_ORG_ROLES + [None])
    async def test_non_super_admin_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.delete(
            app.url_path_for("delete_organization", organization_id=default_organization_id),
            headers=headers,
        )
        assert response.status_code == 403

    async def test_super_admin_deletes_and_cascades_projects(
        self, client: AsyncClient, default_project, make_org_user, session: Session
    ):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.delete(
            app.url_path_for("delete_organization", organization_id=default_organization_id),
            headers=headers,
        )
        assert response.status_code == 200
        assert session.get(Organization, default_organization_id) is None

        from app.models import Project

        assert session.get(Project, default_project.id) is None

    async def test_not_found(self, client: AsyncClient, make_org_user):
        _, headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.delete(
            app.url_path_for("delete_organization", organization_id=uuid.uuid4()), headers=headers
        )
        assert response.status_code == 404


class TestAddMember:
    @pytest.mark.parametrize("role", ALL_ORG_ROLES + [None])
    async def test_non_super_admin_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for("add_member", organization_id=default_organization_id),
            headers=headers,
            json={"user_id": str(uuid.uuid4()), "role": OrgRole.RESPO_BENEVOLE.value},
        )
        assert response.status_code == 403

    async def test_super_admin_adds_existing_user(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        target_user, _ = make_org_user(role=None)  # exists, no membership yet
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.post(
            app.url_path_for("add_member", organization_id=default_organization_id),
            headers=admin_headers,
            json={"user_id": target_user.id, "role": OrgRole.RESPO_COMMISSION.value},
        )
        assert response.status_code == 200
        data = response.json()
        assert data["role"] == OrgRole.RESPO_COMMISSION.value
        assert data["user"]["email"] == target_user.email

        membership = session.get(UserOrganization, (target_user.id, default_organization_id))
        assert membership is not None
        assert membership.role == OrgRole.RESPO_COMMISSION

    async def test_user_can_belong_to_multiple_orgs(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        """Confirms the multi-org membership model -- no unique(user_id) constraint."""
        target_user, _ = make_org_user(role=OrgRole.ORG_ADMIN)  # already in default_organization
        other_org = Organization(id=uuid.uuid4(), name="Second Org")
        session.add(other_org)
        session.commit()

        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("add_member", organization_id=other_org.id),
            headers=admin_headers,
            json={"user_id": target_user.id, "role": OrgRole.RESPO_BENEVOLE.value},
        )
        assert response.status_code == 200

        memberships = (
            session.execute(
                select(UserOrganization).where(UserOrganization.user_id == target_user.id)
            )
            .scalars()
            .all()
        )
        assert len(memberships) == 2

    async def test_unknown_user_id_not_found(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)
        response = await client.post(
            app.url_path_for("add_member", organization_id=default_organization_id),
            headers=admin_headers,
            json={"user_id": str(uuid.uuid4()), "role": OrgRole.RESPO_BENEVOLE.value},
        )
        assert response.status_code == 404

    async def test_duplicate_membership_rejected(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        target_user, _ = make_org_user(role=OrgRole.RESPO_BENEVOLE)  # already a member
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.post(
            app.url_path_for("add_member", organization_id=default_organization_id),
            headers=admin_headers,
            json={"user_id": target_user.id, "role": OrgRole.ORG_ADMIN.value},
        )
        assert response.status_code == 400

    async def test_organization_not_found(self, client: AsyncClient, make_org_user):
        target_user, _ = make_org_user(role=None)
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.post(
            app.url_path_for("add_member", organization_id=uuid.uuid4()),
            headers=admin_headers,
            json={"user_id": target_user.id, "role": OrgRole.RESPO_BENEVOLE.value},
        )
        assert response.status_code == 404


class TestUpdateMemberRole:
    @pytest.mark.parametrize("role", [OrgRole.RESPO_BENEVOLE, OrgRole.RESPO_COMMISSION, None])
    async def test_insufficient_role_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        target_user, _ = make_org_user(role=OrgRole.RESPO_COMMISSION)
        _, headers = make_org_user(role=role)
        response = await client.post(
            app.url_path_for(
                "update_member_role",
                organization_id=default_organization_id,
                user_id=target_user.id,
            ),
            headers=headers,
            json={"role": OrgRole.ORG_ADMIN.value},
        )
        assert response.status_code == 403

    async def test_org_admin_of_same_org_can_change_role(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        target_user, _ = make_org_user(role=OrgRole.RESPO_COMMISSION)
        _, org_admin_headers = make_org_user(role=OrgRole.ORG_ADMIN)

        response = await client.post(
            app.url_path_for(
                "update_member_role",
                organization_id=default_organization_id,
                user_id=target_user.id,
            ),
            headers=org_admin_headers,
            json={"role": OrgRole.RESPO_BENEVOLE.value},
        )
        assert response.status_code == 200
        assert response.json()["role"] == OrgRole.RESPO_BENEVOLE.value

        membership = session.get(UserOrganization, (target_user.id, default_organization_id))
        assert membership.role == OrgRole.RESPO_BENEVOLE

    async def test_org_admin_of_different_org_forbidden(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        """An org_admin has no authority over an organization they don't belong to."""
        other_org = Organization(id=uuid.uuid4(), name="Other Org")
        session.add(other_org)
        session.commit()

        target_user, _ = make_org_user(
            role=OrgRole.RESPO_COMMISSION
        )  # member of default_organization

        # org_admin, but of other_org, not default_organization
        outsider = User(
            id=uuid.uuid4(), email="outsider@test.com", hashed_password="x", name="Outsider"
        )
        session.add(outsider)
        session.commit()
        session.add(
            UserOrganization(
                user_id=outsider.id, organization_id=other_org.id, role=OrgRole.ORG_ADMIN
            )
        )
        session.commit()
        outsider_token = security.create_jwt_token(str(outsider.id), 60 * 60 * 24, refresh=False)[0]
        outsider_headers = {"Authorization": f"Bearer {outsider_token}"}

        response = await client.post(
            app.url_path_for(
                "update_member_role",
                organization_id=default_organization_id,
                user_id=target_user.id,
            ),
            headers=outsider_headers,
            json={"role": OrgRole.ORG_ADMIN.value},
        )
        assert response.status_code == 403

    async def test_super_admin_changes_role(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        target_user, _ = make_org_user(role=OrgRole.RESPO_COMMISSION)
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.post(
            app.url_path_for(
                "update_member_role",
                organization_id=default_organization_id,
                user_id=target_user.id,
            ),
            headers=admin_headers,
            json={"role": OrgRole.ORG_ADMIN.value},
        )
        assert response.status_code == 200
        assert response.json()["role"] == OrgRole.ORG_ADMIN.value

        membership = session.get(UserOrganization, (target_user.id, default_organization_id))
        assert membership.role == OrgRole.ORG_ADMIN

    async def test_membership_not_found(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        non_member, _ = make_org_user(role=None)
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.post(
            app.url_path_for(
                "update_member_role", organization_id=default_organization_id, user_id=non_member.id
            ),
            headers=admin_headers,
            json={"role": OrgRole.ORG_ADMIN.value},
        )
        assert response.status_code == 404


class TestRemoveMember:
    @pytest.mark.parametrize("role", ALL_ORG_ROLES + [None])
    async def test_non_super_admin_forbidden(
        self, client: AsyncClient, default_organization: Organization, make_org_user, role
    ):
        target_user, _ = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        _, headers = make_org_user(role=role)
        response = await client.delete(
            app.url_path_for(
                "remove_member", organization_id=default_organization_id, user_id=target_user.id
            ),
            headers=headers,
        )
        assert response.status_code == 403

    async def test_super_admin_removes_member(
        self,
        client: AsyncClient,
        default_organization: Organization,
        make_org_user,
        session: Session,
    ):
        target_user, _ = make_org_user(role=OrgRole.RESPO_BENEVOLE)
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.delete(
            app.url_path_for(
                "remove_member", organization_id=default_organization_id, user_id=target_user.id
            ),
            headers=admin_headers,
        )
        assert response.status_code == 200
        assert session.get(UserOrganization, (target_user.id, default_organization_id)) is None
        # user account itself is untouched
        assert session.get(User, target_user.id) is not None

    async def test_membership_not_found(
        self, client: AsyncClient, default_organization: Organization, make_org_user
    ):
        non_member, _ = make_org_user(role=None)
        _, admin_headers = make_org_user(role=None, global_role=GlobalRole.SUPER_ADMIN)

        response = await client.delete(
            app.url_path_for(
                "remove_member", organization_id=default_organization_id, user_id=non_member.id
            ),
            headers=admin_headers,
        )
        assert response.status_code == 404