diff --git a/backend/app/api/endpoints/accounts.py b/backend/app/api/endpoints/accounts.py index 0e0f457f..ea2e8624 100644 --- a/backend/app/api/endpoints/accounts.py +++ b/backend/app/api/endpoints/accounts.py @@ -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, diff --git a/backend/app/schemas/user.py b/backend/app/schemas/user.py index 044e84e2..22c99928 100644 --- a/backend/app/schemas/user.py +++ b/backend/app/schemas/user.py @@ -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 diff --git a/backend/tests/test_invite_seat_enforcement.py b/backend/tests/test_invite_seat_enforcement.py index 5d002775..d0436474 100644 --- a/backend/tests/test_invite_seat_enforcement.py +++ b/backend/tests/test_invite_seat_enforcement.py @@ -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