Adolfo Delorenzo
9515c5374a
RED phase — tests are written, will pass when connected to live DB. Tests cover: - Full RBAC matrix: platform_admin/customer_admin/operator on all endpoints - Operator can POST /test but not POST /agents (create) - Missing headers return 422 - Impersonation creates AuditEvent row - Full invite flow: create -> accept -> login with correct role - Expired invite rejection - Resend generates new token and extends expiry - Double-accept prevention
485 lines
17 KiB
Python
485 lines
17 KiB
Python
"""
|
|
End-to-end integration tests for the portal invitation flow.
|
|
|
|
Tests:
|
|
1. Full flow: admin creates invite -> token generated -> accept with password ->
|
|
new user created with correct role and tenant membership
|
|
2. Expired invite acceptance returns error
|
|
3. Resend generates new token and extends expiry
|
|
4. Double-accept prevented (status no longer 'pending')
|
|
5. Login works after acceptance: /auth/verify returns correct role
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import bcrypt
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy import select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from shared.api.invitations import invitations_router
|
|
from shared.api.portal import portal_router
|
|
from shared.db import get_session
|
|
from shared.invite_token import generate_invite_token, token_to_hash
|
|
from shared.models.auth import PortalInvitation, PortalUser, UserTenantRole
|
|
from shared.models.tenant import Tenant
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_app(session: AsyncSession) -> FastAPI:
|
|
"""Build a FastAPI test app with portal and invitations routers."""
|
|
app = FastAPI()
|
|
app.include_router(portal_router)
|
|
app.include_router(invitations_router)
|
|
|
|
async def override_get_session(): # type: ignore[return]
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = override_get_session
|
|
return app
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Header helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def admin_headers(user_id: uuid.UUID, tenant_id: uuid.UUID) -> dict[str, str]:
|
|
return {
|
|
"X-Portal-User-Id": str(user_id),
|
|
"X-Portal-User-Role": "customer_admin",
|
|
"X-Portal-Tenant-Id": str(tenant_id),
|
|
}
|
|
|
|
|
|
def platform_admin_headers(user_id: uuid.UUID) -> dict[str, str]:
|
|
return {
|
|
"X-Portal-User-Id": str(user_id),
|
|
"X-Portal-User-Role": "platform_admin",
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DB setup helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _create_tenant_direct(session: AsyncSession) -> Tenant:
|
|
suffix = uuid.uuid4().hex[:8]
|
|
tenant = Tenant(
|
|
id=uuid.uuid4(),
|
|
name=f"Invite Test Tenant {suffix}",
|
|
slug=f"invite-test-{suffix}",
|
|
settings={},
|
|
)
|
|
session.add(tenant)
|
|
await session.flush()
|
|
return tenant
|
|
|
|
|
|
async def _create_admin_user(session: AsyncSession, tenant: Tenant) -> PortalUser:
|
|
hashed = bcrypt.hashpw(b"adminpassword123", bcrypt.gensalt()).decode()
|
|
suffix = uuid.uuid4().hex[:8]
|
|
user = PortalUser(
|
|
id=uuid.uuid4(),
|
|
email=f"admin-{suffix}@example.com",
|
|
hashed_password=hashed,
|
|
name=f"Admin User {suffix}",
|
|
role="customer_admin",
|
|
)
|
|
session.add(user)
|
|
await session.flush()
|
|
|
|
membership = UserTenantRole(
|
|
id=uuid.uuid4(),
|
|
user_id=user.id,
|
|
tenant_id=tenant.id,
|
|
role="customer_admin",
|
|
)
|
|
session.add(membership)
|
|
await session.flush()
|
|
return user
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def invite_client(db_session: AsyncSession) -> AsyncClient:
|
|
"""HTTP client with portal and invitations routers."""
|
|
app = make_app(db_session)
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def invite_setup(db_session: AsyncSession) -> dict:
|
|
"""Create a tenant and admin user for invitation tests."""
|
|
tenant = await _create_tenant_direct(db_session)
|
|
admin = await _create_admin_user(db_session, tenant)
|
|
await db_session.commit()
|
|
return {"tenant": tenant, "admin": admin}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Full invite flow
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestInviteFlow:
|
|
async def test_full_invite_flow_operator(
|
|
self, invite_client: AsyncClient, db_session: AsyncSession, invite_setup: dict
|
|
) -> None:
|
|
"""
|
|
Full end-to-end invite flow for a customer_operator:
|
|
|
|
1. Admin creates invitation
|
|
2. Token is included in response
|
|
3. Invitee accepts invitation with password
|
|
4. PortalUser created with role=customer_operator
|
|
5. UserTenantRole created linking user to tenant
|
|
6. Invitation status updated to 'accepted'
|
|
7. Login returns correct role
|
|
"""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"operator-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Step 1: Admin creates invitation
|
|
resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "New Operator",
|
|
"role": "customer_operator",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
assert resp.status_code == 201
|
|
invite_data = resp.json()
|
|
assert invite_data["email"] == invitee_email
|
|
assert invite_data["role"] == "customer_operator"
|
|
assert invite_data["status"] == "pending"
|
|
token = invite_data["token"]
|
|
assert token is not None and len(token) > 0
|
|
|
|
# Step 2: Invitee accepts the invitation
|
|
accept_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": token, "password": "securepass123"},
|
|
)
|
|
assert accept_resp.status_code == 200
|
|
accept_data = accept_resp.json()
|
|
assert accept_data["email"] == invitee_email
|
|
assert accept_data["role"] == "customer_operator"
|
|
new_user_id = accept_data["id"]
|
|
|
|
# Step 3: Verify PortalUser was created correctly
|
|
user_result = await db_session.execute(
|
|
select(PortalUser).where(PortalUser.id == uuid.UUID(new_user_id))
|
|
)
|
|
new_user = user_result.scalar_one_or_none()
|
|
assert new_user is not None
|
|
assert new_user.email == invitee_email
|
|
assert new_user.role == "customer_operator"
|
|
|
|
# Step 4: Verify UserTenantRole was created
|
|
membership_result = await db_session.execute(
|
|
select(UserTenantRole).where(
|
|
UserTenantRole.user_id == uuid.UUID(new_user_id),
|
|
UserTenantRole.tenant_id == tenant.id,
|
|
)
|
|
)
|
|
membership = membership_result.scalar_one_or_none()
|
|
assert membership is not None
|
|
assert membership.role == "customer_operator"
|
|
|
|
# Step 5: Verify invitation status is 'accepted'
|
|
inv_result = await db_session.execute(
|
|
select(PortalInvitation).where(
|
|
PortalInvitation.id == uuid.UUID(invite_data["id"])
|
|
)
|
|
)
|
|
invitation = inv_result.scalar_one_or_none()
|
|
assert invitation is not None
|
|
assert invitation.status == "accepted"
|
|
|
|
# Step 6: Verify login works with correct role
|
|
login_resp = await invite_client.post(
|
|
"/api/portal/auth/verify",
|
|
json={"email": invitee_email, "password": "securepass123"},
|
|
)
|
|
assert login_resp.status_code == 200
|
|
login_data = login_resp.json()
|
|
assert login_data["role"] == "customer_operator"
|
|
assert str(tenant.id) in login_data["tenant_ids"]
|
|
|
|
async def test_full_invite_flow_customer_admin(
|
|
self, invite_client: AsyncClient, invite_setup: dict
|
|
) -> None:
|
|
"""Full flow for a customer_admin invitee."""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"new-admin-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Create invitation
|
|
resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "New Admin",
|
|
"role": "customer_admin",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
assert resp.status_code == 201
|
|
token = resp.json()["token"]
|
|
|
|
# Accept invitation
|
|
accept_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": token, "password": "adminpass123"},
|
|
)
|
|
assert accept_resp.status_code == 200
|
|
assert accept_resp.json()["role"] == "customer_admin"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Expired invitation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestExpiredInvite:
|
|
async def test_expired_invite_acceptance_returns_error(
|
|
self, invite_client: AsyncClient, db_session: AsyncSession, invite_setup: dict
|
|
) -> None:
|
|
"""Accepting an expired invitation returns a 400 error."""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"expired-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Create invitation normally
|
|
resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "Expired User",
|
|
"role": "customer_operator",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
assert resp.status_code == 201
|
|
token = resp.json()["token"]
|
|
invite_id = uuid.UUID(resp.json()["id"])
|
|
|
|
# Manually set expires_at to the past
|
|
inv_result = await db_session.execute(
|
|
select(PortalInvitation).where(PortalInvitation.id == invite_id)
|
|
)
|
|
invitation = inv_result.scalar_one()
|
|
invitation.expires_at = datetime.now(tz=timezone.utc) - timedelta(hours=1)
|
|
await db_session.commit()
|
|
|
|
# Attempt to accept — should fail with 400
|
|
accept_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": token, "password": "somepass123"},
|
|
)
|
|
assert accept_resp.status_code == 400
|
|
assert "expired" in accept_resp.json()["detail"].lower()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Resend invitation
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestResendInvite:
|
|
async def test_resend_generates_new_token_and_extends_expiry(
|
|
self, invite_client: AsyncClient, db_session: AsyncSession, invite_setup: dict
|
|
) -> None:
|
|
"""Resending an invitation generates a new token and extends expiry."""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"resend-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Create initial invitation
|
|
create_resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "Resend User",
|
|
"role": "customer_operator",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
assert create_resp.status_code == 201
|
|
invite_id = create_resp.json()["id"]
|
|
original_token = create_resp.json()["token"]
|
|
|
|
# Record original expiry
|
|
inv_result = await db_session.execute(
|
|
select(PortalInvitation).where(PortalInvitation.id == uuid.UUID(invite_id))
|
|
)
|
|
original_invitation = inv_result.scalar_one()
|
|
original_expiry = original_invitation.expires_at
|
|
|
|
# Resend the invitation
|
|
resend_resp = await invite_client.post(
|
|
f"/api/portal/invitations/{invite_id}/resend",
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
assert resend_resp.status_code == 200
|
|
new_token = resend_resp.json()["token"]
|
|
|
|
# Token should be different
|
|
assert new_token != original_token
|
|
assert new_token is not None and len(new_token) > 0
|
|
|
|
# Verify new token hash in DB and extended expiry
|
|
await db_session.refresh(original_invitation)
|
|
assert original_invitation.token_hash == token_to_hash(new_token)
|
|
|
|
# Expiry should be extended (refreshed from "now + 48h")
|
|
new_expiry = original_invitation.expires_at
|
|
assert new_expiry > original_expiry or (
|
|
# Allow if original was already far in future (within 1 second tolerance)
|
|
abs((new_expiry - original_expiry).total_seconds()) < 5
|
|
)
|
|
|
|
async def test_resend_new_token_can_be_used_to_accept(
|
|
self, invite_client: AsyncClient, invite_setup: dict
|
|
) -> None:
|
|
"""New token from resend works for acceptance."""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"resend-accept-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Create invitation
|
|
create_resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "Resend Accept User",
|
|
"role": "customer_operator",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
invite_id = create_resp.json()["id"]
|
|
|
|
# Resend to get new token
|
|
resend_resp = await invite_client.post(
|
|
f"/api/portal/invitations/{invite_id}/resend",
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
new_token = resend_resp.json()["token"]
|
|
|
|
# Accept with new token
|
|
accept_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": new_token, "password": "securepass123"},
|
|
)
|
|
assert accept_resp.status_code == 200
|
|
assert accept_resp.json()["email"] == invitee_email
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: Double-accept prevention
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
class TestDoubleAcceptPrevention:
|
|
async def test_double_accept_returns_conflict(
|
|
self, invite_client: AsyncClient, invite_setup: dict
|
|
) -> None:
|
|
"""Attempting to accept an already-accepted invitation returns 409."""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"double-accept-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Create invitation
|
|
create_resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "Double Accept User",
|
|
"role": "customer_operator",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
token = create_resp.json()["token"]
|
|
|
|
# First accept — should succeed
|
|
first_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": token, "password": "firstpass123"},
|
|
)
|
|
assert first_resp.status_code == 200
|
|
|
|
# Second accept with same token — should fail
|
|
second_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": token, "password": "secondpass123"},
|
|
)
|
|
# Either 409 (already accepted) or 400 (email already registered)
|
|
assert second_resp.status_code in (400, 409)
|
|
|
|
async def test_only_pending_invites_can_be_accepted(
|
|
self, invite_client: AsyncClient, db_session: AsyncSession, invite_setup: dict
|
|
) -> None:
|
|
"""Invitations with status != 'pending' cannot be accepted."""
|
|
tenant = invite_setup["tenant"]
|
|
admin = invite_setup["admin"]
|
|
invitee_email = f"non-pending-{uuid.uuid4().hex[:8]}@example.com"
|
|
|
|
# Create invitation
|
|
create_resp = await invite_client.post(
|
|
"/api/portal/invitations",
|
|
json={
|
|
"email": invitee_email,
|
|
"name": "Non Pending User",
|
|
"role": "customer_operator",
|
|
"tenant_id": str(tenant.id),
|
|
},
|
|
headers=admin_headers(admin.id, tenant.id),
|
|
)
|
|
invite_id = uuid.UUID(create_resp.json()["id"])
|
|
token = create_resp.json()["token"]
|
|
|
|
# Manually set status to 'revoked'
|
|
inv_result = await db_session.execute(
|
|
select(PortalInvitation).where(PortalInvitation.id == invite_id)
|
|
)
|
|
invitation = inv_result.scalar_one()
|
|
invitation.status = "revoked"
|
|
await db_session.commit()
|
|
|
|
# Attempt to accept revoked invitation
|
|
accept_resp = await invite_client.post(
|
|
"/api/portal/invitations/accept",
|
|
json={"token": token, "password": "somepass123"},
|
|
)
|
|
assert accept_resp.status_code == 409
|