Adolfo Delorenzo
f9ce3d650f
- Create shared/api/templates.py with templates_router
- GET /api/portal/templates: list active templates (any authenticated user)
- GET /api/portal/templates/{id}: get template detail (any authenticated user)
- POST /api/portal/templates/{id}/deploy: create Agent snapshot (tenant_admin only)
- customer_operator returns 403 on deploy (RBAC enforced)
- Export templates_router from shared/api/__init__.py
- Mount templates_router in gateway/main.py (Phase 5 section)
- 11 integration tests pass (list, detail, deploy, RBAC, 404, snapshot independence)
450 lines
14 KiB
Python
450 lines
14 KiB
Python
"""
|
|
Integration tests for agent template API endpoints.
|
|
|
|
Covers:
|
|
- GET /api/portal/templates returns 7+ seeded templates
|
|
- GET /api/portal/templates/{id} returns correct template detail
|
|
- POST /api/portal/templates/{id}/deploy creates agent snapshot with is_active=True
|
|
- POST deploy as customer_operator returns 403
|
|
- POST deploy with invalid UUID returns 404
|
|
- Deployed agent fields match template (snapshot verification)
|
|
|
|
Test infrastructure follows the same pattern as test_portal_rbac.py:
|
|
- Session override via app.dependency_overrides
|
|
- X-Portal-User-Id / X-Portal-User-Role / X-Portal-Tenant-Id header injection
|
|
- db_session fixture from tests/conftest.py (Alembic migrations applied)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Any
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from fastapi import FastAPI
|
|
from httpx import ASGITransport, AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from shared.api.portal import portal_router
|
|
from shared.api.templates import templates_router
|
|
from shared.db import get_session
|
|
from shared.models.auth import PortalUser, UserTenantRole
|
|
from shared.models.tenant import Tenant
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# App factory
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def make_app(session: AsyncSession) -> FastAPI:
|
|
"""Build a minimal FastAPI test app with portal + templates routers."""
|
|
app = FastAPI()
|
|
app.include_router(portal_router)
|
|
app.include_router(templates_router)
|
|
|
|
async def override_get_session(): # type: ignore[return]
|
|
yield session
|
|
|
|
app.dependency_overrides[get_session] = override_get_session
|
|
return app
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RBAC header helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
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",
|
|
}
|
|
|
|
|
|
def customer_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 customer_operator_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_operator",
|
|
"X-Portal-Tenant-Id": str(tenant_id),
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# DB setup helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def _create_tenant(session: AsyncSession, name: str | None = None) -> Tenant:
|
|
suffix = uuid.uuid4().hex[:8]
|
|
tenant = Tenant(
|
|
id=uuid.uuid4(),
|
|
name=name or f"Template Test Tenant {suffix}",
|
|
slug=f"tmpl-test-{suffix}",
|
|
settings={},
|
|
)
|
|
session.add(tenant)
|
|
await session.flush()
|
|
return tenant
|
|
|
|
|
|
async def _create_user(session: AsyncSession, role: str = "customer_admin") -> PortalUser:
|
|
import bcrypt
|
|
|
|
suffix = uuid.uuid4().hex[:8]
|
|
hashed = bcrypt.hashpw(b"testpassword123", bcrypt.gensalt()).decode()
|
|
user = PortalUser(
|
|
id=uuid.uuid4(),
|
|
email=f"tmpl-test-{suffix}@example.com",
|
|
hashed_password=hashed,
|
|
name=f"Template Test User {suffix}",
|
|
role=role,
|
|
)
|
|
session.add(user)
|
|
await session.flush()
|
|
return user
|
|
|
|
|
|
async def _grant_membership(
|
|
session: AsyncSession,
|
|
user: PortalUser,
|
|
tenant: Tenant,
|
|
role: str,
|
|
) -> UserTenantRole:
|
|
membership = UserTenantRole(
|
|
id=uuid.uuid4(),
|
|
user_id=user.id,
|
|
tenant_id=tenant.id,
|
|
role=role,
|
|
)
|
|
session.add(membership)
|
|
await session.flush()
|
|
return membership
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Fixtures
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def template_setup(db_session: AsyncSession) -> dict[str, Any]:
|
|
"""
|
|
Set up tenant, users, and memberships for template tests.
|
|
|
|
Returns:
|
|
- tenant: the primary test tenant
|
|
- platform_admin: platform_admin user
|
|
- customer_admin: customer_admin user with membership in tenant
|
|
- operator: customer_operator user with membership in tenant
|
|
"""
|
|
tenant = await _create_tenant(db_session)
|
|
platform_admin = await _create_user(db_session, role="platform_admin")
|
|
customer_admin = await _create_user(db_session, role="customer_admin")
|
|
operator = await _create_user(db_session, role="customer_operator")
|
|
|
|
await _grant_membership(db_session, customer_admin, tenant, "customer_admin")
|
|
await _grant_membership(db_session, operator, tenant, "customer_operator")
|
|
|
|
await db_session.commit()
|
|
|
|
return {
|
|
"tenant": tenant,
|
|
"platform_admin": platform_admin,
|
|
"customer_admin": customer_admin,
|
|
"operator": operator,
|
|
}
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def templates_client(db_session: AsyncSession) -> AsyncClient:
|
|
"""HTTP client with templates router mounted."""
|
|
app = make_app(db_session)
|
|
async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c:
|
|
yield c
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: GET /api/portal/templates
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_templates(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""GET /api/portal/templates returns 200 with 7+ active seeded templates."""
|
|
admin = template_setup["platform_admin"]
|
|
headers = platform_admin_headers(admin.id)
|
|
|
|
response = await templates_client.get("/api/portal/templates", headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 7, f"Expected at least 7 seeded templates, got {len(data)}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_templates_returns_expected_fields(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""Template list items include all required response fields."""
|
|
admin = template_setup["platform_admin"]
|
|
headers = platform_admin_headers(admin.id)
|
|
|
|
response = await templates_client.get("/api/portal/templates", headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
templates = response.json()
|
|
assert len(templates) > 0
|
|
|
|
# Check first template has all required fields
|
|
first = templates[0]
|
|
for field in ("id", "name", "role", "description", "category", "persona",
|
|
"system_prompt", "model_preference", "tool_assignments",
|
|
"escalation_rules", "is_active", "sort_order", "created_at"):
|
|
assert field in first, f"Missing field: {field}"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_templates_customer_admin_can_browse(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""customer_admin can browse templates (any authenticated user can)."""
|
|
admin = template_setup["customer_admin"]
|
|
tenant = template_setup["tenant"]
|
|
headers = customer_admin_headers(admin.id, tenant.id)
|
|
|
|
response = await templates_client.get("/api/portal/templates", headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
assert len(response.json()) >= 7
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_list_templates_operator_can_browse(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""customer_operator can also browse templates."""
|
|
operator = template_setup["operator"]
|
|
tenant = template_setup["tenant"]
|
|
headers = customer_operator_headers(operator.id, tenant.id)
|
|
|
|
response = await templates_client.get("/api/portal/templates", headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
assert len(response.json()) >= 7
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: GET /api/portal/templates/{id}
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_template_detail(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""GET single template returns correct fields including AI transparency clause."""
|
|
admin = template_setup["platform_admin"]
|
|
headers = platform_admin_headers(admin.id)
|
|
|
|
# First get the list to find a real template ID
|
|
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
|
assert list_resp.status_code == 200
|
|
templates = list_resp.json()
|
|
assert len(templates) > 0
|
|
|
|
template_id = templates[0]["id"]
|
|
response = await templates_client.get(f"/api/portal/templates/{template_id}", headers=headers)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == template_id
|
|
assert data["name"] == templates[0]["name"]
|
|
assert data["role"] == templates[0]["role"]
|
|
assert data["is_active"] is True
|
|
# System prompt should contain AI transparency clause
|
|
assert "When directly asked if you are an AI" in data["system_prompt"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_template_not_found(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""GET with non-existent UUID returns 404."""
|
|
admin = template_setup["platform_admin"]
|
|
headers = platform_admin_headers(admin.id)
|
|
nonexistent_id = str(uuid.uuid4())
|
|
|
|
response = await templates_client.get(f"/api/portal/templates/{nonexistent_id}", headers=headers)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Tests: POST /api/portal/templates/{id}/deploy
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deploy_template(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
db_session: AsyncSession,
|
|
) -> None:
|
|
"""POST deploy creates an independent Agent snapshot with is_active=True."""
|
|
admin = template_setup["customer_admin"]
|
|
tenant = template_setup["tenant"]
|
|
headers = customer_admin_headers(admin.id, tenant.id)
|
|
|
|
# Get a template to deploy
|
|
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
|
assert list_resp.status_code == 200
|
|
templates = list_resp.json()
|
|
template = templates[0]
|
|
template_id = template["id"]
|
|
|
|
response = await templates_client.post(
|
|
f"/api/portal/templates/{template_id}/deploy",
|
|
json={"tenant_id": str(tenant.id)},
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
agent = data["agent"]
|
|
|
|
# Agent snapshot should be active
|
|
assert agent["is_active"] is True
|
|
# Agent fields should match template
|
|
assert agent["name"] == template["name"]
|
|
assert agent["role"] == template["role"]
|
|
assert agent["persona"] == template["persona"]
|
|
assert agent["system_prompt"] == template["system_prompt"]
|
|
assert agent["model_preference"] == template["model_preference"]
|
|
assert agent["tool_assignments"] == template["tool_assignments"]
|
|
# Agent should be scoped to the tenant
|
|
assert agent["tenant_id"] == str(tenant.id)
|
|
# Agent should have an id
|
|
assert agent["id"] is not None
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deploy_template_platform_admin(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""platform_admin can deploy templates to any tenant."""
|
|
admin = template_setup["platform_admin"]
|
|
tenant = template_setup["tenant"]
|
|
headers = platform_admin_headers(admin.id)
|
|
|
|
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
|
template = list_resp.json()[0]
|
|
template_id = template["id"]
|
|
|
|
response = await templates_client.post(
|
|
f"/api/portal/templates/{template_id}/deploy",
|
|
json={"tenant_id": str(tenant.id)},
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["agent"]["is_active"] is True
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deploy_template_rbac_operator_forbidden(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""POST deploy as customer_operator returns 403."""
|
|
operator = template_setup["operator"]
|
|
tenant = template_setup["tenant"]
|
|
headers = customer_operator_headers(operator.id, tenant.id)
|
|
|
|
# Get a template ID
|
|
admin = template_setup["platform_admin"]
|
|
list_resp = await templates_client.get(
|
|
"/api/portal/templates",
|
|
headers=platform_admin_headers(admin.id),
|
|
)
|
|
template_id = list_resp.json()[0]["id"]
|
|
|
|
response = await templates_client.post(
|
|
f"/api/portal/templates/{template_id}/deploy",
|
|
json={"tenant_id": str(tenant.id)},
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deploy_template_not_found(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""POST deploy with invalid/nonexistent template UUID returns 404."""
|
|
admin = template_setup["customer_admin"]
|
|
tenant = template_setup["tenant"]
|
|
headers = customer_admin_headers(admin.id, tenant.id)
|
|
nonexistent_id = str(uuid.uuid4())
|
|
|
|
response = await templates_client.post(
|
|
f"/api/portal/templates/{nonexistent_id}/deploy",
|
|
json={"tenant_id": str(tenant.id)},
|
|
headers=headers,
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_deploy_creates_independent_snapshot(
|
|
templates_client: AsyncClient,
|
|
template_setup: dict[str, Any],
|
|
) -> None:
|
|
"""Deploying the same template twice creates two independent agents."""
|
|
admin = template_setup["customer_admin"]
|
|
tenant = template_setup["tenant"]
|
|
headers = customer_admin_headers(admin.id, tenant.id)
|
|
|
|
list_resp = await templates_client.get("/api/portal/templates", headers=headers)
|
|
template_id = list_resp.json()[0]["id"]
|
|
|
|
resp1 = await templates_client.post(
|
|
f"/api/portal/templates/{template_id}/deploy",
|
|
json={"tenant_id": str(tenant.id)},
|
|
headers=headers,
|
|
)
|
|
resp2 = await templates_client.post(
|
|
f"/api/portal/templates/{template_id}/deploy",
|
|
json={"tenant_id": str(tenant.id)},
|
|
headers=headers,
|
|
)
|
|
|
|
assert resp1.status_code == 201
|
|
assert resp2.status_code == 201
|
|
# Both agents should have distinct IDs
|
|
agent1_id = resp1.json()["agent"]["id"]
|
|
agent2_id = resp2.json()["agent"]["id"]
|
|
assert agent1_id != agent2_id
|