Lars
25d3976226
All checks were successful
Deploy Development / deploy (push) Successful in 35s
Test Suite / pytest-backend (push) Successful in 9s
Test Suite / lint-backend (push) Successful in 2s
Test Suite / compose-smoke (push) Has been skipped
Test Suite / k6 /api/health Baseline (push) Successful in 18s
Test Suite / playwright-smoke (push) Successful in 13s
Co-authored-by: Cursor <cursoragent@cursor.com>
92 lines
2.6 KiB
Python
92 lines
2.6 KiB
Python
"""Test data factories (PostgreSQL required)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
|
|
from auth import hash_password
|
|
from db import get_connection
|
|
from services.actors import create_actor
|
|
|
|
|
|
def create_tenant(*, slug: str | None = None, name: str = "Test Tenant") -> str:
|
|
slug = slug or f"t-{uuid.uuid4().hex[:10]}"
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"INSERT INTO tenants (slug, name) VALUES (%s, %s) RETURNING id",
|
|
(slug, name),
|
|
)
|
|
tenant_id = str(cur.fetchone()[0])
|
|
conn.commit()
|
|
return tenant_id
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def create_user(
|
|
*,
|
|
email: str | None = None,
|
|
password: str = "test-password-123",
|
|
display_name: str = "Test User",
|
|
portal_role: str = "user",
|
|
) -> dict:
|
|
email = email or f"user-{uuid.uuid4().hex[:8]}@example.com"
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO users (email, password_hash, display_name, portal_role)
|
|
VALUES (%s, %s, %s, %s)
|
|
RETURNING id, email, display_name, portal_role
|
|
""",
|
|
(email, hash_password(password), display_name, portal_role),
|
|
)
|
|
row = cur.fetchone()
|
|
conn.commit()
|
|
return {
|
|
"id": str(row[0]),
|
|
"email": row[1],
|
|
"display_name": row[2],
|
|
"portal_role": row[3],
|
|
"password": password,
|
|
}
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def add_membership(*, tenant_id: str, user_id: str, tenant_role: str = "member") -> None:
|
|
conn = get_connection()
|
|
try:
|
|
with conn.cursor() as cur:
|
|
cur.execute(
|
|
"""
|
|
INSERT INTO tenant_memberships (tenant_id, user_id, tenant_role)
|
|
VALUES (%s, %s, %s)
|
|
ON CONFLICT (tenant_id, user_id) DO UPDATE SET tenant_role = EXCLUDED.tenant_role
|
|
""",
|
|
(tenant_id, user_id, tenant_role),
|
|
)
|
|
conn.commit()
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def provision_user_in_tenant(
|
|
*,
|
|
tenant_role: str = "member",
|
|
portal_role: str = "user",
|
|
) -> dict:
|
|
tenant_id = create_tenant()
|
|
user = create_user(portal_role=portal_role)
|
|
add_membership(tenant_id=tenant_id, user_id=user["id"], tenant_role=tenant_role)
|
|
actor = create_actor(
|
|
tenant_id=tenant_id,
|
|
actor_type="human",
|
|
name=user["display_name"],
|
|
user_id=user["id"],
|
|
)
|
|
return {**user, "tenant_id": tenant_id, "actor_id": actor["id"]}
|