resolutionflow/backend/scripts/seed_test_users.py

#!/usr/bin/env python3
"""
Create test user accounts for local development.

Creates 4 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

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
    },
]


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 ----
            if cfg["key"] == "team_engineer":
                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.
            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,
                                       created_at, email_verified_at)
                    VALUES (:id, :email, :pw, :name, 'engineer', :is_sa, :is_ta, true,
                            :account_id, :account_role, :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"],
                    "now": now,
                },
            )

            # Set account owner (skip for team_engineer — they don't own the account)
            if cfg["key"] != "team_engineer":
                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},
                )

            print(f"  [OK]    {cfg['email']:40s}  account_role={cfg['account_role']:<10s}  plan={cfg['plan'] or '(shared)'}")

    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()


if __name__ == "__main__":
    print("\n[*] ResolutionFlow — Test User Seeder")
    print("=" * 60)
    asyncio.run(main())