From 02fc47c832127ca15abd6c78ccd9368deae8d998 Mon Sep 17 00:00:00 2001 From: Michael Chihlas Date: Thu, 28 May 2026 12:40:48 -0400 Subject: [PATCH] feat(l1): seat_enforcement service for engineer + L1 seat limits MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Shared helper used by invite, accept-invite, and role-change endpoints (integrated in T8). Counts active users by role against role-specific seat limit on subscription (engineer → seat_limit, l1_tech → l1_seat_limit). None limit = unlimited. Co-Authored-By: Claude Opus 4.7 --- backend/app/schemas/seat_enforcement.py | 18 +++ backend/app/services/seat_enforcement.py | 63 ++++++++ backend/tests/test_seat_enforcement.py | 195 +++++++++++++++++++++++ 3 files changed, 276 insertions(+) create mode 100644 backend/app/schemas/seat_enforcement.py create mode 100644 backend/app/services/seat_enforcement.py create mode 100644 backend/tests/test_seat_enforcement.py diff --git a/backend/app/schemas/seat_enforcement.py b/backend/app/schemas/seat_enforcement.py new file mode 100644 index 00000000..aa9104c0 --- /dev/null +++ b/backend/app/schemas/seat_enforcement.py @@ -0,0 +1,18 @@ +from typing import Literal, Optional + +from pydantic import BaseModel + + +Role = Literal['engineer', 'l1_tech'] + + +class SeatCheckResult(BaseModel): + available: bool + current: int + limit: Optional[int] # None = unlimited + role: Role + + +class SeatUsage(BaseModel): + engineer: SeatCheckResult + l1_tech: SeatCheckResult diff --git a/backend/app/services/seat_enforcement.py b/backend/app/services/seat_enforcement.py new file mode 100644 index 00000000..3756fabb --- /dev/null +++ b/backend/app/services/seat_enforcement.py @@ -0,0 +1,63 @@ +from typing import Literal + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.account import Account +from app.models.subscription import Subscription +from app.models.user import User +from app.schemas.seat_enforcement import SeatCheckResult + + +Role = Literal['engineer', 'l1_tech'] + + +def _limit_for_role(subscription: Subscription, role: Role) -> int | None: + if role == 'engineer': + return subscription.seat_limit + if role == 'l1_tech': + return subscription.l1_seat_limit + raise ValueError(f"Unknown role: {role}") + + +async def check_seat_available( + account: Account, + subscription: Subscription, + role: Role, + db: AsyncSession, +) -> SeatCheckResult: + """ + Count active users with the given role in the account, compare against + the role-specific seat limit on the subscription. Returns availability. + + None limit = unlimited (returns available=True). + """ + limit = _limit_for_role(subscription, role) + + stmt = ( + select(func.count(User.id)) + .where(User.account_id == account.id) + .where(User.account_role == role) + .where(User.is_active.is_(True)) + ) + current = (await db.execute(stmt)).scalar_one() + + if limit is None: + return SeatCheckResult(available=True, current=current, limit=None, role=role) + return SeatCheckResult( + available=current < limit, + current=current, + limit=limit, + role=role, + ) + + +async def get_seat_usage( + account: Account, + subscription: Subscription, + db: AsyncSession, +) -> tuple[SeatCheckResult, SeatCheckResult]: + """Return (engineer, l1_tech) seat-usage tuple for the seat-counter widget.""" + eng = await check_seat_available(account, subscription, 'engineer', db) + l1 = await check_seat_available(account, subscription, 'l1_tech', db) + return eng, l1 diff --git a/backend/tests/test_seat_enforcement.py b/backend/tests/test_seat_enforcement.py new file mode 100644 index 00000000..1302b77f --- /dev/null +++ b/backend/tests/test_seat_enforcement.py @@ -0,0 +1,195 @@ +"""Integration tests for the seat_enforcement service. + +Uses the test_db fixture (real async DB, fresh schema per test) to exercise +the SQL counting logic in check_seat_available / get_seat_usage. +""" + +import uuid + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession + +from app.models.account import Account +from app.models.subscription import Subscription +from app.models.user import User +from app.services.seat_enforcement import check_seat_available, get_seat_usage + + +# --------------------------------------------------------------------------- +# Test-local DB helpers +# --------------------------------------------------------------------------- + +async def _make_account(db: AsyncSession, *, suffix: str | None = None) -> Account: + """Create and flush a minimal Account row.""" + s = suffix or str(uuid.uuid4())[:8] + account = Account( + id=uuid.uuid4(), + name=f"Test Account {s}", + display_code=s[:8], + ) + db.add(account) + await db.flush() + return account + + +async def _make_subscription( + db: AsyncSession, + account: Account, + *, + seat_limit: int | None = None, + l1_seat_limit: int | None = None, +) -> Subscription: + """Create and flush a Subscription for the given account.""" + sub = Subscription( + account_id=account.id, + plan="pro", + status="active", + seat_limit=seat_limit, + l1_seat_limit=l1_seat_limit, + ) + db.add(sub) + await db.flush() + return sub + + +async def _make_user( + db: AsyncSession, + account: Account, + *, + account_role: str = "engineer", + is_active: bool = True, + suffix: str | None = None, +) -> User: + """Create and flush a User row in the given account.""" + s = suffix or str(uuid.uuid4())[:8] + user = User( + id=uuid.uuid4(), + email=f"user-{s}@example.com", + name=f"User {s}", + account_id=account.id, + account_role=account_role, + role="engineer", + is_active=is_active, + ) + db.add(user) + await db.flush() + return user + + +# --------------------------------------------------------------------------- +# Tests +# --------------------------------------------------------------------------- + +@pytest.mark.asyncio +async def test_engineer_seat_available_when_under_limit(test_db: AsyncSession): + """check_seat_available returns available=True when current < seat_limit.""" + account = await _make_account(test_db) + sub = await _make_subscription(test_db, account, seat_limit=5) + + for _ in range(3): + await _make_user(test_db, account, account_role="engineer") + + result = await check_seat_available(account, sub, "engineer", test_db) + + assert result.available is True + assert result.current == 3 + assert result.limit == 5 + assert result.role == "engineer" + + +@pytest.mark.asyncio +async def test_engineer_seat_unavailable_when_at_limit(test_db: AsyncSession): + """check_seat_available returns available=False when current == seat_limit.""" + account = await _make_account(test_db) + sub = await _make_subscription(test_db, account, seat_limit=2) + + for _ in range(2): + await _make_user(test_db, account, account_role="engineer") + + result = await check_seat_available(account, sub, "engineer", test_db) + + assert result.available is False + assert result.current == 2 + assert result.limit == 2 + + +@pytest.mark.asyncio +async def test_l1_uses_separate_seat_limit(test_db: AsyncSession): + """Engineer limit hit does not affect l1_tech availability.""" + account = await _make_account(test_db) + # seat_limit exhausted, l1_seat_limit still has room + sub = await _make_subscription(test_db, account, seat_limit=2, l1_seat_limit=3) + + # Fill engineer seats to the limit + for _ in range(2): + await _make_user(test_db, account, account_role="engineer") + + # Add one L1 user (below limit) + await _make_user(test_db, account, account_role="l1_tech") + + eng_result = await check_seat_available(account, sub, "engineer", test_db) + l1_result = await check_seat_available(account, sub, "l1_tech", test_db) + + assert eng_result.available is False, "engineer seats should be full" + assert eng_result.current == 2 + + assert l1_result.available is True, "l1_tech seats should still be available" + assert l1_result.current == 1 + assert l1_result.limit == 3 + + +@pytest.mark.asyncio +async def test_unlimited_seat_limit_is_always_available(test_db: AsyncSession): + """seat_limit=None means unlimited; available=True regardless of count.""" + account = await _make_account(test_db) + sub = await _make_subscription(test_db, account, seat_limit=None) + + # Add many engineer users + for _ in range(10): + await _make_user(test_db, account, account_role="engineer") + + result = await check_seat_available(account, sub, "engineer", test_db) + + assert result.available is True + assert result.current == 10 + assert result.limit is None + + +@pytest.mark.asyncio +async def test_get_seat_usage_returns_engineer_l1_tuple(test_db: AsyncSession): + """get_seat_usage returns a (engineer, l1_tech) tuple in the correct order.""" + account = await _make_account(test_db) + sub = await _make_subscription(test_db, account, seat_limit=5, l1_seat_limit=3) + + await _make_user(test_db, account, account_role="engineer") + await _make_user(test_db, account, account_role="l1_tech") + await _make_user(test_db, account, account_role="l1_tech") + + eng, l1 = await get_seat_usage(account, sub, test_db) + + assert eng.role == "engineer" + assert eng.current == 1 + assert eng.limit == 5 + assert eng.available is True + + assert l1.role == "l1_tech" + assert l1.current == 2 + assert l1.limit == 3 + assert l1.available is True + + +@pytest.mark.asyncio +async def test_inactive_users_not_counted(test_db: AsyncSession): + """Inactive (is_active=False) users are excluded from the seat count.""" + account = await _make_account(test_db) + sub = await _make_subscription(test_db, account, seat_limit=3) + + # 1 active, 2 inactive + await _make_user(test_db, account, account_role="engineer", is_active=True) + await _make_user(test_db, account, account_role="engineer", is_active=False) + await _make_user(test_db, account, account_role="engineer", is_active=False) + + result = await check_seat_available(account, sub, "engineer", test_db) + + assert result.current == 1 + assert result.available is True