diff --git a/backend/app/api/deps.py b/backend/app/api/deps.py index 6314e63e..0862b448 100644 --- a/backend/app/api/deps.py +++ b/backend/app/api/deps.py @@ -199,6 +199,53 @@ async def require_engineer_or_admin( ) +async def require_l1( + current_user: Annotated[User, Depends(get_current_active_user)] +) -> User: + """L1 tech exact-match (with super_admin bypass for support).""" + if current_user.is_super_admin: + return current_user + if current_user.account_role != "l1_tech": + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="L1 tech role required", + ) + return current_user + + +async def require_l1_or_coverage( + current_user: Annotated[User, Depends(get_current_active_user)] +) -> User: + """L1 endpoints: l1_tech, owners, super_admin, or engineers with can_cover_l1=True.""" + if current_user.is_super_admin: + return current_user + role = current_user.account_role + if role == "l1_tech": + return current_user + if role == "owner": + return current_user + if role == "engineer" and current_user.can_cover_l1: + return current_user + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="L1 access requires l1_tech role or engineer coverage flag", + ) + + +async def require_l1_or_above( + current_user: Annotated[User, Depends(get_current_active_user)] +) -> User: + """Any tier from l1_tech upward (l1_tech, engineer, owner, super_admin).""" + if current_user.is_super_admin: + return current_user + if current_user.account_role in ("l1_tech", "engineer", "owner"): + return current_user + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="L1 or above required", + ) + + async def require_team_admin( current_user: Annotated[User, Depends(get_current_active_user)] ) -> User: diff --git a/backend/tests/test_deps_l1.py b/backend/tests/test_deps_l1.py new file mode 100644 index 00000000..b302faae --- /dev/null +++ b/backend/tests/test_deps_l1.py @@ -0,0 +1,99 @@ +"""Unit tests for L1-related dependency guards. + +Uses MagicMock user objects — no database required. +""" + +from unittest.mock import MagicMock +from uuid import uuid4 + +import pytest +from fastapi import HTTPException + +from app.api.deps import require_l1, require_l1_or_coverage, require_l1_or_above + + +def _make_user(account_role="engineer", is_super_admin=False, can_cover_l1=False): + user = MagicMock() + user.id = uuid4() + user.account_role = account_role + user.is_super_admin = is_super_admin + user.can_cover_l1 = can_cover_l1 + return user + + +# --------------------------------------------------------------------------- +# require_l1 +# --------------------------------------------------------------------------- + + +async def test_require_l1_passes_for_l1_tech(): + user = _make_user(account_role="l1_tech") + result = await require_l1(current_user=user) + assert result is user + + +async def test_require_l1_passes_for_super_admin(): + user = _make_user(account_role="owner", is_super_admin=True) + result = await require_l1(current_user=user) + assert result is user + + +async def test_require_l1_blocks_engineer(): + user = _make_user(account_role="engineer") + with pytest.raises(HTTPException) as exc: + await require_l1(current_user=user) + assert exc.value.status_code == 403 + + +# --------------------------------------------------------------------------- +# require_l1_or_coverage +# --------------------------------------------------------------------------- + + +async def test_require_l1_or_coverage_passes_l1_tech(): + user = _make_user(account_role="l1_tech") + result = await require_l1_or_coverage(current_user=user) + assert result is user + + +async def test_require_l1_or_coverage_passes_engineer_with_flag(): + user = _make_user(account_role="engineer", can_cover_l1=True) + result = await require_l1_or_coverage(current_user=user) + assert result is user + + +async def test_require_l1_or_coverage_blocks_engineer_without_flag(): + user = _make_user(account_role="engineer", can_cover_l1=False) + with pytest.raises(HTTPException) as exc: + await require_l1_or_coverage(current_user=user) + assert exc.value.status_code == 403 + + +async def test_require_l1_or_coverage_passes_owner_always(): + user = _make_user(account_role="owner") + result = await require_l1_or_coverage(current_user=user) + assert result is user + + +# --------------------------------------------------------------------------- +# require_l1_or_above +# --------------------------------------------------------------------------- + + +async def test_require_l1_or_above_passes_engineer(): + user = _make_user(account_role="engineer") + result = await require_l1_or_above(current_user=user) + assert result is user + + +async def test_require_l1_or_above_passes_l1_tech(): + user = _make_user(account_role="l1_tech") + result = await require_l1_or_above(current_user=user) + assert result is user + + +async def test_require_l1_or_above_blocks_viewer(): + user = _make_user(account_role="viewer") + with pytest.raises(HTTPException) as exc: + await require_l1_or_above(current_user=user) + assert exc.value.status_code == 403