porchlight/tests/e2e/setup_db.py

#!/usr/bin/env python3
"""Seed the e2e test database with test fixtures.

Outputs JSON with the created test data (magic link tokens, usernames, etc.)
so the JS tests can use them.

Requires OIDC_OP_SQLITE_PATH env var pointing to the app's SQLite DB.
"""

import asyncio
import json
import os
import sys

import aiosqlite

from porchlight.authn.password import PasswordService
from porchlight.invite.service import MagicLinkService
from porchlight.models import PasswordCredential, User
from porchlight.store.sqlite.repositories import (
    SQLiteCredentialRepository,
    SQLiteMagicLinkRepository,
    SQLiteUserRepository,
)


async def _create_user_with_password(
    user_repo: SQLiteUserRepository,
    cred_repo: SQLiteCredentialRepository,
    password_service: PasswordService,
    user: User,
    password: str,
) -> None:
    """Helper to create a user and set their password credential."""
    await user_repo.create(user)
    password_hash = password_service.hash(password)
    await cred_repo.create_password(PasswordCredential(user_id=user.userid, password_hash=password_hash))


async def _seed_test_users(
    user_repo: SQLiteUserRepository,
    cred_repo: SQLiteCredentialRepository,
    password_service: PasswordService,
    result: dict[str, str],
) -> None:
    """Create all test users with passwords."""
    # Login test user
    await _create_user_with_password(
        user_repo,
        cred_repo,
        password_service,
        User(userid="test-user-01", username="testuser", groups=["users"]),
        "testpassword123",
    )
    result["login_username"] = "testuser"
    result["login_password"] = "testpassword123"

    # Credentials management test user
    await _create_user_with_password(
        user_repo,
        cred_repo,
        password_service,
        User(userid="test-user-02", username="creduser", groups=["users"]),
        "credpassword123",
    )
    result["cred_username"] = "creduser"
    result["cred_password"] = "credpassword123"

    # WebAuthn registration test user
    await _create_user_with_password(
        user_repo,
        cred_repo,
        password_service,
        User(userid="test-user-03", username="webauthnuser", groups=["users"]),
        "webauthnpass123",
    )
    result["webauthn_username"] = "webauthnuser"
    result["webauthn_password"] = "webauthnpass123"
    result["webauthn_userid"] = "test-user-03"

    # Profile management test user
    await _create_user_with_password(
        user_repo,
        cred_repo,
        password_service,
        User(
            userid="test-user-04",
            username="profileuser",
            given_name="Alice",
            family_name="Smith",
            preferred_username="asmith",
            email="alice@example.com",
            phone_number="+12025551234",
            picture="https://example.com/alice.jpg",
            locale="en",
            groups=["users"],
        ),
        "profilepass123",
    )
    result["profile_username"] = "profileuser"
    result["profile_password"] = "profilepass123"

    # Admin user for admin page tests
    await _create_user_with_password(
        user_repo,
        cred_repo,
        password_service,
        User(
            userid="test-user-05",
            username="adminuser",
            given_name="Admin",
            family_name="User",
            email="admin@example.com",
            groups=["admin", "users"],
        ),
        "adminpass123",
    )
    result["admin_username"] = "adminuser"
    result["admin_password"] = "adminpass123"
    result["admin_userid"] = "test-user-05"

    # Disposable user for admin delete test
    await user_repo.create(User(userid="test-user-06", username="disposableuser", groups=["users"]))
    result["disposable_userid"] = "test-user-06"
    result["disposable_username"] = "disposableuser"


async def seed() -> None:
    db_path = os.environ.get("OIDC_OP_SQLITE_PATH")
    if not db_path:
        print("OIDC_OP_SQLITE_PATH not set", file=sys.stderr)
        sys.exit(1)

    db = await aiosqlite.connect(db_path)
    db.row_factory = aiosqlite.Row

    user_repo = SQLiteUserRepository(db)
    cred_repo = SQLiteCredentialRepository(db)
    magic_link_repo = SQLiteMagicLinkRepository(db)
    password_service = PasswordService()
    magic_link_service = MagicLinkService(repo=magic_link_repo)

    result: dict[str, str] = {}

    # Create magic link for registration test
    link = await magic_link_service.create(username="newuser")
    result["register_token"] = link.token
    result["register_username"] = "newuser"

    # Create all test users
    await _seed_test_users(user_repo, cred_repo, password_service, result)

    # Create an expired/used magic link for negative test
    expired_link = await magic_link_service.create(username="expired")
    await magic_link_service.mark_used(expired_link.token)
    result["used_token"] = expired_link.token

    await db.commit()
    await db.close()
    print(json.dumps(result))


asyncio.run(seed())