feat(l1): PATCH /accounts/me/members/{id}/coverage for engineer L1-coverage flag

Owner-only endpoint to toggle can_cover_l1 on an engineer user. 422 if target
role is not engineer (owners/super_admins already see L1 surface; viewers/
l1_techs don't need this flag). 404 for cross-account targets.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-28 13:07:09 -04:00
parent 7056ed9e6d
commit e15897c76f
3 changed files with 150 additions and 1 deletions

View File

@@ -21,7 +21,7 @@ from app.models.subscription import Subscription
from app.models.user import User
from app.schemas.account import AccountResponse, AccountUpdate, AccountInviteCreate, AccountInviteResponse, AccountInviteBulkCreate, AccountInviteBulkResponse, TransferOwnershipRequest
from app.schemas.subscription import SubscriptionResponse, PlanLimitsResponse, UsageResponse, SubscriptionDetails
from app.schemas.user import UserResponse, AccountRoleUpdate
from app.schemas.user import UserResponse, AccountRoleUpdate, CoverageUpdate
from app.core.security import verify_password
from app.api.deps import get_current_active_user, require_account_owner, require_engineer_or_admin
from app.services.seat_enforcement import check_seat_available, get_seat_usage
@@ -228,6 +228,43 @@ async def update_member_role(
return user
@router.patch("/me/members/{user_id}/coverage", response_model=UserResponse)
async def update_member_coverage(
user_id: UUID,
data: CoverageUpdate,
db: Annotated[AsyncSession, Depends(get_db)],
current_user: Annotated[User, Depends(require_account_owner)],
):
"""Toggle the `can_cover_l1` flag on an engineer in your account.
Owner-only. Returns 404 if target user not in your account. Returns 422
if target user's role is not 'engineer' (coverage flag only applies to
engineers — owners/super_admins already see L1 surface; viewers/l1_techs
don't need this flag).
"""
result = await db.execute(
select(User).where(
User.id == user_id,
User.account_id == current_user.account_id,
)
)
target = result.scalar_one_or_none()
if target is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="User not found in your account",
)
if target.account_role != "engineer":
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="can_cover_l1 only applies to engineers",
)
target.can_cover_l1 = data.can_cover_l1
await db.commit()
await db.refresh(target)
return target
@router.post("/me/transfer-ownership", response_model=AccountResponse)
async def transfer_ownership(
data: TransferOwnershipRequest,

View File

@@ -60,6 +60,7 @@ class UserResponse(UserBase):
email_verified_at: Optional[datetime] = None
onboarding_step_completed: Optional[int] = None
onboarding_dismissed: bool = False
can_cover_l1: bool = False
class Config:
from_attributes = True
@@ -73,3 +74,7 @@ class AccountRoleUpdate(BaseModel):
# Ownership changes must go through the explicit transfer-ownership flow so
# account.owner_id stays consistent with user.account_role.
account_role: str = Field(..., pattern="^(admin|engineer|viewer|l1_tech)$")
class CoverageUpdate(BaseModel):
can_cover_l1: bool

View File

@@ -455,3 +455,110 @@ async def test_get_seats_blocked_for_viewer(client: AsyncClient, test_db: AsyncS
resp = await client.get("/api/v1/accounts/me/seats", headers=viewer_headers)
assert resp.status_code == 403, resp.text
# ---------------------------------------------------------------------------
# PATCH /me/members/{user_id}/coverage — engineer L1-coverage flag
# ---------------------------------------------------------------------------
@pytest.mark.asyncio
async def test_coverage_owner_can_toggle_engineer(client: AsyncClient, test_db: AsyncSession):
"""Owner can set can_cover_l1=True on an engineer; response reflects new value."""
owner = await _register(client, email="owner_cov1@example.com")
account_id = uuid.UUID(owner["account_id"])
headers = await _login(client, email="owner_cov1@example.com")
engineer = await _add_member(test_db, account_id, role="engineer", suffix="cov1")
resp = await client.patch(
f"/api/v1/accounts/me/members/{engineer.id}/coverage",
json={"can_cover_l1": True},
headers=headers,
)
assert resp.status_code == 200, resp.text
body = resp.json()
assert body["can_cover_l1"] is True
# Toggle back to False
resp2 = await client.patch(
f"/api/v1/accounts/me/members/{engineer.id}/coverage",
json={"can_cover_l1": False},
headers=headers,
)
assert resp2.status_code == 200, resp2.text
assert resp2.json()["can_cover_l1"] is False
@pytest.mark.asyncio
async def test_coverage_non_owner_is_forbidden(client: AsyncClient, test_db: AsyncSession):
"""A non-owner engineer cannot toggle coverage on themselves or others."""
from app.core.security import get_password_hash
owner = await _register(client, email="owner_cov2@example.com")
account_id = uuid.UUID(owner["account_id"])
# Create an engineer with a known password
eng_password = "EngPass123!"
engineer = User(
id=uuid.uuid4(),
email="eng_cov2@example.com",
name="Eng Cov2",
account_id=account_id,
account_role="engineer",
role="engineer",
is_active=True,
password_hash=get_password_hash(eng_password),
)
test_db.add(engineer)
await test_db.commit()
eng_headers = await _login(client, email="eng_cov2@example.com", password=eng_password)
resp = await client.patch(
f"/api/v1/accounts/me/members/{engineer.id}/coverage",
json={"can_cover_l1": True},
headers=eng_headers,
)
assert resp.status_code == 403, resp.text
@pytest.mark.asyncio
async def test_coverage_viewer_role_returns_422(client: AsyncClient, test_db: AsyncSession):
"""PATCH coverage on a viewer → 422 (coverage flag only applies to engineers)."""
owner = await _register(client, email="owner_cov3@example.com")
account_id = uuid.UUID(owner["account_id"])
headers = await _login(client, email="owner_cov3@example.com")
viewer = await _add_member(test_db, account_id, role="viewer", suffix="cov3")
resp = await client.patch(
f"/api/v1/accounts/me/members/{viewer.id}/coverage",
json={"can_cover_l1": True},
headers=headers,
)
assert resp.status_code == 422, resp.text
assert "engineer" in resp.json()["detail"].lower()
@pytest.mark.asyncio
async def test_coverage_cross_account_returns_404(client: AsyncClient, test_db: AsyncSession):
"""PATCH coverage on a user from a different account → 404 (tenancy isolation)."""
# Account A
owner_a = await _register(client, email="owner_cov_a@example.com")
account_a_id = uuid.UUID(owner_a["account_id"])
headers_a = await _login(client, email="owner_cov_a@example.com")
# Account B — a separate registration creates a new account
owner_b = await _register(client, email="owner_cov_b@example.com")
account_b_id = uuid.UUID(owner_b["account_id"])
# Add an engineer to account B
engineer_b = await _add_member(test_db, account_b_id, role="engineer", suffix="covb")
# Owner of account A tries to patch account B's engineer — must 404
resp = await client.patch(
f"/api/v1/accounts/me/members/{engineer_b.id}/coverage",
json={"can_cover_l1": True},
headers=headers_a,
)
assert resp.status_code == 404, resp.text