Michael Chihlas
6937bcaabd
l1-workspace.spec.ts covers: - L1 user lands on /l1, intakes a problem, takes notes (autosave), resolves - L1 cannot access /pilot, /trees/new, /escalations (route guards) - Engineer with can_cover_l1 sees the L1 Workspace nav + coverage banner - escalate-without-walk path via direct API call returns escalated session Seed script adds l1@resolutionflow.example.com (l1_tech) and engineer-coverage@resolutionflow.example.com (engineer + can_cover_l1). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
240 lines
9.4 KiB
Python
240 lines
9.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
Create test user accounts for local development.
|
||
|
||
Creates 6 accounts:
|
||
1. Super Admin – platform-wide admin (manages everything)
|
||
2. Pro Solo User – single user on a "pro" plan
|
||
3. Team Admin – admin of a team account ("team" plan)
|
||
4. Team Engineer – regular engineer on the same team account
|
||
5. L1 Tech – l1_tech role on the Acme MSP team (E2E: L1 happy path)
|
||
6. Coverage Engineer – engineer with can_cover_l1=True (E2E: coverage banner)
|
||
|
||
Usage:
|
||
cd backend
|
||
python -m scripts.seed_test_users
|
||
"""
|
||
|
||
import asyncio
|
||
import random
|
||
import string
|
||
import uuid
|
||
from datetime import datetime, timezone
|
||
|
||
from sqlalchemy import text
|
||
from sqlalchemy.ext.asyncio import create_async_engine
|
||
|
||
from app.core.config import settings
|
||
from app.core.security import get_password_hash
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Configuration – change passwords here if you like
|
||
# ---------------------------------------------------------------------------
|
||
SHARED_PASSWORD = "TestPass123!"
|
||
|
||
USERS = [
|
||
{
|
||
"key": "super_admin",
|
||
"name": "RF Super Admin",
|
||
"email": "admin@resolutionflow.example.com",
|
||
"is_super_admin": True,
|
||
"is_team_admin": False,
|
||
"account_name": "ResolutionFlow Platform",
|
||
"account_role": "owner",
|
||
"plan": "team",
|
||
},
|
||
{
|
||
"key": "pro_solo",
|
||
"name": "Pat Solo",
|
||
"email": "pro@resolutionflow.example.com",
|
||
"is_super_admin": False,
|
||
"is_team_admin": False,
|
||
"account_name": "Pat's MSP",
|
||
"account_role": "owner",
|
||
"plan": "pro",
|
||
},
|
||
{
|
||
"key": "team_admin",
|
||
"name": "Alex Manager",
|
||
"email": "teamadmin@resolutionflow.example.com",
|
||
"is_super_admin": False,
|
||
"is_team_admin": True,
|
||
"account_name": "Acme MSP",
|
||
"account_role": "owner", # owns the account; is_team_admin=True gives admin powers
|
||
"plan": "team",
|
||
},
|
||
{
|
||
"key": "team_engineer",
|
||
"name": "Jordan Tech",
|
||
"email": "engineer@resolutionflow.example.com",
|
||
"is_super_admin": False,
|
||
"is_team_admin": False,
|
||
"account_name": "Acme MSP", # same shared account
|
||
"account_role": "engineer",
|
||
"plan": None, # uses the team_admin's account & subscription
|
||
"can_cover_l1": False,
|
||
},
|
||
{
|
||
"key": "l1_tech",
|
||
"name": "Lee L1Tech",
|
||
"email": "l1@resolutionflow.example.com",
|
||
"is_super_admin": False,
|
||
"is_team_admin": False,
|
||
"account_name": "Acme MSP", # same shared account as team_admin
|
||
"account_role": "l1_tech",
|
||
"plan": None, # uses the team_admin's account & subscription
|
||
"can_cover_l1": False,
|
||
},
|
||
{
|
||
"key": "coverage_engineer",
|
||
"name": "Casey Coverage",
|
||
"email": "engineer-coverage@resolutionflow.example.com",
|
||
"is_super_admin": False,
|
||
"is_team_admin": False,
|
||
"account_name": "Acme MSP", # same shared account as team_admin
|
||
"account_role": "engineer",
|
||
"plan": None, # uses the team_admin's account & subscription
|
||
"can_cover_l1": True,
|
||
},
|
||
]
|
||
|
||
|
||
def _display_code() -> str:
|
||
return "".join(random.choices(string.ascii_uppercase + string.digits, k=8))
|
||
|
||
|
||
async def main() -> None:
|
||
# Must use ADMIN_DATABASE_URL (BYPASSRLS) — Phase 4 enabled RLS on users.
|
||
# The app-role connection has no tenant context at seed time and would see 0 rows.
|
||
admin_url = getattr(settings, "ADMIN_DATABASE_URL", None) or settings.DATABASE_URL
|
||
engine = create_async_engine(admin_url, echo=False)
|
||
password_hash = get_password_hash(SHARED_PASSWORD)
|
||
now = datetime.now(timezone.utc)
|
||
team_account_id: uuid.UUID | None = None
|
||
|
||
async with engine.begin() as conn:
|
||
for cfg in USERS:
|
||
# Check if user already exists
|
||
result = await conn.execute(
|
||
text("SELECT id, account_id FROM users WHERE email = :email"),
|
||
{"email": cfg["email"]},
|
||
)
|
||
row = result.first()
|
||
if row:
|
||
# Backfill email_verified_at for existing rows so older test
|
||
# users created before this script set the field still bypass
|
||
# the 7-day verification grace.
|
||
await conn.execute(
|
||
text("""
|
||
UPDATE users
|
||
SET email_verified_at = COALESCE(email_verified_at, :now)
|
||
WHERE email = :email
|
||
"""),
|
||
{"email": cfg["email"], "now": now},
|
||
)
|
||
print(f" [SKIP] {cfg['email']} already exists (email_verified_at backfilled if null)")
|
||
if cfg["key"] == "team_admin":
|
||
team_account_id = row.account_id
|
||
continue
|
||
|
||
# ---- Create or reuse Account ----
|
||
# Users that share the Acme MSP account (no own account to create)
|
||
_acme_members = {"team_engineer", "l1_tech", "coverage_engineer"}
|
||
if cfg["key"] in _acme_members:
|
||
if team_account_id is None:
|
||
result = await conn.execute(
|
||
text("SELECT id FROM accounts WHERE name = :name"),
|
||
{"name": "Acme MSP"},
|
||
)
|
||
acme = result.first()
|
||
if acme:
|
||
team_account_id = acme.id
|
||
if team_account_id is None:
|
||
print(f" [ERROR] Cannot create {cfg['email']} — Acme MSP account not found.")
|
||
continue
|
||
account_id = team_account_id
|
||
else:
|
||
account_id = uuid.uuid4()
|
||
await conn.execute(
|
||
text("""
|
||
INSERT INTO accounts (id, name, display_code, created_at, updated_at)
|
||
VALUES (:id, :name, :code, :now, :now)
|
||
"""),
|
||
{"id": account_id, "name": cfg["account_name"], "code": _display_code(), "now": now},
|
||
)
|
||
if cfg["key"] == "team_admin":
|
||
team_account_id = account_id
|
||
|
||
# ---- Create User ----
|
||
user_id = uuid.uuid4()
|
||
# email_verified_at is stamped at seed time so test users bypass the
|
||
# 7-day verification grace immediately. Without this, fixtures hit
|
||
# require_verified_email_after_grace once their created_at ages past
|
||
# 7 days and get walled out of protected routes.
|
||
can_cover_l1 = cfg.get("can_cover_l1", False)
|
||
await conn.execute(
|
||
text("""
|
||
INSERT INTO users (id, email, password_hash, name, role, is_super_admin,
|
||
is_team_admin, is_active, account_id, account_role,
|
||
can_cover_l1, created_at, email_verified_at)
|
||
VALUES (:id, :email, :pw, :name, 'engineer', :is_sa, :is_ta, true,
|
||
:account_id, :account_role, :can_cover_l1, :now, :now)
|
||
"""),
|
||
{
|
||
"id": user_id,
|
||
"email": cfg["email"],
|
||
"pw": password_hash,
|
||
"name": cfg["name"],
|
||
"is_sa": cfg["is_super_admin"],
|
||
"is_ta": cfg["is_team_admin"],
|
||
"account_id": account_id,
|
||
"account_role": cfg["account_role"],
|
||
"can_cover_l1": can_cover_l1,
|
||
"now": now,
|
||
},
|
||
)
|
||
|
||
# Set account owner (skip for shared-account members — they don't own the account)
|
||
if cfg["key"] not in _acme_members:
|
||
await conn.execute(
|
||
text("UPDATE accounts SET owner_id = :uid WHERE id = :aid"),
|
||
{"uid": user_id, "aid": account_id},
|
||
)
|
||
|
||
# ---- Create Subscription (once per account) ----
|
||
if cfg["plan"] is not None:
|
||
await conn.execute(
|
||
text("""
|
||
INSERT INTO subscriptions (id, account_id, plan, status, cancel_at_period_end, created_at, updated_at)
|
||
VALUES (:id, :aid, :plan, 'active', false, :now, :now)
|
||
"""),
|
||
{"id": uuid.uuid4(), "aid": account_id, "plan": cfg["plan"], "now": now},
|
||
)
|
||
|
||
cover_flag = " [can_cover_l1]" if can_cover_l1 else ""
|
||
print(f" [OK] {cfg['email']:40s} account_role={cfg['account_role']:<12s} plan={cfg['plan'] or '(shared)'}{cover_flag}")
|
||
|
||
await engine.dispose()
|
||
|
||
print()
|
||
print("=" * 60)
|
||
print(" Test accounts ready!")
|
||
print(f" Password for all accounts: {SHARED_PASSWORD}")
|
||
print("=" * 60)
|
||
print()
|
||
print(" Accounts:")
|
||
print(f" Super Admin : admin@resolutionflow.example.com")
|
||
print(f" Pro Solo : pro@resolutionflow.example.com")
|
||
print(f" Team Admin : teamadmin@resolutionflow.example.com")
|
||
print(f" Team Engineer : engineer@resolutionflow.example.com")
|
||
print(f" L1 Tech : l1@resolutionflow.example.com")
|
||
print(f" Coverage Engineer : engineer-coverage@resolutionflow.example.com")
|
||
print()
|
||
|
||
|
||
if __name__ == "__main__":
|
||
print("\n[*] ResolutionFlow — Test User Seeder")
|
||
print("=" * 60)
|
||
asyncio.run(main())
|