feat(l1): account L1 category settings API (owner/admin write)

GET /accounts/me/l1-categories (require_l1_or_above) returns enabled + available
+ hard_floor; PATCH (require_account_owner_or_admin) sets the enabled set, dropping
unknown/hard-floored keys via l1_category_service. New L1CategoriesResponse/Update
schemas. 6 API tests green (incl. engineer + l1_tech write both 403); test_accounts
regression 36 passed.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-30 20:01:32 -04:00
parent 04d2cfb9a5
commit 1d3f9d0a8a
3 changed files with 181 additions and 1 deletions

View File

@@ -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