[US-11.9 / OBT-237] Let project managers manage member roles (not other managers)

app/api/projects/access.py

@@ -15,6 +15,7 @@
     ProjectUserAccessRoleUpdate,
 )
 from app.services import project_service
+from app.services.project import assert_can_grant_access, assert_can_modify_member_role
 
 router = APIRouter()
 
@@ -30,7 +31,7 @@ async def grant_user_access(
     actor: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ) -> ProjectUserAccessResponse:
-    await assert_project_access(db, actor, project_id)
+    await assert_can_grant_access(db, actor, project_id)
     await project_service.get_project_or_404(db, project_id)
     access = await project_service.grant_user_access(db, project_id, payload.user_id, payload.role)
     return ProjectUserAccessResponse.model_validate(access)
@@ -118,7 +119,7 @@ async def update_user_access_role(
     actor: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ) -> ProjectUserAccessResponse:
-    await assert_project_access(db, actor, project_id)
+    await assert_can_modify_member_role(db, actor, project_id, user_id)
     access = await project_service.update_user_access_role(db, project_id, user_id, payload.role)
     return ProjectUserAccessResponse.model_validate(access)
 
@@ -133,7 +134,7 @@ async def revoke_user_access(
     actor: User = Depends(get_current_user),
     db: AsyncSession = Depends(get_db),
 ) -> None:
-    await assert_project_access(db, actor, project_id)
+    await assert_can_modify_member_role(db, actor, project_id, user_id)
     await project_service.revoke_user_access(db, project_id, user_id)
 
 

app/models/project.py

@@ -41,7 +41,7 @@ class ProjectLocationUpdate(BaseModel):
 
 class ProjectGrantUserAccess(BaseModel):
     user_id: str
-    role: str = Field(default="member", max_length=30)
+    role: Literal["member", "manager"] = "member"
 
 
 class ProjectUserAccessRoleUpdate(BaseModel):

app/services/project/__init__.py

@@ -1,9 +1,13 @@
+from app.services.project.assert_can_grant_access import assert_can_grant_access
+from app.services.project.assert_can_modify_member_role import assert_can_modify_member_role
 from app.services.project.can_access_project import can_access_project
 from app.services.project.create_project import create_project
 from app.services.project.get_project_by_id import get_project_by_id
 from app.services.project.get_project_or_404 import get_project_or_404
+from app.services.project.get_user_project_access import get_user_project_access
 from app.services.project.grant_organization_access import grant_organization_access
 from app.services.project.grant_user_access import grant_user_access
