working

backend/seed_e2e.py

@@ -0,0 +1,90 @@
+"""Fast re-seeding function for e2e tests - only seeds essential data."""
+
+import os
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from auth import get_password_hash
+from models import (
+    ROLE_ADMIN,
+    ROLE_DEFINITIONS,
+    ROLE_REGULAR,
+    Role,
+    User,
+)
+
+
+async def seed_base_data(db: AsyncSession) -> None:
+    """
+    Seed only the minimal data needed for e2e tests:
+    - Roles (admin, regular) with permissions
+    - Test users (from env vars)
+    """
+    # Get environment variables with defaults
+    dev_user_email = os.getenv("DEV_USER_EMAIL", "user@example.com")
+    dev_user_password = os.getenv("DEV_USER_PASSWORD", "user123")
+    dev_admin_email = os.getenv("DEV_ADMIN_EMAIL", "admin@example.com")
+    dev_admin_password = os.getenv("DEV_ADMIN_PASSWORD", "admin123")
+
+    # Create roles with permissions
+    for role_name, role_config in ROLE_DEFINITIONS.items():
+        result = await db.execute(select(Role).where(Role.name == role_name))
+        role = result.scalar_one_or_none()
+
+        if not role:
+            role = Role(name=role_name, description=role_config["description"])
+            db.add(role)
+            await db.flush()  # Get the role ID
+
+        # Set permissions for the role
+        await role.set_permissions(db, role_config["permissions"])
+
+    await db.flush()  # Ensure roles are committed before creating users
+
+    # Get roles for users
+    admin_role_result = await db.execute(select(Role).where(Role.name == ROLE_ADMIN))
+    admin_role = admin_role_result.scalar_one()
+
+    regular_role_result = await db.execute(
+        select(Role).where(Role.name == ROLE_REGULAR)
+    )
+    regular_role = regular_role_result.scalar_one()
+
+    # Create regular dev user
+    regular_user_result = await db.execute(
+        select(User).where(User.email == dev_user_email)
+    )
+    regular_user = regular_user_result.scalar_one_or_none()
+
+    if not regular_user:
+        regular_user = User(
+            email=dev_user_email,
+            hashed_password=get_password_hash(dev_user_password),
+            roles=[regular_role],
+        )
+        db.add(regular_user)
+    else:
+        # Update existing user
+        regular_user.hashed_password = get_password_hash(dev_user_password)
+        regular_user.roles = [regular_role]
+
+    # Create admin dev user
+    admin_user_result = await db.execute(
+        select(User).where(User.email == dev_admin_email)
+    )
+    admin_user = admin_user_result.scalar_one_or_none()
+
+    if not admin_user:
+        admin_user = User(
+            email=dev_admin_email,
+            hashed_password=get_password_hash(dev_admin_password),
+            roles=[admin_role],
+        )
+        db.add(admin_user)
+    else:
+        # Update existing user
+        admin_user.hashed_password = get_password_hash(dev_admin_password)
+        admin_user.roles = [admin_role]
+
+    await db.commit()