feat(l1): add require_l1, require_l1_or_coverage, require_l1_or_above deps
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
@@ -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:
|
||||
|
||||
99
backend/tests/test_deps_l1.py
Normal file
99
backend/tests/test_deps_l1.py
Normal file
@@ -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
|
||||
Reference in New Issue
Block a user