diff --git a/backend/app/api/endpoints/accounts.py b/backend/app/api/endpoints/accounts.py index ea2e8624..c211a4e0 100644 --- a/backend/app/api/endpoints/accounts.py +++ b/backend/app/api/endpoints/accounts.py @@ -23,9 +23,17 @@ from app.schemas.account import AccountResponse, AccountUpdate, AccountInviteCre from app.schemas.subscription import SubscriptionResponse, PlanLimitsResponse, UsageResponse, SubscriptionDetails 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.api.deps import ( + get_current_active_user, + require_account_owner, + require_account_owner_or_admin, + require_engineer_or_admin, + require_l1_or_above, +) +from app.services import l1_category_service from app.services.seat_enforcement import check_seat_available, get_seat_usage from app.schemas.seat_enforcement import SeatUsage +from app.schemas.l1_categories import L1CategoriesResponse, L1CategoriesUpdate _SEAT_CHECKED_ROLES = frozenset({"engineer", "l1_tech"}) @@ -164,6 +172,45 @@ async def get_my_account_seat_usage( return SeatUsage(engineer=engineer, l1_tech=l1_tech) +@router.get("/me/l1-categories", response_model=L1CategoriesResponse) +async def get_l1_categories( + db: Annotated[AsyncSession, Depends(get_db)], + current_user: Annotated[User, Depends(require_l1_or_above)], +): + """The account's enabled L1 AI-build categories + the available + hard-floor lists. + + Readable by any L1-or-above user (the walker needs to know what's buildable); + only owners/admins may change it (PATCH below). + """ + enabled = await l1_category_service.get_enabled_categories(current_user.account_id, db) + return L1CategoriesResponse( + enabled=enabled, + available=l1_category_service.DEFAULT_L1_CATEGORIES, + hard_floor=l1_category_service.HARD_FLOOR_FORBIDDEN, + ) + + +@router.patch("/me/l1-categories", response_model=L1CategoriesResponse) +async def set_l1_categories( + payload: L1CategoriesUpdate, + db: Annotated[AsyncSession, Depends(get_db)], + current_user: Annotated[User, Depends(require_account_owner_or_admin)], +): + """Set the account's enabled L1 categories (owner/admin only). + + Unknown and hard-floored keys are dropped by the service before persisting. + """ + enabled = await l1_category_service.set_enabled_categories( + current_user.account_id, payload.enabled, db + ) + await db.commit() + return L1CategoriesResponse( + enabled=enabled, + available=l1_category_service.DEFAULT_L1_CATEGORIES, + hard_floor=l1_category_service.HARD_FLOOR_FORBIDDEN, + ) + + @router.patch("/me", response_model=AccountResponse) async def update_my_account( data: AccountUpdate, diff --git a/backend/app/schemas/l1_categories.py b/backend/app/schemas/l1_categories.py new file mode 100644 index 00000000..5b5180c0 --- /dev/null +++ b/backend/app/schemas/l1_categories.py @@ -0,0 +1,14 @@ +"""Schemas for the account L1 AI-build category settings surface (Phase 2A).""" +from pydantic import BaseModel + + +class L1CategoriesResponse(BaseModel): + """Current enabled set + the full available list + the read-only hard floor.""" + enabled: list[str] + available: list[str] + hard_floor: list[str] + + +class L1CategoriesUpdate(BaseModel): + """Owner/admin write: the new enabled set (unknown/hard-floored keys dropped).""" + enabled: list[str] diff --git a/backend/tests/test_l1_categories_api.py b/backend/tests/test_l1_categories_api.py new file mode 100644 index 00000000..fe838f0a --- /dev/null +++ b/backend/tests/test_l1_categories_api.py @@ -0,0 +1,119 @@ +"""Tests for the account L1 AI-build category settings API (Phase 2A). + +GET /accounts/me/l1-categories — readable by L1-or-above. +PATCH /accounts/me/l1-categories — owner/admin only; drops unknown/hard-floored keys. +""" +import uuid + +import pytest +from httpx import AsyncClient +from sqlalchemy import delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.subscription import Subscription +from app.models.user import User + + +async def _register(client: AsyncClient, *, email: str) -> dict: + resp = await client.post( + "/api/v1/auth/register", + json={"email": email, "password": "TestPassword123!", "name": "Test User"}, + ) + assert resp.status_code in (200, 201), resp.text + return resp.json() + + +async def _login(client: AsyncClient, *, email: str) -> dict: + resp = await client.post( + "/api/v1/auth/login/json", + json={"email": email, "password": "TestPassword123!"}, + ) + assert resp.status_code == 200, resp.text + return {"Authorization": f"Bearer {resp.json()['access_token']}"} + + +async def _ensure_subscription(db: AsyncSession, account_id: uuid.UUID) -> None: + await db.execute(delete(Subscription).where(Subscription.account_id == account_id)) + db.add(Subscription(account_id=account_id, plan="pro", status="active")) + await db.commit() + + +async def _make_user(client: AsyncClient, db: AsyncSession, *, email: str, account_role: str) -> dict: + """Register → promote role → ensure subscription → login (after the role change).""" + data = await _register(client, email=email) + uid = uuid.UUID(data["id"]) + acct_id = uuid.UUID(data["account_id"]) + user = (await db.execute(select(User).where(User.id == uid))).scalar_one() + user.account_role = account_role + await db.commit() + await _ensure_subscription(db, acct_id) + headers = await _login(client, email=email) + return {"headers": headers, "account_id": acct_id, "user_id": uid} + + +@pytest.mark.asyncio +async def test_get_categories_returns_enabled_available_hard_floor(client: AsyncClient, test_db: AsyncSession): + info = await _make_user(client, test_db, email="cat_owner_get@example.com", account_role="owner") + r = await client.get("/api/v1/accounts/me/l1-categories", headers=info["headers"]) + assert r.status_code == 200, r.text + body = r.json() + assert "enabled" in body and "available" in body and "hard_floor" in body + # New account defaults to the full available allowlist (10 keys). + assert len(body["available"]) == 10 + assert "password_reset" in body["available"] + assert "registry_edit" in body["hard_floor"] + + +@pytest.mark.asyncio +async def test_get_categories_readable_by_l1_tech(client: AsyncClient, test_db: AsyncSession): + info = await _make_user(client, test_db, email="cat_l1_get@example.com", account_role="l1_tech") + r = await client.get("/api/v1/accounts/me/l1-categories", headers=info["headers"]) + assert r.status_code == 200, r.text + + +@pytest.mark.asyncio +async def test_patch_categories_owner_can_set(client: AsyncClient, test_db: AsyncSession): + info = await _make_user(client, test_db, email="cat_owner_patch@example.com", account_role="owner") + r = await client.patch( + "/api/v1/accounts/me/l1-categories", + json={"enabled": ["printer", "vpn_connect"]}, + headers=info["headers"], + ) + assert r.status_code == 200, r.text + assert set(r.json()["enabled"]) == {"printer", "vpn_connect"} + + +@pytest.mark.asyncio +async def test_patch_categories_drops_unknown_and_hard_floored(client: AsyncClient, test_db: AsyncSession): + info = await _make_user(client, test_db, email="cat_owner_drop@example.com", account_role="owner") + r = await client.patch( + "/api/v1/accounts/me/l1-categories", + json={"enabled": ["printer", "registry_edit", "bogus_key"]}, + headers=info["headers"], + ) + assert r.status_code == 200, r.text + # registry_edit (hard floor) and bogus_key (unknown) are dropped. + assert r.json()["enabled"] == ["printer"] + + +@pytest.mark.asyncio +async def test_patch_categories_forbidden_for_l1_tech(client: AsyncClient, test_db: AsyncSession): + info = await _make_user(client, test_db, email="cat_l1_patch@example.com", account_role="l1_tech") + r = await client.patch( + "/api/v1/accounts/me/l1-categories", + json={"enabled": ["printer"]}, + headers=info["headers"], + ) + assert r.status_code == 403, r.text + + +@pytest.mark.asyncio +async def test_patch_categories_forbidden_for_engineer(client: AsyncClient, test_db: AsyncSession): + """Write is owner/admin only — engineers (who pass require_engineer_or_admin) are blocked.""" + info = await _make_user(client, test_db, email="cat_eng_patch@example.com", account_role="engineer") + r = await client.patch( + "/api/v1/accounts/me/l1-categories", + json={"enabled": ["printer"]}, + headers=info["headers"], + ) + assert r.status_code == 403, r.text