98 lines
3.2 KiB
Python
98 lines
3.2 KiB
Python
"""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 - clear profile fields for test isolation
|
|
regular_user.hashed_password = get_password_hash(dev_user_password)
|
|
regular_user.roles = [regular_role]
|
|
regular_user.contact_email = None
|
|
regular_user.telegram = None
|
|
regular_user.signal = None
|
|
regular_user.nostr_npub = None
|
|
|
|
# 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 - clear profile fields for test isolation
|
|
admin_user.hashed_password = get_password_hash(dev_admin_password)
|
|
admin_user.roles = [admin_role]
|
|
admin_user.contact_email = None
|
|
admin_user.telegram = None
|
|
admin_user.signal = None
|
|
admin_user.nostr_npub = None
|
|
|
|
await db.commit()
|