+from app.services.project.is_project_manager import is_project_manager
 from app.services.project.list_all_projects import list_all_projects
 from app.services.project.list_project_organization_access import (
     list_project_organization_access,
@@ -24,12 +28,16 @@
 from app.services.project.update_user_access_role import update_user_access_role
 
 __all__ = [
+    "assert_can_grant_access",
+    "assert_can_modify_member_role",
     "can_access_project",
     "create_project",
     "get_project_by_id",
     "get_project_or_404",
+    "get_user_project_access",
     "grant_organization_access",
     "grant_user_access",
+    "is_project_manager",
     "list_all_projects",
     "list_project_organization_access",
     "list_project_user_access",

app/services/project/assert_can_grant_access.py

@@ -0,0 +1,12 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exceptions import AuthorizationError
+from app.db.models.auth import User
+from app.services.project.is_project_manager import is_project_manager
+
+
+async def assert_can_grant_access(db: AsyncSession, actor: User, project_id: str) -> None:
+    if actor.is_platform_admin:
+        return
+    if not await is_project_manager(db, actor.id, project_id):
+        raise AuthorizationError("You must be a manager of this project")

app/services/project/assert_can_modify_member_role.py

@@ -0,0 +1,21 @@
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.core.exceptions import AuthorizationError, NotFoundError
+from app.db.models.auth import User
+from app.db.models.org import MemberRole
+from app.services.project.get_user_project_access import get_user_project_access
+from app.services.project.is_project_manager import is_project_manager
+
+
+async def assert_can_modify_member_role(
+    db: AsyncSession, actor: User, project_id: str, target_user_id: str
+) -> None:
+    if actor.is_platform_admin:
+        return
+    if not await is_project_manager(db, actor.id, project_id):
+        raise AuthorizationError("You must be a manager of this project")
+    target = await get_user_project_access(db, project_id, target_user_id)
+    if target is None:
+        raise NotFoundError("User access not found for this project")
+    if target.role == MemberRole.MANAGER:
+        raise AuthorizationError("Managers cannot change or remove another manager")

app/services/project/get_user_project_access.py

@@ -0,0 +1,15 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.project import ProjectUserAccess
+
+
+async def get_user_project_access(
+    db: AsyncSession, project_id: str, user_id: str
+) -> ProjectUserAccess | None:
+    stmt = select(ProjectUserAccess).where(
+        ProjectUserAccess.project_id == project_id,
+        ProjectUserAccess.user_id == user_id,
+    )
+    result = await db.execute(stmt)
+    return result.scalar_one_or_none()

app/services/project/is_project_manager.py

@@ -0,0 +1,18 @@
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.db.models.org import MemberRole
+from app.db.models.project import ProjectUserAccess
+
+
+async def is_project_manager(db: AsyncSession, user_id: str, project_id: str) -> bool:
+    result = await db.execute(
+        select(ProjectUserAccess.id)
+        .where(
+            ProjectUserAccess.project_id == project_id,
+            ProjectUserAccess.user_id == user_id,
+            ProjectUserAccess.role == MemberRole.MANAGER,
+        )
+        .limit(1)
+    )
+    return result.scalar_one_or_none() is not None

tests/baker.py

@@ -242,8 +242,10 @@ async def make_project_user_access(
     db: AsyncSession,
     project_id: str,
     user_id: str,
+    *,
+    role: str = "member",
 ) -> ProjectUserAccess:
-    access = ProjectUserAccess(project_id=project_id, user_id=user_id)
+    access = ProjectUserAccess(project_id=project_id, user_id=user_id, role=role)
     db.add(access)
     await db.commit()
     await db.refresh(access)
@@ -596,7 +598,6 @@ async def grant_app_role(
     role_key: str = "user",
     label: str | None = None,
 ) -> UserAppRole:
-    """Convenience: ensure a role with `role_key` exists for `app` and assign it to `user`."""
     role = await make_role(
         db,
         app.id,

tests/test_project_access_roles.py

@@ -0,0 +1,110 @@
+import pytest
+
+from app.core.exceptions import AuthorizationError, NotFoundError
+from app.services import project_service
+from tests.baker import (
+    make_language,
+    make_project,
+    make_project_user_access,
+    make_user,
+)
+
+
+async def _project(db):
+    lang = await make_language(db, code="prj")
+    return await make_project(db, language_id=lang.id)
+
+
+@pytest.mark.asyncio
+async def test_is_project_manager(db_session) -> None:
+    project = await _project(db_session)
+    manager = await make_user(db_session, email="[email protected]")
+    member = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, manager.id, role="manager")
+    await make_project_user_access(db_session, project.id, member.id, role="member")
+
+    assert await project_service.is_project_manager(db_session, manager.id, project.id) is True
+    assert await project_service.is_project_manager(db_session, member.id, project.id) is False
+
+
+@pytest.mark.asyncio
+async def test_grant_access_allows_admin_and_manager(db_session) -> None:
+    project = await _project(db_session)
+    admin = await make_user(db_session, email="[email protected]", is_platform_admin=True)
+    manager = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, manager.id, role="manager")
+
+    await project_service.assert_can_grant_access(db_session, admin, project.id)
+    await project_service.assert_can_grant_access(db_session, manager, project.id)
+
+
+@pytest.mark.asyncio
+async def test_grant_access_forbidden_for_member(db_session) -> None:
+    project = await _project(db_session)
+    member = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, member.id, role="member")
+
+    with pytest.raises(AuthorizationError, match="manager of this project"):
+        await project_service.assert_can_grant_access(db_session, member, project.id)
+
+
+@pytest.mark.asyncio
+async def test_manager_can_modify_a_member(db_session) -> None:
+    project = await _project(db_session)
+    manager = await make_user(db_session, email="[email protected]")
+    member = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, manager.id, role="manager")
+    await make_project_user_access(db_session, project.id, member.id, role="member")
+
+    await project_service.assert_can_modify_member_role(db_session, manager, project.id, member.id)
+
+
+@pytest.mark.asyncio
+async def test_manager_cannot_modify_another_manager(db_session) -> None:
+    project = await _project(db_session)
+    manager = await make_user(db_session, email="[email protected]")
+    other_manager = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, manager.id, role="manager")
+    await make_project_user_access(db_session, project.id, other_manager.id, role="manager")
+
+    with pytest.raises(AuthorizationError, match="another manager"):
+        await project_service.assert_can_modify_member_role(
+            db_session, manager, project.id, other_manager.id
+        )
+
+
+@pytest.mark.asyncio
+async def test_admin_can_modify_a_manager(db_session) -> None:
+    project = await _project(db_session)
+    admin = await make_user(db_session, email="[email protected]", is_platform_admin=True)
+    manager = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, manager.id, role="manager")
+
+    await project_service.assert_can_modify_member_role(db_session, admin, project.id, manager.id)
+
+
+@pytest.mark.asyncio
+async def test_non_manager_cannot_modify_roles(db_session) -> None:
+    project = await _project(db_session)
+    member = await make_user(db_session, email="[email protected]")
+    target = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, member.id, role="member")
+    await make_project_user_access(db_session, project.id, target.id, role="member")
+
+    with pytest.raises(AuthorizationError, match="manager of this project"):
+        await project_service.assert_can_modify_member_role(
+            db_session, member, project.id, target.id
+        )
+
+
+@pytest.mark.asyncio
+async def test_modify_missing_target_raises_not_found(db_session) -> None:
+    project = await _project(db_session)
+    manager = await make_user(db_session, email="[email protected]")
+    ghost = await make_user(db_session, email="[email protected]")
+    await make_project_user_access(db_session, project.id, manager.id, role="manager")
+
+    with pytest.raises(NotFoundError, match="access not found"):
+        await project_service.assert_can_modify_member_role(
+            db_session, manager, project.id, ghost.id
+        )